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 silent_updates_enabled: bool,
254}
255
256impl WorkerRuntimeConfig {
257    pub const DEFAULT_MAX_IN_FLIGHT: usize = 2;
258
259    /// Builds worker config from typed environment overrides.
260    pub fn from_environment(
261        environment: &EnvironmentConfig,
262    ) -> Result<Self, WorkerRuntimeConfigError> {
263        Ok(Self {
264            embedding_endpoint: validate_worker_endpoint(
265                environment.workers.embedding_endpoint.clone(),
266            )?,
267            ocr_endpoint: validate_worker_endpoint(environment.workers.ocr_endpoint.clone())?,
268            vision_endpoint: validate_worker_endpoint(environment.workers.vision_endpoint.clone())?,
269            extractor_endpoint: validate_worker_endpoint(
270                environment.workers.extractor_endpoint.clone(),
271            )?,
272            max_in_flight: environment
273                .workers
274                .max_in_flight
275                .unwrap_or(Self::DEFAULT_MAX_IN_FLIGHT),
276            silent_updates_enabled: environment.workers.silent_updates_enabled.unwrap_or(false),
277        })
278    }
279
280    /// Returns the configured endpoint for a worker kind.
281    pub fn endpoint_for(&self, kind: WorkerKind) -> Option<&str> {
282        match kind {
283            WorkerKind::Embedding => self.embedding_endpoint.as_deref(),
284            WorkerKind::Ocr => self.ocr_endpoint.as_deref(),
285            WorkerKind::Vision => self.vision_endpoint.as_deref(),
286            WorkerKind::Extractor => self.extractor_endpoint.as_deref(),
287        }
288    }
289}
290
291/// Worker runtime configuration validation error.
292#[derive(Debug, Clone, PartialEq, Eq)]
293pub enum WorkerRuntimeConfigError {
294    InvalidEndpoint(String),
295}
296
297impl fmt::Display for WorkerRuntimeConfigError {
298    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
299        match self {
300            Self::InvalidEndpoint(value) => write!(
301                formatter,
302                "worker endpoint '{value}' must use http:// and include a host"
303            ),
304        }
305    }
306}
307
308impl Error for WorkerRuntimeConfigError {}
309
310/// Resident agent protocol runtime configuration.
311#[derive(Debug, Clone, PartialEq, Eq)]
312pub struct AgentRuntimeConfig {
313    pub mcp_streamable_http_enabled: bool,
314    pub mcp_endpoint: String,
315    pub mcp_allowed_origins: Vec<String>,
316    pub access_policy: AgentAccessPolicy,
317    pub audit_sink_enabled: bool,
318    pub audit_queue_depth: usize,
319}
320
321impl AgentRuntimeConfig {
322    pub const DEFAULT_AUDIT_QUEUE_DEPTH: usize = 1024;
323
324    /// Builds agent protocol config from typed environment overrides.
325    pub fn from_environment(
326        environment: &EnvironmentConfig,
327        request_timeout: Duration,
328    ) -> Result<Self, AgentRuntimeConfigError> {
329        let max_runtime_ms = agent_runtime_budget_ms(request_timeout);
330        let access_policy = AgentAccessPolicy::new(
331            split_csv(environment.agent.mcp_allowed_scopes.as_deref())?,
332            environment
333                .agent
334                .mcp_allow_unspecified_scope
335                .unwrap_or(false),
336            environment
337                .agent
338                .mcp_max_limit
339                .unwrap_or(AgentAccessPolicy::DEFAULT_MAX_LIMIT),
340            environment
341                .agent
342                .mcp_max_context_bytes
343                .unwrap_or(AgentAccessPolicy::DEFAULT_MAX_CONTEXT_BYTES),
344            max_runtime_ms,
345            environment.agent.mcp_allow_remote_clients.unwrap_or(false),
346        )
347        .map_err(AgentRuntimeConfigError::Policy)?;
348
349        Ok(Self {
350            mcp_streamable_http_enabled: environment
351                .agent
352                .mcp_streamable_http_enabled
353                .unwrap_or(false),
354            mcp_endpoint: validate_endpoint(
355                environment.agent.mcp_endpoint.as_deref().unwrap_or("/mcp"),
356            )?,
357            mcp_allowed_origins: split_csv(environment.agent.mcp_allowed_origins.as_deref())?,
358            access_policy,
359            audit_sink_enabled: environment.agent.audit_sink_enabled.unwrap_or(false),
360            audit_queue_depth: environment
361                .agent
362                .audit_queue_depth
363                .unwrap_or(Self::DEFAULT_AUDIT_QUEUE_DEPTH),
364        })
365    }
366
367    /// Returns a copy with streamable HTTP forced on by a CLI command.
368    pub fn with_streamable_http_enabled(mut self) -> Self {
369        self.mcp_streamable_http_enabled = true;
370        self
371    }
372}
373
374/// Agent runtime configuration validation error.
375#[derive(Debug, Clone, PartialEq, Eq)]
376pub enum AgentRuntimeConfigError {
377    InvalidEndpoint(String),
378    EmptyListValue,
379    Policy(AgentPolicyError),
380}
381
382impl fmt::Display for AgentRuntimeConfigError {
383    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
384        match self {
385            Self::InvalidEndpoint(value) => {
386                write!(
387                    formatter,
388                    "MCP endpoint '{value}' must be an absolute HTTP path"
389                )
390            }
391            Self::EmptyListValue => {
392                write!(formatter, "MCP comma-separated values must not be empty")
393            }
394            Self::Policy(error) => write!(formatter, "{error}"),
395        }
396    }
397}
398
399impl Error for AgentRuntimeConfigError {}
400
401/// Error raised while composing foundational runtime configuration.
402#[derive(Debug, Clone, PartialEq, Eq)]
403pub enum RuntimeConfigurationError {
404    Environment(EnvError),
405    Paths(PathError),
406    Network(NetworkConfigError),
407    NetworkRuntime(NetworkRuntimeError),
408    Agent(AgentRuntimeConfigError),
409    Retrieval(RetrievalRuntimeConfigError),
410    Workers(WorkerRuntimeConfigError),
411    FileIndex(FileIndexRuntimeConfigError),
412    Updates(UpdateRuntimeConfigError),
413}
414
415impl fmt::Display for RuntimeConfigurationError {
416    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
417        match self {
418            Self::Environment(error) => write!(formatter, "{error}"),
419            Self::Paths(error) => write!(formatter, "{error}"),
420            Self::Network(error) => write!(formatter, "{error}"),
421            Self::NetworkRuntime(error) => write!(formatter, "{error}"),
422            Self::Agent(error) => write!(formatter, "{error}"),
423            Self::Retrieval(error) => write!(formatter, "{error}"),
424            Self::Workers(error) => write!(formatter, "{error}"),
425            Self::FileIndex(error) => write!(formatter, "{error}"),
426            Self::Updates(error) => write!(formatter, "{error}"),
427        }
428    }
429}
430
431impl Error for RuntimeConfigurationError {}
432
433/// Retrieval runtime configuration validation error.
434#[derive(Debug, Clone, PartialEq, Eq)]
435pub enum RetrievalRuntimeConfigError {
436    InvalidBackend(ReadModelBackendModeError),
437    InvalidRerankBackend(RerankModeError),
438    InvalidProvider(EmbeddingProviderKindError),
439    EmptyModelName(&'static str),
440    MissingRemoteValue(&'static str),
441    InvalidRemoteBaseUrl(String),
442    DimensionTooLarge(usize),
443}
444
445impl fmt::Display for RetrievalRuntimeConfigError {
446    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
447        match self {
448            Self::InvalidBackend(error) => write!(formatter, "{error}"),
449            Self::InvalidRerankBackend(error) => write!(formatter, "{error}"),
450            Self::InvalidProvider(error) => write!(formatter, "{error}"),
451            Self::EmptyModelName(variable) => {
452                write!(formatter, "{variable} must not be blank")
453            }
454            Self::MissingRemoteValue(variable) => {
455                write!(
456                    formatter,
457                    "{variable} is required when a read model backend is external"
458                )
459            }
460            Self::InvalidRemoteBaseUrl(value) => {
461                write!(
462                    formatter,
463                    "embedding base URL '{value}' must use http:// or https://"
464                )
465            }
466            Self::DimensionTooLarge(value) => {
467                write!(formatter, "embedding dimension {value} does not fit in u32")
468            }
469        }
470    }
471}
472
473impl Error for RetrievalRuntimeConfigError {}
474
475fn retrieval_config_from_environment(
476    overrides: &RetrievalEnvOverrides,
477) -> Result<ReadModelBackendConfig, RetrievalRuntimeConfigError> {
478    let semantic_mode = parse_backend_mode(overrides.semantic_backend.as_deref())?;
479    let vector_mode = parse_backend_mode(overrides.vector_backend.as_deref())?;
480    let remote_required = semantic_mode == ReadModelBackendMode::External
481        || vector_mode == ReadModelBackendMode::External;
482    require_remote_model_metadata(overrides, remote_required)?;
483    let dimension = match overrides.embedding_dimension {
484        Some(value) => u32::try_from(value)
485            .map_err(|_| RetrievalRuntimeConfigError::DimensionTooLarge(value))?,
486        None => LOCAL_VECTOR_DIMENSION,
487    };
488    let text_model = model_name_override(
489        overrides.text_embedding_model.as_deref(),
490        RELAY_KNOWLEDGE_TEXT_EMBEDDING_MODEL,
491        LOCAL_VECTOR_MODEL,
492    )?;
493    let semantic_model = model_name_override(
494        overrides.text_embedding_model.as_deref(),
495        RELAY_KNOWLEDGE_TEXT_EMBEDDING_MODEL,
496        LOCAL_SEMANTIC_MODEL,
497    )?;
498    let image_model = model_name_override(
499        overrides.image_embedding_model.as_deref(),
500        RELAY_KNOWLEDGE_IMAGE_EMBEDDING_MODEL,
501        "relay-local-image-hash-v1",
502    )?;
503
504    let remote_embedding = remote_embedding_config_from_environment(overrides, remote_required)?;
505    let rerank = rerank_config_from_environment(overrides)?;
506
507    Ok(ReadModelBackendConfig {
508        semantic_mode,
509        vector_mode,
510        semantic_model: ReadModelMetadata {
511            name: semantic_model,
512            dimension,
513        },
514        vector_model: ReadModelMetadata {
515            name: text_model,
516            dimension,
517        },
518        image_model: ReadModelMetadata {
519            name: image_model,
520            dimension,
521        },
522        remote_embedding,
523        rerank,
524    })
525}
526
527fn rerank_config_from_environment(
528    overrides: &RetrievalEnvOverrides,
529) -> Result<RerankConfig, RetrievalRuntimeConfigError> {
530    let mode = overrides
531        .rerank_backend
532        .as_deref()
533        .map(RerankMode::parse)
534        .transpose()
535        .map_err(RetrievalRuntimeConfigError::InvalidRerankBackend)?
536        .unwrap_or(RerankMode::Local);
537    let model = match mode {
538        RerankMode::Disabled => None,
539        RerankMode::Local => Some(model_name_override(
540            overrides.rerank_model.as_deref(),
541            RELAY_KNOWLEDGE_RERANK_MODEL,
542            LOCAL_RERANK_MODEL,
543        )?),
544        RerankMode::External => overrides
545            .rerank_model
546            .as_deref()
547            .map(|model| model_name_override(Some(model), RELAY_KNOWLEDGE_RERANK_MODEL, ""))
548            .transpose()?,
549    };
550    let timeout = overrides
551        .rerank_timeout_ms
552        .map(Duration::from_millis)
553        .unwrap_or(DEFAULT_RERANK_TIMEOUT);
554
555    Ok(RerankConfig {
556        mode,
557        model,
558        timeout,
559        candidate_multiplier: overrides
560            .rerank_candidate_multiplier
561            .unwrap_or(DEFAULT_RERANK_CANDIDATE_MULTIPLIER),
562        max_candidates: overrides
563            .rerank_max_candidates
564            .unwrap_or(DEFAULT_RERANK_MAX_CANDIDATES),
565    })
566}
567
568fn require_remote_model_metadata(
569    overrides: &RetrievalEnvOverrides,
570    required: bool,
571) -> Result<(), RetrievalRuntimeConfigError> {
572    if !required {
573        return Ok(());
574    }
575    if overrides.text_embedding_model.is_none() {
576        return Err(RetrievalRuntimeConfigError::MissingRemoteValue(
577            RELAY_KNOWLEDGE_TEXT_EMBEDDING_MODEL,
578        ));
579    }
580    if overrides.embedding_dimension.is_none() {
581        return Err(RetrievalRuntimeConfigError::MissingRemoteValue(
582            RELAY_KNOWLEDGE_EMBEDDING_DIMENSION,
583        ));
584    }
585
586    Ok(())
587}
588
589fn remote_embedding_config_from_environment(
590    overrides: &RetrievalEnvOverrides,
591    required: bool,
592) -> Result<Option<RemoteEmbeddingConfig>, RetrievalRuntimeConfigError> {
593    if !required {
594        return Ok(None);
595    }
596    let provider = overrides
597        .llm_provider
598        .as_deref()
599        .map(EmbeddingProviderKind::parse)
600        .transpose()
601        .map_err(RetrievalRuntimeConfigError::InvalidProvider)?
602        .unwrap_or(EmbeddingProviderKind::OpenAiCompatible);
603    let base_url = required_remote_value(
604        overrides.embedding_base_url.as_deref(),
605        RELAY_KNOWLEDGE_EMBEDDING_BASE_URL,
606    )?;
607    if !base_url.starts_with("http://") && !base_url.starts_with("https://") {
608        return Err(RetrievalRuntimeConfigError::InvalidRemoteBaseUrl(base_url));
609    }
610    let api_key = required_remote_value(
611        overrides.embedding_api_key.as_deref(),
612        RELAY_KNOWLEDGE_EMBEDDING_API_KEY,
613    )?;
614    let batch_size = overrides
615        .embedding_batch_size
616        .unwrap_or(DEFAULT_EMBEDDING_BATCH_SIZE);
617    let timeout = overrides
618        .embedding_timeout_ms
619        .map(Duration::from_millis)
620        .unwrap_or(DEFAULT_EMBEDDING_TIMEOUT);
621    let max_concurrency = overrides
622        .embedding_max_concurrency
623        .unwrap_or(DEFAULT_EMBEDDING_MAX_CONCURRENCY);
624
625    Ok(Some(RemoteEmbeddingConfig {
626        provider,
627        base_url,
628        api_key,
629        batch_size,
630        timeout,
631        max_concurrency,
632    }))
633}
634
635fn required_remote_value(
636    value: Option<&str>,
637    variable: &'static str,
638) -> Result<String, RetrievalRuntimeConfigError> {
639    match value.map(str::trim) {
640        Some(trimmed) if !trimmed.is_empty() => Ok(trimmed.to_owned()),
641        _ => Err(RetrievalRuntimeConfigError::MissingRemoteValue(variable)),
642    }
643}
644
645fn model_name_override(
646    value: Option<&str>,
647    variable: &'static str,
648    default: &'static str,
649) -> Result<String, RetrievalRuntimeConfigError> {
650    match value {
651        Some(raw) => {
652            let trimmed = raw.trim();
653            if trimmed.is_empty() {
654                Err(RetrievalRuntimeConfigError::EmptyModelName(variable))
655            } else {
656                Ok(trimmed.to_owned())
657            }
658        }
659        None => Ok(default.to_owned()),
660    }
661}
662
663fn parse_backend_mode(
664    value: Option<&str>,
665) -> Result<ReadModelBackendMode, RetrievalRuntimeConfigError> {
666    value
667        .map(ReadModelBackendMode::parse)
668        .transpose()
669        .map_err(RetrievalRuntimeConfigError::InvalidBackend)
670        .map(|mode| mode.unwrap_or(ReadModelBackendMode::Local))
671}
672
673fn validate_endpoint(value: &str) -> Result<String, AgentRuntimeConfigError> {
674    let trimmed = value.trim();
675    if !trimmed.starts_with('/')
676        || trimmed.contains(char::is_whitespace)
677        || trimmed.contains('?')
678        || trimmed.contains('#')
679    {
680        return Err(AgentRuntimeConfigError::InvalidEndpoint(value.to_owned()));
681    }
682
683    Ok(trimmed.to_owned())
684}
685
686fn validate_worker_endpoint(
687    value: Option<String>,
688) -> Result<Option<String>, WorkerRuntimeConfigError> {
689    value
690        .map(|endpoint| {
691            let trimmed = endpoint.trim();
692            if is_valid_worker_http_endpoint(trimmed) {
693                Ok(trimmed.to_owned())
694            } else {
695                Err(WorkerRuntimeConfigError::InvalidEndpoint(endpoint))
696            }
697        })
698        .transpose()
699}
700
701fn is_valid_worker_http_endpoint(value: &str) -> bool {
702    let Some(remainder) = value.strip_prefix("http://") else {
703        return false;
704    };
705    let authority = remainder
706        .split_once('/')
707        .map_or(remainder, |(authority, _)| authority);
708    if authority.is_empty() || authority.contains(char::is_whitespace) {
709        return false;
710    }
711    if let Some((host, port)) = authority.rsplit_once(':') {
712        return !host.is_empty() && port.parse::<u16>().is_ok_and(|port| port > 0);
713    }
714
715    !authority.is_empty()
716}
717
718fn split_csv(value: Option<&str>) -> Result<Vec<String>, AgentRuntimeConfigError> {
719    value
720        .map(|items| {
721            items
722                .split(',')
723                .map(str::trim)
724                .map(|item| {
725                    if item.is_empty() {
726                        Err(AgentRuntimeConfigError::EmptyListValue)
727                    } else {
728                        Ok(item.to_owned())
729                    }
730                })
731                .collect()
732        })
733        .unwrap_or_else(|| Ok(Vec::new()))
734}
735
736fn split_semicolon(value: Option<&str>) -> Result<Vec<String>, FileIndexRuntimeConfigError> {
737    value
738        .map(|items| {
739            items
740                .split(';')
741                .map(str::trim)
742                .map(|item| {
743                    if item.is_empty() {
744                        Err(FileIndexRuntimeConfigError::EmptyListValue)
745                    } else {
746                        Ok(item.to_owned())
747                    }
748                })
749                .collect()
750        })
751        .unwrap_or_else(|| Ok(Vec::new()))
752}
753
754fn file_index_root_from_environment(
755    scope_id: &'static str,
756    root: String,
757    platform: PlatformKind,
758) -> Result<FileIndexRootConfig, FileIndexRuntimeConfigError> {
759    if !is_absolute_file_index_root(&root, platform) {
760        return Err(FileIndexRuntimeConfigError::RelativeRoot(root));
761    }
762
763    Ok(FileIndexRootConfig::new(scope_id, PathBuf::from(root)))
764}
765
766fn is_absolute_file_index_root(root: &str, platform: PlatformKind) -> bool {
767    match platform {
768        PlatformKind::Windows => is_absolute_windows_path(root),
769        _ => PathBuf::from(root).is_absolute(),
770    }
771}
772
773fn is_absolute_windows_path(root: &str) -> bool {
774    let bytes = root.as_bytes();
775    let drive_rooted = bytes.len() >= 3
776        && bytes[0].is_ascii_alphabetic()
777        && bytes[1] == b':'
778        && matches!(bytes[2], b'\\' | b'/');
779    if drive_rooted {
780        return true;
781    }
782
783    if !(root.starts_with("\\\\") || root.starts_with("//")) {
784        return false;
785    }
786    root[2..]
787        .split(['\\', '/'])
788        .filter(|component| !component.is_empty())
789        .take(2)
790        .count()
791        == 2
792}
793
794fn stable_hash64(bytes: &[u8]) -> u64 {
795    const FNV_OFFSET_BASIS: u64 = 0xcbf29ce484222325;
796    const FNV_PRIME: u64 = 0x100000001b3;
797
798    let mut hash = FNV_OFFSET_BASIS;
799    for byte in bytes {
800        hash ^= u64::from(*byte);
801        hash = hash.wrapping_mul(FNV_PRIME);
802    }
803
804    hash
805}
806
807fn duration_millis(duration: Duration) -> u64 {
808    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
809}
810
811fn agent_runtime_budget_ms(request_timeout: Duration) -> u64 {
812    let budget = request_timeout.saturating_sub(Duration::from_millis(1));
813    duration_millis(budget).max(1)
814}
815
816#[cfg(test)]
817#[path = "runtime_tests.rs"]
818mod tests;
relay_knowledge/application/runtime.rs

relay_knowledge/application/
runtime.rs
