relay_knowledge/application/
runtime.rs

1use std::{
2    error::Error,
3    fmt,
4    path::{Component, PathBuf},
5    time::Duration,
6};
7
8use crate::{
9    api::{AgentAccessPolicy, AgentPolicyError},
10    domain::{RerankMode, RerankModeError, WorkerKind},
11    env::{
12        EnvError, EnvironmentConfig, PlatformKind, RELAY_KNOWLEDGE_EMBEDDING_API_KEY,
13        RELAY_KNOWLEDGE_EMBEDDING_BASE_URL, RELAY_KNOWLEDGE_EMBEDDING_DIMENSION,
14        RELAY_KNOWLEDGE_IMAGE_EMBEDDING_MODEL, RELAY_KNOWLEDGE_RERANK_MODEL,
15        RELAY_KNOWLEDGE_TEXT_EMBEDDING_MODEL, RetrievalEnvOverrides,
16    },
17    net::{NetworkConfig, NetworkConfigError, NetworkRuntime, NetworkRuntimeError},
18    observability::{ObservabilityRuntime, TelemetryConfig},
19    paths::{PathError, RuntimePaths, default_user_document_roots},
20    retrieval::{
21        DEFAULT_EMBEDDING_BATCH_SIZE, DEFAULT_EMBEDDING_MAX_CONCURRENCY, DEFAULT_EMBEDDING_TIMEOUT,
22        DEFAULT_RERANK_CANDIDATE_MULTIPLIER, DEFAULT_RERANK_MAX_CANDIDATES, DEFAULT_RERANK_TIMEOUT,
23        EmbeddingProviderKind, EmbeddingProviderKindError, LOCAL_RERANK_MODEL,
24        LOCAL_SEMANTIC_MODEL, LOCAL_VECTOR_DIMENSION, LOCAL_VECTOR_MODEL, ReadModelBackendConfig,
25        ReadModelBackendMode, ReadModelBackendModeError, ReadModelMetadata, RemoteEmbeddingConfig,
26        RerankConfig,
27    },
28};
29
30/// Resolved foundation configuration shared by all interfaces.
31#[derive(Debug, Clone)]
32pub struct RuntimeConfiguration {
33    pub paths: RuntimePaths,
34    pub network: NetworkRuntime,
35    pub observability: ObservabilityRuntime,
36    pub agent: AgentRuntimeConfig,
37    pub retrieval: ReadModelBackendConfig,
38    pub workers: WorkerRuntimeConfig,
39    pub file_index: FileIndexRuntimeConfig,
40}
41
42impl RuntimeConfiguration {
43    /// Resolves runtime configuration from the current process environment.
44    pub async fn from_process_environment() -> Result<Self, RuntimeConfigurationError> {
45        let environment =
46            EnvironmentConfig::from_process().map_err(RuntimeConfigurationError::Environment)?;
47
48        Self::from_environment(&environment).await
49    }
50
51    /// Resolves runtime configuration from a typed environment snapshot.
52    pub async fn from_environment(
53        environment: &EnvironmentConfig,
54    ) -> Result<Self, RuntimeConfigurationError> {
55        let network = NetworkConfig::from_overrides(&environment.network)
56            .map_err(RuntimeConfigurationError::Network)?;
57        let observability =
58            ObservabilityRuntime::new(TelemetryConfig::from_environment(&environment.telemetry));
59        let agent = AgentRuntimeConfig::from_environment(environment, network.http.request_timeout)
60            .map_err(RuntimeConfigurationError::Agent)?;
61        let retrieval = retrieval_config_from_environment(&environment.retrieval)
62            .map_err(RuntimeConfigurationError::Retrieval)?;
63        let workers = WorkerRuntimeConfig::from_environment(environment)
64            .map_err(RuntimeConfigurationError::Workers)?;
65        let file_index = FileIndexRuntimeConfig::from_environment(environment)
66            .map_err(RuntimeConfigurationError::FileIndex)?;
67
68        Ok(Self {
69            paths: RuntimePaths::resolve(&environment.platform, &environment.paths)
70                .map_err(RuntimeConfigurationError::Paths)?,
71            network: NetworkRuntime::from_config(network),
72            observability,
73            agent,
74            retrieval,
75            workers,
76            file_index,
77        })
78    }
79}
80
81/// Runtime budgets and authorized roots for local file-location indexing.
82#[derive(Debug, Clone, PartialEq, Eq)]
83pub struct FileIndexRuntimeConfig {
84    pub enabled: bool,
85    pub roots: Vec<FileIndexRootConfig>,
86    pub excludes: Vec<String>,
87    pub max_depth: usize,
88    pub max_file_bytes: u64,
89    pub scan_interval: Duration,
90    pub scan_timeout: Duration,
91    pub max_files_per_root: usize,
92    pub query_timeout: Duration,
93}
94
95impl FileIndexRuntimeConfig {
96    pub const DEFAULT_MAX_DEPTH: usize = 32;
97    pub const DEFAULT_MAX_FILE_BYTES: u64 = 512 * 1024 * 1024;
98    pub const DEFAULT_SCAN_INTERVAL: Duration = Duration::from_secs(900);
99    pub const DEFAULT_SCAN_TIMEOUT: Duration = Duration::from_secs(300);
100    pub const DEFAULT_MAX_FILES_PER_ROOT: usize = 50_000;
101    pub const DEFAULT_QUERY_TIMEOUT: Duration = Duration::from_millis(750);
102
103    pub fn from_environment(
104        environment: &EnvironmentConfig,
105    ) -> Result<Self, FileIndexRuntimeConfigError> {
106        let mut roots = default_user_document_roots(&environment.platform)
107            .map_err(FileIndexRuntimeConfigError::Paths)?
108            .into_iter()
109            .map(|path| FileIndexRootConfig::new("user-documents", path))
110            .collect::<Vec<_>>();
111        for root in split_semicolon(environment.file_index.roots.as_deref())? {
112            roots.push(file_index_root_from_environment(
113                "local-files",
114                root,
115                environment.platform.platform,
116            )?);
117        }
118        roots.sort_by(|left, right| {
119            left.scope_id
120                .cmp(&right.scope_id)
121                .then(left.root_id.cmp(&right.root_id))
122        });
123        roots.dedup_by(|left, right| {
124            left.scope_id == right.scope_id && left.root_id == right.root_id
125        });
126
127        Ok(Self {
128            enabled: environment.file_index.enabled.unwrap_or(false),
129            roots,
130            excludes: split_semicolon(environment.file_index.excludes.as_deref())?,
131            max_depth: environment
132                .file_index
133                .max_depth
134                .unwrap_or(Self::DEFAULT_MAX_DEPTH),
135            max_file_bytes: environment
136                .file_index
137                .max_file_bytes
138                .unwrap_or(Self::DEFAULT_MAX_FILE_BYTES),
139            scan_interval: Duration::from_millis(
140                environment
141                    .file_index
142                    .scan_interval_ms
143                    .unwrap_or(duration_millis(Self::DEFAULT_SCAN_INTERVAL)),
144            ),
145            scan_timeout: Duration::from_millis(
146                environment
147                    .file_index
148                    .scan_timeout_ms
149                    .unwrap_or(duration_millis(Self::DEFAULT_SCAN_TIMEOUT)),
150            ),
151            max_files_per_root: environment
152                .file_index
153                .max_files_per_root
154                .unwrap_or(Self::DEFAULT_MAX_FILES_PER_ROOT),
155            query_timeout: Duration::from_millis(
156                environment
157                    .file_index
158                    .query_timeout_ms
159                    .unwrap_or(duration_millis(Self::DEFAULT_QUERY_TIMEOUT)),
160            ),
161        })
162    }
163}
164
165/// One authorized local file index root.
166#[derive(Debug, Clone, PartialEq, Eq)]
167pub struct FileIndexRootConfig {
168    pub scope_id: String,
169    pub root_id: String,
170    pub root_path: PathBuf,
171}
172
173impl FileIndexRootConfig {
174    pub fn new(scope_id: impl Into<String>, root_path: PathBuf) -> Self {
175        let root_path = normalize_file_index_root_path(root_path);
176        let root_id = format!(
177            "root-{:016x}",
178            stable_hash64(root_path.to_string_lossy().as_bytes())
179        );
180
181        Self {
182            scope_id: scope_id.into(),
183            root_id,
184            root_path,
185        }
186    }
187}
188
189fn normalize_file_index_root_path(root_path: PathBuf) -> PathBuf {
190    if let Ok(canonical) = std::fs::canonicalize(&root_path) {
191        return canonical;
192    }
193
194    let mut normalized = PathBuf::new();
195    for component in root_path.components() {
196        match component {
197            Component::CurDir => {}
198            Component::ParentDir => normalized.push(".."),
199            Component::Prefix(prefix) => normalized.push(prefix.as_os_str()),
200            Component::RootDir => normalized.push(component.as_os_str()),
201            Component::Normal(value) => normalized.push(value),
202        }
203    }
204
205    if normalized.as_os_str().is_empty() {
206        root_path
207    } else {
208        normalized
209    }
210}
211
212/// File index runtime validation error.
213#[derive(Debug, Clone, PartialEq, Eq)]
214pub enum FileIndexRuntimeConfigError {
215    EmptyListValue,
216    RelativeRoot(String),
217    Paths(PathError),
218}
219
220impl fmt::Display for FileIndexRuntimeConfigError {
221    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
222        match self {
223            Self::EmptyListValue => {
224                write!(formatter, "file index lists must not contain empty values")
225            }
226            Self::RelativeRoot(path) => {
227                write!(
228                    formatter,
229                    "file index root '{path}' must be an absolute path"
230                )
231            }
232            Self::Paths(error) => write!(formatter, "{error}"),
233        }
234    }
235}
236
237impl Error for FileIndexRuntimeConfigError {}
238
239/// External worker runtime configuration and deterministic fallback policy.
240#[derive(Debug, Clone, PartialEq, Eq)]
241pub struct WorkerRuntimeConfig {
242    pub embedding_endpoint: Option<String>,
243    pub ocr_endpoint: Option<String>,
244    pub vision_endpoint: Option<String>,
245    pub extractor_endpoint: Option<String>,
246    pub max_in_flight: usize,
247    pub silent_updates_enabled: bool,
248}
249
250impl WorkerRuntimeConfig {
251    pub const DEFAULT_MAX_IN_FLIGHT: usize = 2;
252
253    /// Builds worker config from typed environment overrides.
254    pub fn from_environment(
255        environment: &EnvironmentConfig,
256    ) -> Result<Self, WorkerRuntimeConfigError> {
257        Ok(Self {
258            embedding_endpoint: validate_worker_endpoint(
259                environment.workers.embedding_endpoint.clone(),
260            )?,
261            ocr_endpoint: validate_worker_endpoint(environment.workers.ocr_endpoint.clone())?,
262            vision_endpoint: validate_worker_endpoint(environment.workers.vision_endpoint.clone())?,
263            extractor_endpoint: validate_worker_endpoint(
264                environment.workers.extractor_endpoint.clone(),
265            )?,
266            max_in_flight: environment
267                .workers
268                .max_in_flight
269                .unwrap_or(Self::DEFAULT_MAX_IN_FLIGHT),
270            silent_updates_enabled: environment.workers.silent_updates_enabled.unwrap_or(false),
271        })
272    }
273
274    /// Returns the configured endpoint for a worker kind.
275    pub fn endpoint_for(&self, kind: WorkerKind) -> Option<&str> {
276        match kind {
277            WorkerKind::Embedding => self.embedding_endpoint.as_deref(),
278            WorkerKind::Ocr => self.ocr_endpoint.as_deref(),
279            WorkerKind::Vision => self.vision_endpoint.as_deref(),
280            WorkerKind::Extractor => self.extractor_endpoint.as_deref(),
281        }
282    }
283}
284
285/// Worker runtime configuration validation error.
286#[derive(Debug, Clone, PartialEq, Eq)]
287pub enum WorkerRuntimeConfigError {
288    InvalidEndpoint(String),
289}
290
291impl fmt::Display for WorkerRuntimeConfigError {
292    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
293        match self {
294            Self::InvalidEndpoint(value) => write!(
295                formatter,
296                "worker endpoint '{value}' must use http:// and include a host"
297            ),
298        }
299    }
300}
301
302impl Error for WorkerRuntimeConfigError {}
303
304/// Resident agent protocol runtime configuration.
305#[derive(Debug, Clone, PartialEq, Eq)]
306pub struct AgentRuntimeConfig {
307    pub mcp_streamable_http_enabled: bool,
308    pub mcp_endpoint: String,
309    pub mcp_allowed_origins: Vec<String>,
310    pub access_policy: AgentAccessPolicy,
311    pub audit_sink_enabled: bool,
312    pub audit_queue_depth: usize,
313}
314
315impl AgentRuntimeConfig {
316    pub const DEFAULT_AUDIT_QUEUE_DEPTH: usize = 1024;
317
318    /// Builds agent protocol config from typed environment overrides.
319    pub fn from_environment(
320        environment: &EnvironmentConfig,
321        request_timeout: Duration,
322    ) -> Result<Self, AgentRuntimeConfigError> {
323        let max_runtime_ms = agent_runtime_budget_ms(request_timeout);
324        let access_policy = AgentAccessPolicy::new(
325            split_csv(environment.agent.mcp_allowed_scopes.as_deref())?,
326            environment
327                .agent
328                .mcp_allow_unspecified_scope
329                .unwrap_or(false),
330            environment
331                .agent
332                .mcp_max_limit
333                .unwrap_or(AgentAccessPolicy::DEFAULT_MAX_LIMIT),
334            environment
335                .agent
336                .mcp_max_context_bytes
337                .unwrap_or(AgentAccessPolicy::DEFAULT_MAX_CONTEXT_BYTES),
338            max_runtime_ms,
339            environment.agent.mcp_allow_remote_clients.unwrap_or(false),
340        )
341        .map_err(AgentRuntimeConfigError::Policy)?;
342
343        Ok(Self {
344            mcp_streamable_http_enabled: environment
345                .agent
346                .mcp_streamable_http_enabled
347                .unwrap_or(false),
348            mcp_endpoint: validate_endpoint(
349                environment.agent.mcp_endpoint.as_deref().unwrap_or("/mcp"),
350            )?,
351            mcp_allowed_origins: split_csv(environment.agent.mcp_allowed_origins.as_deref())?,
352            access_policy,
353            audit_sink_enabled: environment.agent.audit_sink_enabled.unwrap_or(false),
354            audit_queue_depth: environment
355                .agent
356                .audit_queue_depth
357                .unwrap_or(Self::DEFAULT_AUDIT_QUEUE_DEPTH),
358        })
359    }
360
361    /// Returns a copy with streamable HTTP forced on by a CLI command.
362    pub fn with_streamable_http_enabled(mut self) -> Self {
363        self.mcp_streamable_http_enabled = true;
364        self
365    }
366}
367
368/// Agent runtime configuration validation error.
369#[derive(Debug, Clone, PartialEq, Eq)]
370pub enum AgentRuntimeConfigError {
371    InvalidEndpoint(String),
372    EmptyListValue,
373    Policy(AgentPolicyError),
374}
375
376impl fmt::Display for AgentRuntimeConfigError {
377    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
378        match self {
379            Self::InvalidEndpoint(value) => {
380                write!(
381                    formatter,
382                    "MCP endpoint '{value}' must be an absolute HTTP path"
383                )
384            }
385            Self::EmptyListValue => {
386                write!(formatter, "MCP comma-separated values must not be empty")
387            }
388            Self::Policy(error) => write!(formatter, "{error}"),
389        }
390    }
391}
392
393impl Error for AgentRuntimeConfigError {}
394
395/// Error raised while composing foundational runtime configuration.
396#[derive(Debug, Clone, PartialEq, Eq)]
397pub enum RuntimeConfigurationError {
398    Environment(EnvError),
399    Paths(PathError),
400    Network(NetworkConfigError),
401    NetworkRuntime(NetworkRuntimeError),
402    Agent(AgentRuntimeConfigError),
403    Retrieval(RetrievalRuntimeConfigError),
404    Workers(WorkerRuntimeConfigError),
405    FileIndex(FileIndexRuntimeConfigError),
406}
407
408impl fmt::Display for RuntimeConfigurationError {
409    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
410        match self {
411            Self::Environment(error) => write!(formatter, "{error}"),
412            Self::Paths(error) => write!(formatter, "{error}"),
413            Self::Network(error) => write!(formatter, "{error}"),
414            Self::NetworkRuntime(error) => write!(formatter, "{error}"),
415            Self::Agent(error) => write!(formatter, "{error}"),
416            Self::Retrieval(error) => write!(formatter, "{error}"),
417            Self::Workers(error) => write!(formatter, "{error}"),
418            Self::FileIndex(error) => write!(formatter, "{error}"),
419        }
420    }
421}
422
423impl Error for RuntimeConfigurationError {}
424
425/// Retrieval runtime configuration validation error.
426#[derive(Debug, Clone, PartialEq, Eq)]
427pub enum RetrievalRuntimeConfigError {
428    InvalidBackend(ReadModelBackendModeError),
429    InvalidRerankBackend(RerankModeError),
430    InvalidProvider(EmbeddingProviderKindError),
431    EmptyModelName(&'static str),
432    MissingRemoteValue(&'static str),
433    InvalidRemoteBaseUrl(String),
434    DimensionTooLarge(usize),
435}
436
437impl fmt::Display for RetrievalRuntimeConfigError {
438    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
439        match self {
440            Self::InvalidBackend(error) => write!(formatter, "{error}"),
441            Self::InvalidRerankBackend(error) => write!(formatter, "{error}"),
442            Self::InvalidProvider(error) => write!(formatter, "{error}"),
443            Self::EmptyModelName(variable) => {
444                write!(formatter, "{variable} must not be blank")
445            }
446            Self::MissingRemoteValue(variable) => {
447                write!(
448                    formatter,
449                    "{variable} is required when a read model backend is external"
450                )
451            }
452            Self::InvalidRemoteBaseUrl(value) => {
453                write!(
454                    formatter,
455                    "embedding base URL '{value}' must use http:// or https://"
456                )
457            }
458            Self::DimensionTooLarge(value) => {
459                write!(formatter, "embedding dimension {value} does not fit in u32")
460            }
461        }
462    }
463}
464
465impl Error for RetrievalRuntimeConfigError {}
466
467fn retrieval_config_from_environment(
468    overrides: &RetrievalEnvOverrides,
469) -> Result<ReadModelBackendConfig, RetrievalRuntimeConfigError> {
470    let semantic_mode = parse_backend_mode(overrides.semantic_backend.as_deref())?;
471    let vector_mode = parse_backend_mode(overrides.vector_backend.as_deref())?;
472    let remote_required = semantic_mode == ReadModelBackendMode::External
473        || vector_mode == ReadModelBackendMode::External;
474    require_remote_model_metadata(overrides, remote_required)?;
475    let dimension = match overrides.embedding_dimension {
476        Some(value) => u32::try_from(value)
477            .map_err(|_| RetrievalRuntimeConfigError::DimensionTooLarge(value))?,
478        None => LOCAL_VECTOR_DIMENSION,
479    };
480    let text_model = model_name_override(
481        overrides.text_embedding_model.as_deref(),
482        RELAY_KNOWLEDGE_TEXT_EMBEDDING_MODEL,
483        LOCAL_VECTOR_MODEL,
484    )?;
485    let semantic_model = model_name_override(
486        overrides.text_embedding_model.as_deref(),
487        RELAY_KNOWLEDGE_TEXT_EMBEDDING_MODEL,
488        LOCAL_SEMANTIC_MODEL,
489    )?;
490    let image_model = model_name_override(
491        overrides.image_embedding_model.as_deref(),
492        RELAY_KNOWLEDGE_IMAGE_EMBEDDING_MODEL,
493        "relay-local-image-hash-v1",
494    )?;
495
496    let remote_embedding = remote_embedding_config_from_environment(overrides, remote_required)?;
497    let rerank = rerank_config_from_environment(overrides)?;
498
499    Ok(ReadModelBackendConfig {
500        semantic_mode,
501        vector_mode,
502        semantic_model: ReadModelMetadata {
503            name: semantic_model,
504            dimension,
505        },
506        vector_model: ReadModelMetadata {
507            name: text_model,
508            dimension,
509        },
510        image_model: ReadModelMetadata {
511            name: image_model,
512            dimension,
513        },
514        remote_embedding,
515        rerank,
516    })
517}
518
519fn rerank_config_from_environment(
520    overrides: &RetrievalEnvOverrides,
521) -> Result<RerankConfig, RetrievalRuntimeConfigError> {
522    let mode = overrides
523        .rerank_backend
524        .as_deref()
525        .map(RerankMode::parse)
526        .transpose()
527        .map_err(RetrievalRuntimeConfigError::InvalidRerankBackend)?
528        .unwrap_or(RerankMode::Local);
529    let model = match mode {
530        RerankMode::Disabled => None,
531        RerankMode::Local => Some(model_name_override(
532            overrides.rerank_model.as_deref(),
533            RELAY_KNOWLEDGE_RERANK_MODEL,
534            LOCAL_RERANK_MODEL,
535        )?),
536        RerankMode::External => overrides
537            .rerank_model
538            .as_deref()
539            .map(|model| model_name_override(Some(model), RELAY_KNOWLEDGE_RERANK_MODEL, ""))
540            .transpose()?,
541    };
542    let timeout = overrides
543        .rerank_timeout_ms
544        .map(Duration::from_millis)
545        .unwrap_or(DEFAULT_RERANK_TIMEOUT);
546
547    Ok(RerankConfig {
548        mode,
549        model,
550        timeout,
551        candidate_multiplier: overrides
552            .rerank_candidate_multiplier
553            .unwrap_or(DEFAULT_RERANK_CANDIDATE_MULTIPLIER),
554        max_candidates: overrides
555            .rerank_max_candidates
556            .unwrap_or(DEFAULT_RERANK_MAX_CANDIDATES),
557    })
558}
559
560fn require_remote_model_metadata(
561    overrides: &RetrievalEnvOverrides,
562    required: bool,
563) -> Result<(), RetrievalRuntimeConfigError> {
564    if !required {
565        return Ok(());
566    }
567    if overrides.text_embedding_model.is_none() {
568        return Err(RetrievalRuntimeConfigError::MissingRemoteValue(
569            RELAY_KNOWLEDGE_TEXT_EMBEDDING_MODEL,
570        ));
571    }
572    if overrides.embedding_dimension.is_none() {
573        return Err(RetrievalRuntimeConfigError::MissingRemoteValue(
574            RELAY_KNOWLEDGE_EMBEDDING_DIMENSION,
575        ));
576    }
577
578    Ok(())
579}
580
581fn remote_embedding_config_from_environment(
582    overrides: &RetrievalEnvOverrides,
583    required: bool,
584) -> Result<Option<RemoteEmbeddingConfig>, RetrievalRuntimeConfigError> {
585    if !required {
586        return Ok(None);
587    }
588    let provider = overrides
589        .llm_provider
590        .as_deref()
591        .map(EmbeddingProviderKind::parse)
592        .transpose()
593        .map_err(RetrievalRuntimeConfigError::InvalidProvider)?
594        .unwrap_or(EmbeddingProviderKind::OpenAiCompatible);
595    let base_url = required_remote_value(
596        overrides.embedding_base_url.as_deref(),
597        RELAY_KNOWLEDGE_EMBEDDING_BASE_URL,
598    )?;
599    if !base_url.starts_with("http://") && !base_url.starts_with("https://") {
600        return Err(RetrievalRuntimeConfigError::InvalidRemoteBaseUrl(base_url));
601    }
602    let api_key = required_remote_value(
603        overrides.embedding_api_key.as_deref(),
604        RELAY_KNOWLEDGE_EMBEDDING_API_KEY,
605    )?;
606    let batch_size = overrides
607        .embedding_batch_size
608        .unwrap_or(DEFAULT_EMBEDDING_BATCH_SIZE);
609    let timeout = overrides
610        .embedding_timeout_ms
611        .map(Duration::from_millis)
612        .unwrap_or(DEFAULT_EMBEDDING_TIMEOUT);
613    let max_concurrency = overrides
614        .embedding_max_concurrency
615        .unwrap_or(DEFAULT_EMBEDDING_MAX_CONCURRENCY);
616
617    Ok(Some(RemoteEmbeddingConfig {
618        provider,
619        base_url,
620        api_key,
621        batch_size,
622        timeout,
623        max_concurrency,
624    }))
625}
626
627fn required_remote_value(
628    value: Option<&str>,
629    variable: &'static str,
630) -> Result<String, RetrievalRuntimeConfigError> {
631    match value.map(str::trim) {
632        Some(trimmed) if !trimmed.is_empty() => Ok(trimmed.to_owned()),
633        _ => Err(RetrievalRuntimeConfigError::MissingRemoteValue(variable)),
634    }
635}
636
637fn model_name_override(
638    value: Option<&str>,
639    variable: &'static str,
640    default: &'static str,
641) -> Result<String, RetrievalRuntimeConfigError> {
642    match value {
643        Some(raw) => {
644            let trimmed = raw.trim();
645            if trimmed.is_empty() {
646                Err(RetrievalRuntimeConfigError::EmptyModelName(variable))
647            } else {
648                Ok(trimmed.to_owned())
649            }
650        }
651        None => Ok(default.to_owned()),
652    }
653}
654
655fn parse_backend_mode(
656    value: Option<&str>,
657) -> Result<ReadModelBackendMode, RetrievalRuntimeConfigError> {
658    value
659        .map(ReadModelBackendMode::parse)
660        .transpose()
661        .map_err(RetrievalRuntimeConfigError::InvalidBackend)
662        .map(|mode| mode.unwrap_or(ReadModelBackendMode::Local))
663}
664
665fn validate_endpoint(value: &str) -> Result<String, AgentRuntimeConfigError> {
666    let trimmed = value.trim();
667    if !trimmed.starts_with('/')
668        || trimmed.contains(char::is_whitespace)
669        || trimmed.contains('?')
670        || trimmed.contains('#')
671    {
672        return Err(AgentRuntimeConfigError::InvalidEndpoint(value.to_owned()));
673    }
674
675    Ok(trimmed.to_owned())
676}
677
678fn validate_worker_endpoint(
679    value: Option<String>,
680) -> Result<Option<String>, WorkerRuntimeConfigError> {
681    value
682        .map(|endpoint| {
683            let trimmed = endpoint.trim();
684            if is_valid_worker_http_endpoint(trimmed) {
685                Ok(trimmed.to_owned())
686            } else {
687                Err(WorkerRuntimeConfigError::InvalidEndpoint(endpoint))
688            }
689        })
690        .transpose()
691}
692
693fn is_valid_worker_http_endpoint(value: &str) -> bool {
694    let Some(remainder) = value.strip_prefix("http://") else {
695        return false;
696    };
697    let authority = remainder
698        .split_once('/')
699        .map_or(remainder, |(authority, _)| authority);
700    if authority.is_empty() || authority.contains(char::is_whitespace) {
701        return false;
702    }
703    if let Some((host, port)) = authority.rsplit_once(':') {
704        return !host.is_empty() && port.parse::<u16>().is_ok_and(|port| port > 0);
705    }
706
707    !authority.is_empty()
708}
709
710fn split_csv(value: Option<&str>) -> Result<Vec<String>, AgentRuntimeConfigError> {
711    value
712        .map(|items| {
713            items
714                .split(',')
715                .map(str::trim)
716                .map(|item| {
717                    if item.is_empty() {
718                        Err(AgentRuntimeConfigError::EmptyListValue)
719                    } else {
720                        Ok(item.to_owned())
721                    }
722                })
723                .collect()
724        })
725        .unwrap_or_else(|| Ok(Vec::new()))
726}
727
728fn split_semicolon(value: Option<&str>) -> Result<Vec<String>, FileIndexRuntimeConfigError> {
729    value
730        .map(|items| {
731            items
732                .split(';')
733                .map(str::trim)
734                .map(|item| {
735                    if item.is_empty() {
736                        Err(FileIndexRuntimeConfigError::EmptyListValue)
737                    } else {
738                        Ok(item.to_owned())
739                    }
740                })
741                .collect()
742        })
743        .unwrap_or_else(|| Ok(Vec::new()))
744}
745
746fn file_index_root_from_environment(
747    scope_id: &'static str,
748    root: String,
749    platform: PlatformKind,
750) -> Result<FileIndexRootConfig, FileIndexRuntimeConfigError> {
751    if !is_absolute_file_index_root(&root, platform) {
752        return Err(FileIndexRuntimeConfigError::RelativeRoot(root));
753    }
754
755    Ok(FileIndexRootConfig::new(scope_id, PathBuf::from(root)))
756}
757
758fn is_absolute_file_index_root(root: &str, platform: PlatformKind) -> bool {
759    match platform {
760        PlatformKind::Windows => is_absolute_windows_path(root),
761        _ => PathBuf::from(root).is_absolute(),
762    }
763}
764
765fn is_absolute_windows_path(root: &str) -> bool {
766    let bytes = root.as_bytes();
767    let drive_rooted = bytes.len() >= 3
768        && bytes[0].is_ascii_alphabetic()
769        && bytes[1] == b':'
770        && matches!(bytes[2], b'\\' | b'/');
771    if drive_rooted {
772        return true;
773    }
774
775    if !(root.starts_with("\\\\") || root.starts_with("//")) {
776        return false;
777    }
778    root[2..]
779        .split(['\\', '/'])
780        .filter(|component| !component.is_empty())
781        .take(2)
782        .count()
783        == 2
784}
785
786fn stable_hash64(bytes: &[u8]) -> u64 {
787    const FNV_OFFSET_BASIS: u64 = 0xcbf29ce484222325;
788    const FNV_PRIME: u64 = 0x100000001b3;
789
790    let mut hash = FNV_OFFSET_BASIS;
791    for byte in bytes {
792        hash ^= u64::from(*byte);
793        hash = hash.wrapping_mul(FNV_PRIME);
794    }
795
796    hash
797}
798
799fn duration_millis(duration: Duration) -> u64 {
800    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
801}
802
803fn agent_runtime_budget_ms(request_timeout: Duration) -> u64 {
804    let budget = request_timeout.saturating_sub(Duration::from_millis(1));
805    duration_millis(budget).max(1)
806}
807
808#[cfg(test)]
809#[path = "runtime_tests.rs"]
810mod tests;
relay_knowledge/application/runtime.rs

relay_knowledge/application/
runtime.rs
