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

relay_knowledge/application/runtime/
mod.rs
