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

relay_knowledge/application/runtime/
mod.rs
