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

relay_knowledge/application/
runtime.rs
