Skip to main content

aion_server/config/
load.rs

1//! [`ServerConfig`]: the complete merged server configuration, its load /
2//! merge / validate pipeline, and the workflow-package discovery helpers.
3//!
4//! This is the assembly point of the config surface: [`ServerConfig`] holds one
5//! field per `[section]` (defined in [`super::sections`]), and its impls carry
6//! the load/merge/CLI-override/default-fill/validate pipeline plus
7//! [`ServerConfig::into_parts`], which splits the durable store config from the
8//! non-secret [`RuntimeConfig`] runtime view. It is re-exported from the
9//! `config` module so every existing `crate::config::X` path resolves
10//! identically.
11
12use std::{
13    collections::HashSet,
14    fs,
15    path::{Path, PathBuf},
16    time::Duration,
17};
18
19use serde::Deserialize;
20
21use crate::error::ServerError;
22
23use super::{
24    AUTHORING_GLEAM_PATH_EMPTY, AUTHORING_PROJECT_ROOT_REQUIRED, AuthConfig, AuthoringConfig,
25    CORS_ALLOWED_ORIGIN_INVALID, CliOverrides, ClusterConfig, ConfigResolution,
26    DEFAULT_CLUSTER_BROADCAST_CAPACITY, DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES,
27    DEFAULT_DEPLOY_MAX_INFLATED_BYTES, DEFAULT_EVENT_BROADCAST_CAPACITY,
28    DEFAULT_OUTBOX_BACKOFF_BASE_MS, DEFAULT_OUTBOX_BACKOFF_MAX_MS,
29    DEFAULT_OUTBOX_BACKOFF_MULTIPLIER, DEFAULT_OUTBOX_BATCH_SIZE, DEFAULT_OUTBOX_MAX_ATTEMPTS,
30    DEFAULT_OUTBOX_POLL_INTERVAL_MS, DEFAULT_QUERY_TIMEOUT_MS, DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED,
31    DEPLOY_MAX_INFLATED_BYTES_REQUIRED, DeployConfig, DevConfig, DrainConfig, HomeSource,
32    ListenConfig, McpConfig, MetricsConfig, NamespaceConfig, NamespaceMode, NamespacesConfig,
33    OUTBOX_BACKOFF_BASE_REQUIRED, OUTBOX_BACKOFF_MAX_REQUIRED, OUTBOX_BACKOFF_MULTIPLIER_REQUIRED,
34    OUTBOX_BATCH_SIZE_REQUIRED, OUTBOX_MAX_ATTEMPTS_REQUIRED, OUTBOX_POLL_INTERVAL_REQUIRED,
35    OUTBOX_RECONCILE_INTERVAL_REQUIRED, OUTBOX_RECONCILE_STALE_AFTER_REQUIRED, ObservabilityConfig,
36    OpsConsoleAssetSource, OpsConsoleConfig, OutboxConfig, QUERY_TIMEOUT_REQUIRED, RuntimeConfig,
37    RuntimeSection, ServerSection, StoreBackend, StoreConfig, TlsConfig, WebSocketConfig,
38    WorkerConfig, WorkerSupervisionConfig, aion_home, config_error, env, file, home,
39    resolution::fill_home_defaults, sections::RetiredStoreInput,
40};
41
42/// Complete merged server configuration.
43#[derive(Clone, Debug, Deserialize)]
44#[serde(default, deny_unknown_fields)]
45#[derive(Default)]
46pub struct ServerConfig {
47    /// Public listener and transport addresses.
48    pub server: ServerSection,
49    /// Event-store backend configuration.
50    pub store: StoreConfig,
51    /// Engine runtime settings.
52    pub runtime: RuntimeSection,
53    /// Shutdown drain settings.
54    pub drain: DrainConfig,
55    /// Authentication settings defined by the operations config surface.
56    pub auth: AuthConfig,
57    /// Metrics endpoint settings.
58    pub metrics: MetricsConfig,
59    /// Namespace defaults.
60    pub namespaces: NamespacesConfig,
61    /// Optional TLS material for transports that require it.
62    pub tls: Option<TlsConfig>,
63    /// Static ops-console asset bundle location.
64    #[serde(alias = "dashboard")]
65    pub ops_console: OpsConsoleConfig,
66    /// Namespace resolver construction mode retained for existing transports.
67    pub namespace: NamespaceConfig,
68    /// Remote-worker heartbeat policy.
69    pub worker: WorkerConfig,
70    /// WebSocket event streaming policy.
71    pub websocket: WebSocketConfig,
72    /// Workflow package archives loaded into the engine at startup.
73    pub workflow_packages: Vec<PathBuf>,
74    /// Operator deploy API settings.
75    pub deploy: DeployConfig,
76    /// AWL studio and optional server-side Gleam authoring settings.
77    pub authoring: AuthoringConfig,
78    /// Local dev-server surface settings.
79    pub dev: DevConfig,
80    /// Durable-outbox fan-out dispatcher settings.
81    pub outbox: OutboxConfig,
82    /// Agent-observability transcript retention bounds.
83    pub observability: ObservabilityConfig,
84    /// Model Context Protocol surface settings.
85    pub mcp: McpConfig,
86    /// Managed-worker supervision policy. Absent = supervision uncommissioned:
87    /// the server runs no worker processes and says so with the remedy.
88    pub worker_supervision: WorkerSupervisionConfig,
89}
90
91pub(crate) struct LoadedConfig {
92    pub(crate) config: ServerConfig,
93    pub(crate) resolution: ConfigResolution,
94}
95
96/// Expand a leading `~` in the operator-supplied filesystem paths.
97///
98/// #180 review m10: `~` already expands for `AION_HOME` and `--config`, so an
99/// operator naturally writes it in the path VALUES too — where it used to pass
100/// through untouched and silently become a literal `./~` directory. The same
101/// [`home::expand_tilde`] rule applies to `store.data_dir` and
102/// `authoring.workspace_dir` (the two home-adjacent state roots): `~`/`~/…`
103/// expand against `$HOME`, `~user` is refused by name, and a path with no
104/// tilde passes through byte-for-byte.
105fn expand_operator_path_tildes(config: &mut ServerConfig) -> Result<(), ServerError> {
106    if let Some(data_dir) = config.store.data_dir.as_deref() {
107        let expanded = home::expand_tilde(Path::new(data_dir))?;
108        let expanded = expanded.to_str().ok_or_else(|| ServerError::Config {
109            message: format!(
110                "store.data_dir `{data_dir}` expands to a path that is not valid UTF-8"
111            ),
112        })?;
113        config.store.data_dir = Some(expanded.to_owned());
114    }
115    if let Some(workspace_dir) = config.authoring.workspace_dir.as_deref() {
116        config.authoring.workspace_dir = Some(home::expand_tilde(workspace_dir)?);
117    }
118    Ok(())
119}
120
121impl ServerConfig {
122    /// Load and merge config from defaults, optional TOML file, environment, and CLI overrides.
123    ///
124    /// # Errors
125    ///
126    /// Returns [`ServerError::Config`] when file discovery, parsing, environment parsing, CLI
127    /// values, or validation fail.
128    pub fn load(cli: &CliOverrides) -> Result<Self, ServerError> {
129        Ok(Self::load_resolved(cli)?.config)
130    }
131
132    pub(crate) fn load_resolved(cli: &CliOverrides) -> Result<LoadedConfig, ServerError> {
133        let home = aion_home()?;
134        let working_dir = std::env::current_dir().map_err(|source| ServerError::Config {
135            message: format!(
136                "failed to resolve the current directory for config discovery: {source}"
137            ),
138        })?;
139        Self::load_in(cli, &home.path, home.source, &working_dir, true)
140    }
141
142    fn load_in(
143        cli: &CliOverrides,
144        home: &Path,
145        home_source: HomeSource,
146        working_dir: &Path,
147        overlay_environment: bool,
148    ) -> Result<LoadedConfig, ServerError> {
149        // #152: `--config ~/aion.toml` reaches this point unexpanded whenever a
150        // shell did not get to it (quoted values, env files, process managers);
151        // expand it here so the operator's spelling and the loader agree.
152        let explicit = cli
153            .config_path
154            .as_deref()
155            .map(super::home::expand_tilde)
156            .transpose()?;
157        let discovered = file::discover(explicit.as_deref(), home, working_dir)?;
158        let mut config = match discovered.bytes {
159            Some(bytes) => Self::parse_unresolved(&bytes).map_err(|error| ServerError::Config {
160                message: format!("failed to parse {}: {error}", discovered.source),
161            })?,
162            None => Self::default(),
163        };
164        if overlay_environment {
165            env::overlay(&mut config)?;
166        }
167        config.apply_cli_overrides(cli);
168        expand_operator_path_tildes(&mut config)?;
169        #[cfg(not(unix))]
170        let home_explicit = !overlay_environment || std::env::var_os("AION_HOME").is_some();
171        #[cfg(not(unix))]
172        let data_dir_explicit = config.store.data_dir.is_some();
173        #[cfg(not(unix))]
174        let data_root_required = matches!(config.store.backend, StoreBackend::Haematite);
175        #[cfg(not(unix))]
176        let authoring_workspace_explicit = config.authoring.workspace_dir.is_some();
177        config.load_discovered_workflow_packages(cli, working_dir)?;
178        let legacy_notices = fill_home_defaults(&mut config, home, home_source, working_dir)?;
179        config.fill_operational_defaults();
180        config.validate()?;
181        let resolution = ConfigResolution {
182            home: home.to_owned(),
183            source: discovered.source,
184            data_dir: config.store.data_dir.clone(),
185            authoring_workspace: config.authoring.workspace_dir.clone(),
186            legacy_notices,
187            #[cfg(not(unix))]
188            home_explicit,
189            #[cfg(not(unix))]
190            data_dir_explicit,
191            #[cfg(not(unix))]
192            data_root_required,
193            #[cfg(not(unix))]
194            authoring_workspace_explicit,
195        };
196        Ok(LoadedConfig { config, resolution })
197    }
198
199    /// Load with an explicitly supplied home, without environment or CLI
200    /// overlays.
201    ///
202    /// `home_source` is a parameter rather than a constant because the two
203    /// worlds it selects between are exactly what the legacy-directory guard
204    /// turns on, and a test must be able to say which world it is standing in.
205    /// A test that could only ever simulate one of them would pin half the
206    /// behaviour and call it covered.
207    #[cfg(test)]
208    pub(super) fn load_for_test(
209        cli: &CliOverrides,
210        home: &Path,
211        home_source: HomeSource,
212        working_dir: &Path,
213    ) -> Result<LoadedConfig, ServerError> {
214        Self::load_in(cli, home, home_source, working_dir, false)
215    }
216
217    /// Fill operational tuning knobs that have a sane default when omitted, so a
218    /// minimal or empty config boots without forcing the operator to hand-author
219    /// values that are pure tuning. Uses `get_or_insert`, so an explicitly set
220    /// value (including a misconfigured `0`, which [`Self::validate`] still
221    /// rejects) is left untouched; only an absent (`None`) field is defaulted.
222    fn fill_operational_defaults(&mut self) {
223        self.runtime
224            .query_timeout_ms
225            .get_or_insert(DEFAULT_QUERY_TIMEOUT_MS);
226        self.websocket
227            .event_broadcast_capacity
228            .get_or_insert(DEFAULT_EVENT_BROADCAST_CAPACITY);
229        self.websocket
230            .cluster_broadcast_capacity
231            .get_or_insert(DEFAULT_CLUSTER_BROADCAST_CAPACITY);
232        self.fill_outbox_defaults();
233        self.fill_deploy_defaults();
234    }
235
236    /// Fill the durable-outbox tuning knobs with sane defaults when the
237    /// dispatcher is enabled but a knob was omitted, so turning the feature on
238    /// does not force hand-authoring pure tuning. Inert while `outbox.enabled`
239    /// is false (the knobs are never read behind the gate). The reconciliation
240    /// pair is intentionally NOT defaulted: when both are absent reconciliation
241    /// stays dark, so forcing a default would silently commission a sweep.
242    /// `get_or_insert` leaves any explicit value (including a misconfigured `0`,
243    /// which [`Self::validate_outbox`] still rejects) untouched.
244    fn fill_outbox_defaults(&mut self) {
245        if !self.outbox.enabled {
246            return;
247        }
248        self.outbox
249            .poll_interval_ms
250            .get_or_insert(DEFAULT_OUTBOX_POLL_INTERVAL_MS);
251        self.outbox
252            .batch_size
253            .get_or_insert(DEFAULT_OUTBOX_BATCH_SIZE);
254        self.outbox
255            .max_attempts
256            .get_or_insert(DEFAULT_OUTBOX_MAX_ATTEMPTS);
257        self.outbox
258            .backoff_base_ms
259            .get_or_insert(DEFAULT_OUTBOX_BACKOFF_BASE_MS);
260        self.outbox
261            .backoff_multiplier
262            .get_or_insert(DEFAULT_OUTBOX_BACKOFF_MULTIPLIER);
263        self.outbox
264            .backoff_max_ms
265            .get_or_insert(DEFAULT_OUTBOX_BACKOFF_MAX_MS);
266    }
267
268    /// Fill the deploy decompression-bomb ceilings with conservative defaults
269    /// when the deploy surface is enabled but a ceiling was omitted, so turning
270    /// the feature on boots rather than refusing for want of a security knob.
271    /// Inert while `deploy.enabled` is false (the ceilings are never read with
272    /// the surface dark). `get_or_insert` leaves any explicit value (including a
273    /// misconfigured `0` or an inflate ceiling below the archive ceiling, both
274    /// still rejected by [`Self::validate`]) untouched.
275    fn fill_deploy_defaults(&mut self) {
276        if !self.deploy.enabled {
277            return;
278        }
279        self.deploy
280            .max_archive_bytes
281            .get_or_insert(DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES);
282        self.deploy
283            .max_inflated_bytes
284            .get_or_insert(DEFAULT_DEPLOY_MAX_INFLATED_BYTES);
285    }
286
287    fn load_discovered_workflow_packages(
288        &mut self,
289        cli: &CliOverrides,
290        directory: &Path,
291    ) -> Result<(), ServerError> {
292        let discovered_packages = discover_workflow_packages(directory)?;
293        merge_workflow_packages(
294            &mut self.workflow_packages,
295            discovered_packages,
296            &cli.workflow_packages,
297        );
298        Ok(())
299    }
300
301    /// Parse server configuration from TOML bytes and validate it.
302    ///
303    /// # Errors
304    ///
305    /// Returns [`ServerError::Config`] when parsing fails or values are invalid.
306    pub fn from_slice(bytes: &[u8]) -> Result<Self, ServerError> {
307        let home = aion_home()?;
308        Self::from_slice_with_home(bytes, &home.path)
309    }
310
311    /// Parse server configuration using an explicitly supplied Aion home.
312    ///
313    /// This is useful for embedded callers and tests that must not consult or
314    /// touch the process user's real home. It applies the same dynamic defaults
315    /// as [`Self::load`], but does not apply environment or CLI overlays, and
316    /// naming the home here suppresses the working-directory legacy fallback
317    /// exactly as setting `AION_HOME` does — an embedded caller that supplies a
318    /// home is isolating this server's state, not asking to inherit whatever
319    /// directory the process happens to have been started in.
320    ///
321    /// # Errors
322    ///
323    /// Returns [`ServerError::Config`] when parsing, current-directory
324    /// resolution, dynamic path defaulting, or validation fails.
325    pub fn from_slice_with_home(bytes: &[u8], home: &Path) -> Result<Self, ServerError> {
326        let working_dir = std::env::current_dir().map_err(|source| ServerError::Config {
327            message: format!(
328                "failed to resolve the current directory for config defaults: {source}"
329            ),
330        })?;
331        let mut config = Self::parse_unresolved(bytes)?;
332        // The caller named the home in the call itself, which is the same
333        // assertion `AION_HOME` makes and is treated identically: a legacy
334        // directory in the working directory must not override it.
335        fill_home_defaults(&mut config, home, HomeSource::Explicit, &working_dir)?;
336        config.fill_operational_defaults();
337        config.validate()?;
338        Ok(config)
339    }
340
341    fn parse_unresolved(bytes: &[u8]) -> Result<Self, ServerError> {
342        toml::from_slice(bytes).map_err(|source| ServerError::Config {
343            message: format!("invalid server config: {source}"),
344        })
345    }
346
347    /// Load server configuration from an explicit TOML file path.
348    ///
349    /// # Errors
350    ///
351    /// Returns [`ServerError::Config`] when the file is missing, unreadable, unparsable, or invalid.
352    pub fn load_from_path(path: impl Into<PathBuf>) -> Result<Self, ServerError> {
353        file::load_required(&path.into())
354    }
355
356    /// Split store configuration from non-secret runtime settings.
357    #[must_use]
358    pub fn into_parts(self) -> (StoreConfig, RuntimeConfig) {
359        let runtime = RuntimeConfig {
360            listen: ListenConfig {
361                grpc: self.server.grpc_address,
362                http: self.server.listen_address,
363            },
364            tls: self.tls,
365            auth: self.auth,
366            ops_console: self.ops_console,
367            namespace: self.namespace,
368            worker: self.worker,
369            websocket: self.websocket,
370            workflow_packages: self.workflow_packages,
371            deploy: self.deploy,
372            authoring: self.authoring,
373            dev: self.dev,
374            outbox: self.outbox,
375            observability: self.observability,
376            mcp: self.mcp.resolved(),
377            scheduler_threads: self.runtime.scheduler_threads,
378            jit_threshold: self.runtime.jit_threshold,
379            query_timeout: self.runtime.query_timeout_ms.map(Duration::from_millis),
380            default_namespace: self.namespaces.default,
381            auto_create: self.namespaces.auto_create,
382            max_in_flight_activities: self.namespaces.max_in_flight_activities,
383            drain_timeout: Duration::from_secs(self.drain.timeout_seconds),
384            metrics: self.metrics,
385            owned_shards: self.store.owned_shards.clone(),
386            cors_allowed_origins: self.server.cors_allowed_origins.clone(),
387        };
388        (self.store, runtime)
389    }
390
391    fn apply_cli_overrides(&mut self, cli: &CliOverrides) {
392        if let Some(address) = cli.listen_address {
393            self.server.listen_address = address;
394        }
395        if let Some(address) = cli.grpc_address {
396            self.server.grpc_address = address;
397        }
398        if cli.store_url.is_some() {
399            self.store.retired_input = Some(RetiredStoreInput::Flag);
400        }
401        if let Some(threads) = cli.scheduler_threads {
402            self.runtime.scheduler_threads = threads;
403        }
404        if let Some(threshold) = cli.jit_threshold {
405            self.runtime.jit_threshold = Some(threshold);
406        }
407        if let Some(timeout) = cli.drain_timeout_seconds {
408            self.drain.timeout_seconds = timeout;
409        }
410        if let Some(gleam_path) = &cli.gleam_path {
411            self.authoring.gleam_path = Some(gleam_path.clone());
412        }
413        if let Some(project_root) = &cli.authoring_project_root {
414            self.authoring.project_root = Some(project_root.clone());
415        }
416    }
417
418    fn validate(&self) -> Result<(), ServerError> {
419        if let Some(input) = self.store.retired_input {
420            let found = match input {
421                RetiredStoreInput::Backend => "backend = \"libsql\"",
422                RetiredStoreInput::BackendEnvironment => "AION_STORE_BACKEND=libsql",
423                RetiredStoreInput::Url => "store.url",
424                RetiredStoreInput::Environment => "AION_STORE_URL",
425                RetiredStoreInput::Flag => "--store-url",
426            };
427            return config_error(format!(
428                "found retired {found}; use backend = \"haematite\" with store.data_dir and store.node_cache_budget (all three are required — haematite refuses to start without the budget); existing libsql database files are not read by this build"
429            ));
430        }
431        if self.server.listen_address.port() == 0 {
432            return config_error("server.listen_address must use an explicit non-zero port");
433        }
434        if self.server.grpc_address.port() == 0 {
435            return config_error("server.grpc_address must use an explicit non-zero port");
436        }
437        validate_cors_origins(&self.server.cors_allowed_origins)?;
438        if self.runtime.scheduler_threads == 0 {
439            return config_error("runtime.scheduler_threads must be greater than zero");
440        }
441        if self.drain.timeout_seconds == 0 {
442            return config_error("drain.timeout_seconds must be greater than zero");
443        }
444        if self.auth.enabled && self.auth.jwks_url.as_deref().is_none_or(str::is_empty) {
445            return config_error("auth.jwks_url must not be empty when auth.enabled is true");
446        }
447        if self.auth.jwks_refresh_seconds == 0 {
448            return config_error("auth.jwks_refresh_seconds must be greater than zero");
449        }
450        if self.namespaces.default.is_empty() {
451            return config_error("namespaces.default must not be empty");
452        }
453        if matches!(self.store.backend, StoreBackend::Haematite) {
454            if self.store.data_dir.as_deref().is_none_or(str::is_empty) {
455                return config_error(
456                    "store.data_dir must not be empty when store.backend is haematite",
457                );
458            }
459            if self.store.shard_count == 0 {
460                return config_error("store.shard_count must be greater than zero");
461            }
462            if let Some(cluster) = &self.store.cluster {
463                validate_cluster(cluster)?;
464            }
465        } else if self.store.cluster.is_some() {
466            return config_error("store.cluster is only valid when store.backend is haematite");
467        }
468        if let OpsConsoleAssetSource::FileSystem { asset_path } = &self.ops_console.source {
469            if asset_path.as_os_str().is_empty() {
470                return config_error("ops_console.source.FileSystem.asset_path must not be empty");
471            }
472        }
473        if let NamespaceMode::SingleTenant { namespace } = &self.namespace.mode {
474            if namespace.is_empty() {
475                return config_error("namespace.mode.SingleTenant.namespace must not be empty");
476            }
477        }
478        if self.worker.heartbeat_window.is_zero() {
479            return config_error("worker.heartbeat_window must be greater than zero");
480        }
481        self.websocket.validate()?;
482        self.observability.validate()?;
483        self.mcp.validate()?;
484        match self.runtime.query_timeout_ms {
485            None | Some(0) => return config_error(QUERY_TIMEOUT_REQUIRED),
486            Some(_) => {}
487        }
488        if self.deploy.enabled {
489            let max_archive_bytes = match self.deploy.max_archive_bytes {
490                None | Some(0) => return config_error(DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED),
491                Some(value) => value,
492            };
493            let max_inflated_bytes = match self.deploy.max_inflated_bytes {
494                None | Some(0) => return config_error(DEPLOY_MAX_INFLATED_BYTES_REQUIRED),
495                Some(value) => value,
496            };
497            // Both ceilings size in-memory buffers, so they must be
498            // addressable on this platform (32-bit targets).
499            ensure_fits_usize("deploy.max_archive_bytes", max_archive_bytes)?;
500            ensure_fits_usize("deploy.max_inflated_bytes", max_inflated_bytes)?;
501            if max_inflated_bytes < max_archive_bytes {
502                return config_error(format!(
503                    "deploy.max_inflated_bytes ({max_inflated_bytes}) must be at least deploy.max_archive_bytes ({max_archive_bytes}): an inflate ceiling below the upload ceiling would refuse archives the upload ceiling admits, even stored uncompressed"
504                ));
505            }
506        }
507        if let Some(gleam_path) = &self.authoring.gleam_path {
508            // The authoring surface is commissioned by a non-empty gleam_path;
509            // an empty value is a misconfiguration, not "dark".
510            if gleam_path.as_os_str().is_empty() {
511                return config_error(AUTHORING_GLEAM_PATH_EMPTY);
512            }
513            // Commissioning the loop requires a project root with no default
514            // (a Gleam project cannot be invented; the operator provisions it).
515            match &self.authoring.project_root {
516                Some(root) if !root.as_os_str().is_empty() => {}
517                _ => return config_error(AUTHORING_PROJECT_ROOT_REQUIRED),
518            }
519        }
520        self.validate_outbox()?;
521        // Resolution IS the validation: a partial or nonsensical
522        // `[worker_supervision]` section refuses here, at load, rather than at
523        // the first `aion worker start` hours later.
524        self.worker_supervision.resolve()?;
525        Ok(())
526    }
527
528    /// Validate the durable-outbox dispatcher knobs.
529    ///
530    /// All knobs are inert while `outbox.enabled` is false (the dispatcher is
531    /// never spawned), so they are only required — and only checked — once the
532    /// operator commissions the dispatcher. This mirrors the dark-by-default
533    /// `deploy` surface: the on/off gate carries no defaults, and every
534    /// operational value behind it is an explicit operator decision.
535    fn validate_outbox(&self) -> Result<(), ServerError> {
536        if !self.outbox.enabled {
537            return Ok(());
538        }
539        match self.outbox.poll_interval_ms {
540            None | Some(0) => return config_error(OUTBOX_POLL_INTERVAL_REQUIRED),
541            Some(_) => {}
542        }
543        match self.outbox.batch_size {
544            None | Some(0) => return config_error(OUTBOX_BATCH_SIZE_REQUIRED),
545            Some(_) => {}
546        }
547        match self.outbox.max_attempts {
548            None | Some(0) => return config_error(OUTBOX_MAX_ATTEMPTS_REQUIRED),
549            Some(_) => {}
550        }
551        let backoff_base_ms = match self.outbox.backoff_base_ms {
552            None | Some(0) => return config_error(OUTBOX_BACKOFF_BASE_REQUIRED),
553            Some(value) => value,
554        };
555        match self.outbox.backoff_multiplier {
556            None | Some(0) => return config_error(OUTBOX_BACKOFF_MULTIPLIER_REQUIRED),
557            Some(_) => {}
558        }
559        match self.outbox.backoff_max_ms {
560            Some(max) if max >= backoff_base_ms => {}
561            _ => return config_error(OUTBOX_BACKOFF_MAX_REQUIRED),
562        }
563        match (
564            self.outbox.reconcile_interval_ms,
565            self.outbox.reconcile_stale_after_ms,
566        ) {
567            (None, None) => {}
568            (None | Some(0), _) => return config_error(OUTBOX_RECONCILE_INTERVAL_REQUIRED),
569            (_, None | Some(0)) => return config_error(OUTBOX_RECONCILE_STALE_AFTER_REQUIRED),
570            (Some(_), Some(_)) => {}
571        }
572        Ok(())
573    }
574}
575
576/// Validate a `[store.cluster]` section: a non-empty node id, and every member /
577/// peer name non-empty. A cluster of one (no peers, members empty or `[node_id]`)
578/// is valid.
579fn validate_cluster(cluster: &ClusterConfig) -> Result<(), ServerError> {
580    if cluster.node_id.is_empty() {
581        return config_error("store.cluster.node_id must not be empty");
582    }
583    if cluster.members.iter().any(String::is_empty) {
584        return config_error("store.cluster.members entries must not be empty");
585    }
586    if cluster.peers.iter().any(|peer| peer.name.is_empty()) {
587        return config_error("store.cluster.peers entries must name a non-empty node");
588    }
589    if matches!(cluster.failover_poll_interval_ms, Some(0)) {
590        return config_error(
591            "store.cluster.failover_poll_interval_ms must be greater than zero when set",
592        );
593    }
594    if matches!(cluster.failover_confirmations, Some(0)) {
595        return config_error("store.cluster.failover_confirmations must be at least one when set");
596    }
597    Ok(())
598}
599
600/// Validate every `[server] cors_allowed_origins` entry.
601fn validate_cors_origins(origins: &[String]) -> Result<(), ServerError> {
602    for origin in origins {
603        validate_cors_origin(origin)?;
604    }
605    Ok(())
606}
607
608/// Validate one `[server] cors_allowed_origins` entry: it must be a non-empty,
609/// parseable HTTP origin so the `CorsLayer` can match it against the browser's
610/// `Origin` header. A malformed origin can never match a real request, so it is
611/// a misconfiguration caught at startup rather than silently never matching.
612fn validate_cors_origin(origin: &str) -> Result<(), ServerError> {
613    if origin.is_empty() {
614        return config_error(CORS_ALLOWED_ORIGIN_INVALID);
615    }
616    // An origin is scheme + host + optional port and carries no path: reject a
617    // trailing slash or any path segment, which would never equal a browser
618    // `Origin` header value.
619    let scheme_split = origin.split_once("://");
620    let Some((scheme, authority)) = scheme_split else {
621        return config_error(CORS_ALLOWED_ORIGIN_INVALID);
622    };
623    if scheme.is_empty() || authority.is_empty() || authority.contains('/') {
624        return config_error(CORS_ALLOWED_ORIGIN_INVALID);
625    }
626    // It must parse as an HTTP header value (the form the CorsLayer compares).
627    if origin.parse::<axum::http::HeaderValue>().is_err() {
628        return config_error(CORS_ALLOWED_ORIGIN_INVALID);
629    }
630    Ok(())
631}
632
633/// Refuses byte-ceiling values that cannot index memory on this platform.
634fn ensure_fits_usize(key: &str, value: u64) -> Result<(), ServerError> {
635    if usize::try_from(value).is_err() {
636        return config_error(format!(
637            "{key} ({value}) exceeds this platform's addressable memory; set it to at most {}",
638            usize::MAX
639        ));
640    }
641    Ok(())
642}
643
644fn discover_workflow_packages(directory: &Path) -> Result<Vec<PathBuf>, ServerError> {
645    let mut packages = Vec::new();
646    let entries = fs::read_dir(directory).map_err(|source| ServerError::Config {
647        message: format!(
648            "failed to scan workflow packages in `{}`: {source}",
649            directory.display()
650        ),
651    })?;
652
653    for entry in entries {
654        let entry = entry.map_err(|source| ServerError::Config {
655            message: format!(
656                "failed to read workflow package entry in `{}`: {source}",
657                directory.display()
658            ),
659        })?;
660        let path = entry.path();
661        let has_aion_extension = path
662            .extension()
663            .is_some_and(|extension| extension == "aion");
664        if path.is_file() && has_aion_extension {
665            packages.push(path);
666        }
667    }
668
669    packages.sort_by(|left, right| left.as_os_str().cmp(right.as_os_str()));
670    Ok(packages)
671}
672
673fn merge_workflow_packages(
674    workflow_packages: &mut Vec<PathBuf>,
675    discovered_packages: Vec<PathBuf>,
676    cli_packages: &[PathBuf],
677) {
678    let mut seen: HashSet<PathBuf> = workflow_packages
679        .iter()
680        .map(|package| deduplicated_package_key(package))
681        .collect();
682    for package in discovered_packages
683        .into_iter()
684        .chain(cli_packages.iter().cloned())
685    {
686        if seen.insert(deduplicated_package_key(&package)) {
687            workflow_packages.push(package);
688        }
689    }
690}
691
692fn deduplicated_package_key(path: &Path) -> PathBuf {
693    path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
694}
695
696#[cfg(test)]
697#[path = "load_home_tests.rs"]
698mod home_tests;
699
700#[cfg(test)]
701mod tests {
702    use crate::config::{
703        AutoCreate, DEFAULT_AUTHORING_WORKSPACE_DIR, DEFAULT_MAX_IN_FLIGHT_ACTIVITIES, HomeSource,
704        OpsConsoleAssetSource,
705    };
706
707    use super::{
708        CliOverrides, DEFAULT_CLUSTER_BROADCAST_CAPACITY, DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES,
709        DEFAULT_DEPLOY_MAX_INFLATED_BYTES, DEFAULT_EVENT_BROADCAST_CAPACITY,
710        DEFAULT_OUTBOX_BACKOFF_BASE_MS, DEFAULT_OUTBOX_BACKOFF_MAX_MS,
711        DEFAULT_OUTBOX_BACKOFF_MULTIPLIER, DEFAULT_OUTBOX_BATCH_SIZE, DEFAULT_OUTBOX_MAX_ATTEMPTS,
712        DEFAULT_OUTBOX_POLL_INTERVAL_MS, DEFAULT_QUERY_TIMEOUT_MS, ServerConfig, StoreBackend,
713        discover_workflow_packages, merge_workflow_packages,
714    };
715
716    #[test]
717    fn valid_toml_is_parsed_into_typed_config() -> Result<(), Box<dyn std::error::Error>> {
718        let config = ServerConfig::from_slice(
719            br#"
720                [server]
721                listen_address = "127.0.0.1:18080"
722                grpc_address = "127.0.0.1:15051"
723
724                [store]
725                backend = "haematite"
726                data_dir = "aion-data"
727
728                [runtime]
729                scheduler_threads = 2
730                query_timeout_ms = 10000
731
732                [drain]
733                timeout_seconds = 45
734
735                [auth]
736                enabled = true
737                jwks_url = "https://issuer.example.com/.well-known/jwks.json"
738                jwks_refresh_seconds = 60
739
740                [metrics]
741                enabled = true
742
743                [namespaces]
744                default = "production"
745
746                [websocket]
747                outbound_buffer_bound = 16
748                event_broadcast_capacity = 1024
749                cluster_broadcast_capacity = 1024
750            "#,
751        )?;
752
753        assert_eq!(config.store.backend, StoreBackend::Haematite);
754        assert_eq!(config.store.data_dir.as_deref(), Some("aion-data"));
755        assert_eq!(config.runtime.scheduler_threads, 2);
756        assert_eq!(config.runtime.query_timeout_ms, Some(10_000));
757        assert_eq!(config.namespaces.default, "production");
758        // `auto_create` is omitted above, so it resolves to the Open default.
759        assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
760        // `max_in_flight_activities` is omitted above, so it resolves to the
761        // generous platform default.
762        assert_eq!(
763            config.namespaces.max_in_flight_activities,
764            DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
765        );
766        assert_eq!(config.websocket.outbound_buffer_bound, 16);
767        assert_eq!(config.websocket.event_broadcast_capacity, Some(1024));
768        Ok(())
769    }
770
771    #[test]
772    fn namespaces_auto_create_closed_parses() -> Result<(), Box<dyn std::error::Error>> {
773        let config = ServerConfig::from_slice(
774            br#"
775                [namespaces]
776                default = "production"
777                auto_create = "closed"
778            "#,
779        )?;
780        assert_eq!(config.namespaces.default, "production");
781        assert_eq!(config.namespaces.auto_create, AutoCreate::Closed);
782        Ok(())
783    }
784
785    #[test]
786    fn namespaces_auto_create_open_parses() -> Result<(), Box<dyn std::error::Error>> {
787        let config = ServerConfig::from_slice(
788            br#"
789                [namespaces]
790                auto_create = "open"
791            "#,
792        )?;
793        assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
794        Ok(())
795    }
796
797    #[test]
798    fn namespaces_max_in_flight_activities_override_parses()
799    -> Result<(), Box<dyn std::error::Error>> {
800        let config = ServerConfig::from_slice(
801            br#"
802                [namespaces]
803                default = "production"
804                max_in_flight_activities = 32
805            "#,
806        )?;
807        assert_eq!(config.namespaces.max_in_flight_activities, 32);
808        // The override also propagates into the runtime view.
809        let (_store, runtime) = config.into_parts();
810        assert_eq!(runtime.max_in_flight_activities, 32);
811        Ok(())
812    }
813
814    #[test]
815    fn namespaces_max_in_flight_activities_defaults_when_omitted()
816    -> Result<(), Box<dyn std::error::Error>> {
817        // An old/minimal config that predates the field omits it entirely and
818        // resolves to the generous platform default (additive, not a migration).
819        let config = ServerConfig::from_slice(
820            br#"
821                [namespaces]
822                default = "production"
823            "#,
824        )?;
825        assert_eq!(
826            config.namespaces.max_in_flight_activities,
827            DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
828        );
829        Ok(())
830    }
831
832    #[test]
833    fn namespaces_auto_create_rejects_unknown_variant() {
834        let result = ServerConfig::from_slice(
835            br#"
836                [namespaces]
837                auto_create = "sometimes"
838            "#,
839        );
840        assert!(
841            result.is_err(),
842            "an unknown auto_create variant must fail to parse"
843        );
844    }
845
846    #[test]
847    fn missing_event_broadcast_capacity_uses_default() -> Result<(), Box<dyn std::error::Error>> {
848        // The server unconditionally mounts /events/stream, but the channel
849        // capacity is a tuning knob: omitting it must resolve to the default and
850        // boot, not fail startup.
851        let config = ServerConfig::from_slice(
852            br"
853                [runtime]
854                query_timeout_ms = 10000
855
856                [websocket]
857                cluster_broadcast_capacity = 64
858            ",
859        )?;
860        assert_eq!(
861            config.websocket.event_broadcast_capacity,
862            Some(DEFAULT_EVENT_BROADCAST_CAPACITY),
863            "omitted event_broadcast_capacity must resolve to the default"
864        );
865        Ok(())
866    }
867
868    #[test]
869    fn zero_event_broadcast_capacity_fails_startup_validation() {
870        let result = ServerConfig::from_slice(
871            br"
872                [websocket]
873                event_broadcast_capacity = 0
874            ",
875        );
876
877        let message = result
878            .err()
879            .map_or_else(String::new, |error| error.to_string());
880        assert!(
881            message.contains("websocket.event_broadcast_capacity"),
882            "validation message must name the zero-valued key: {message}"
883        );
884    }
885
886    #[test]
887    fn missing_cluster_broadcast_capacity_uses_default() -> Result<(), Box<dyn std::error::Error>> {
888        // A config that sizes the workflow channel but omits the low-rate cluster
889        // channel must resolve the cluster capacity to its default and boot, not
890        // fail loudly.
891        let config = ServerConfig::from_slice(
892            br"
893                [runtime]
894                scheduler_threads = 1
895                query_timeout_ms = 10000
896
897                [websocket]
898                event_broadcast_capacity = 64
899            ",
900        )?;
901        assert_eq!(
902            config.websocket.cluster_broadcast_capacity,
903            Some(DEFAULT_CLUSTER_BROADCAST_CAPACITY),
904            "omitted cluster_broadcast_capacity must resolve to the default"
905        );
906        Ok(())
907    }
908
909    #[test]
910    fn zero_cluster_broadcast_capacity_fails_startup_validation() {
911        let result = ServerConfig::from_slice(
912            br"
913                [runtime]
914                query_timeout_ms = 10000
915
916                [websocket]
917                event_broadcast_capacity = 64
918                cluster_broadcast_capacity = 0
919            ",
920        );
921
922        let message = result
923            .err()
924            .map_or_else(String::new, |error| error.to_string());
925        assert!(
926            message.contains("websocket.cluster_broadcast_capacity"),
927            "validation message must name the zero-valued cluster key: {message}"
928        );
929    }
930
931    /// An omitted `[observability]` section resolves both retention bounds to
932    /// their defaults and boots — retention is on out of the box, never a
933    /// forced operator decision.
934    #[test]
935    fn missing_observability_section_uses_defaults() -> Result<(), Box<dyn std::error::Error>> {
936        let config = ServerConfig::from_slice(
937            br"
938                [runtime]
939                query_timeout_ms = 10000
940
941                [websocket]
942                event_broadcast_capacity = 64
943                cluster_broadcast_capacity = 64
944            ",
945        )?;
946        assert_eq!(
947            config.observability.max_event_bytes,
948            crate::config::DEFAULT_OBSERVABILITY_MAX_EVENT_BYTES
949        );
950        assert_eq!(
951            config.observability.max_stream_events,
952            crate::config::DEFAULT_OBSERVABILITY_MAX_STREAM_EVENTS
953        );
954        // The bounds also ride into the runtime view the server state reads.
955        let (_store, runtime) = config.into_parts();
956        assert_eq!(
957            runtime.observability.max_event_bytes,
958            crate::config::DEFAULT_OBSERVABILITY_MAX_EVENT_BYTES
959        );
960        Ok(())
961    }
962
963    /// Explicit `[observability]` values parse and round-trip into the runtime
964    /// view.
965    #[test]
966    fn observability_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
967        let config = ServerConfig::from_slice(
968            br"
969                [runtime]
970                query_timeout_ms = 10000
971
972                [websocket]
973                event_broadcast_capacity = 64
974                cluster_broadcast_capacity = 64
975
976                [observability]
977                max_event_bytes = 512
978                max_stream_events = 3
979            ",
980        )?;
981        assert_eq!(config.observability.max_event_bytes, 512);
982        assert_eq!(config.observability.max_stream_events, 3);
983        let (_store, runtime) = config.into_parts();
984        assert_eq!(runtime.observability.max_event_bytes, 512);
985        assert_eq!(runtime.observability.max_stream_events, 3);
986        Ok(())
987    }
988
989    /// An omitted `[mcp]` section leaves the surface DARK and every lifetime at
990    /// its documented default. A config that never mentions MCP must not end up
991    /// serving one.
992    #[test]
993    fn missing_mcp_section_leaves_the_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
994        let config = ServerConfig::from_slice(
995            br"
996                [runtime]
997                query_timeout_ms = 10000
998
999                [websocket]
1000                event_broadcast_capacity = 64
1001                cluster_broadcast_capacity = 64
1002            ",
1003        )?;
1004        let (_store, runtime) = config.into_parts();
1005        assert!(!runtime.mcp.enabled);
1006        assert!(runtime.mcp.allowed_origins.is_empty());
1007        assert_eq!(
1008            runtime.mcp.discover_ttl_ms,
1009            crate::config::DEFAULT_MCP_DISCOVER_TTL_MS
1010        );
1011        assert_eq!(
1012            runtime.mcp.task_poll_interval_ms,
1013            crate::config::DEFAULT_MCP_TASK_POLL_INTERVAL_MS
1014        );
1015        Ok(())
1016    }
1017
1018    /// Explicit `[mcp]` values parse and round-trip.
1019    #[test]
1020    fn mcp_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
1021        let config = ServerConfig::from_slice(
1022            br#"
1023                [runtime]
1024                query_timeout_ms = 10000
1025
1026                [websocket]
1027                event_broadcast_capacity = 64
1028                cluster_broadcast_capacity = 64
1029
1030                [mcp]
1031                enabled = true
1032                allowed_origins = ["http://localhost:5173"]
1033                discover_ttl_ms = 1000
1034                tools_list_ttl_ms = 2000
1035                task_poll_interval_ms = 250
1036            "#,
1037        )?;
1038        let (_store, runtime) = config.into_parts();
1039        assert!(runtime.mcp.enabled);
1040        assert_eq!(runtime.mcp.allowed_origins, vec!["http://localhost:5173"]);
1041        assert_eq!(runtime.mcp.discover_ttl_ms, 1_000);
1042        assert_eq!(runtime.mcp.tools_list_ttl_ms, 2_000);
1043        assert_eq!(runtime.mcp.task_poll_interval_ms, 250);
1044        Ok(())
1045    }
1046
1047    /// The retired MCP knobs are REFUSED, by name, rather than accepted and
1048    /// ignored.
1049    ///
1050    /// An MCP task is now a projection of a durable run: it holds no executor
1051    /// and expires never, so neither a task lifetime nor a server-side await
1052    /// interval has anything left to bound. An operator whose file still sets
1053    /// one is told so at boot — silently ignoring the key would leave them
1054    /// believing a lifetime was in force when nothing enforces one.
1055    #[test]
1056    fn the_retired_mcp_task_knobs_are_refused_by_name() {
1057        for key in ["task_ttl_ms", "await_poll_interval_ms"] {
1058            let source = format!(
1059                "
1060                [runtime]
1061                query_timeout_ms = 10000
1062
1063                [websocket]
1064                event_broadcast_capacity = 64
1065                cluster_broadcast_capacity = 64
1066
1067                [mcp]
1068                enabled = true
1069                {key} = 1000
1070            "
1071            );
1072            let message = match ServerConfig::from_slice(source.as_bytes()) {
1073                Err(crate::ServerError::Config { message }) => message,
1074                _ => String::new(),
1075            };
1076            assert!(
1077                message.contains(key),
1078                "{key} must be refused by name: {message}"
1079            );
1080        }
1081    }
1082
1083    /// A zero poll interval is a busy loop, not a fast one. It is refused at
1084    /// startup with an operator-facing message rather than accepted.
1085    #[test]
1086    fn a_zero_mcp_poll_interval_fails_startup_validation() {
1087        let result = ServerConfig::from_slice(
1088            br"
1089                [runtime]
1090                query_timeout_ms = 10000
1091
1092                [websocket]
1093                event_broadcast_capacity = 64
1094                cluster_broadcast_capacity = 64
1095
1096                [mcp]
1097                enabled = true
1098                task_poll_interval_ms = 0
1099            ",
1100        );
1101        let message = match result {
1102            Err(crate::ServerError::Config { message }) => message,
1103            _ => String::new(),
1104        };
1105        assert!(
1106            message.contains("task_poll_interval_ms"),
1107            "a zero poll interval must be refused: {message}"
1108        );
1109    }
1110
1111    /// The same zero on a DARK surface is not a boot failure: nothing reads it,
1112    /// so refusing the boot would be refusing over a value that has no effect.
1113    #[test]
1114    fn a_dark_mcp_surface_does_not_validate_its_unused_knobs()
1115    -> Result<(), Box<dyn std::error::Error>> {
1116        let config = ServerConfig::from_slice(
1117            br"
1118                [runtime]
1119                query_timeout_ms = 10000
1120
1121                [websocket]
1122                event_broadcast_capacity = 64
1123                cluster_broadcast_capacity = 64
1124
1125                [mcp]
1126                task_poll_interval_ms = 0
1127            ",
1128        )?;
1129        let (_store, runtime) = config.into_parts();
1130        assert!(!runtime.mcp.enabled);
1131        assert_eq!(
1132            runtime.mcp.task_poll_interval_ms,
1133            crate::config::DEFAULT_MCP_TASK_POLL_INTERVAL_MS,
1134            "an unused zero resolves to the default rather than to a busy loop"
1135        );
1136        Ok(())
1137    }
1138
1139    /// The flush policy parses and round-trips into the runtime view. Both keys
1140    /// are `Option` with NO default, so what a config states is exactly what the
1141    /// boot path sees — including a hold of zero, which is a real setting
1142    /// ("never wait"), not an absence.
1143    #[test]
1144    fn observability_flush_policy_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>>
1145    {
1146        let config = ServerConfig::from_slice(
1147            br"
1148                [runtime]
1149                query_timeout_ms = 10000
1150
1151                [websocket]
1152                event_broadcast_capacity = 64
1153                cluster_broadcast_capacity = 64
1154
1155                [observability]
1156                max_batch_events = 32
1157                max_batch_hold_ms = 0
1158            ",
1159        )?;
1160        assert_eq!(config.observability.max_batch_events, Some(32));
1161        assert_eq!(config.observability.max_batch_hold_ms, Some(0));
1162        let (_store, runtime) = config.into_parts();
1163        assert_eq!(runtime.observability.max_batch_events, Some(32));
1164        assert_eq!(
1165            runtime.observability.max_batch_hold_ms,
1166            Some(0),
1167            "a stated zero hold survives as a stated zero, never collapsing to absent"
1168        );
1169        Ok(())
1170    }
1171
1172    /// The node cache budget parses in BOTH of haematite's spellings and
1173    /// round-trips into the store view unchanged. There is no default, so what
1174    /// a config states is exactly what the boot path sees.
1175    #[test]
1176    fn node_cache_budget_parses_both_spellings() -> Result<(), Box<dyn std::error::Error>> {
1177        let bounded = ServerConfig::from_slice(
1178            br"
1179                [runtime]
1180                query_timeout_ms = 10000
1181
1182                [store]
1183                node_cache_budget = { bytes = 1073741824 }
1184            ",
1185        )?;
1186        assert_eq!(
1187            bounded.store.node_cache_budget,
1188            Some(haematite::NodeCacheBudget::bytes(1 << 30)?),
1189            "a 1 GiB ceiling arrives as a 1 GiB ceiling"
1190        );
1191
1192        let unlimited = ServerConfig::from_slice(
1193            br#"
1194                [runtime]
1195                query_timeout_ms = 10000
1196
1197                [store]
1198                node_cache_budget = "unlimited"
1199            "#,
1200        )?;
1201        assert_eq!(
1202            unlimited.store.node_cache_budget,
1203            Some(haematite::NodeCacheBudget::Unlimited),
1204            "`unlimited` is a stated choice, never collapsed to absent"
1205        );
1206        Ok(())
1207    }
1208
1209    /// The RETIRED lock-acquisition keys stay LEGAL: a configuration still
1210    /// carrying them — the 0.22.0-era estate shape, including the bridge
1211    /// values written during the 2026-08-24 incident — parses and loads.
1212    /// They are warned and ignored at the haematite connect seam, never
1213    /// refused, so no release turns an existing estate's config into a boot
1214    /// failure over keys the server no longer reads.
1215    #[test]
1216    fn retired_lock_acquisition_keys_stay_legal() -> Result<(), Box<dyn std::error::Error>> {
1217        let config = ServerConfig::from_slice(
1218            br#"
1219                [store]
1220                backend = "haematite"
1221                lock_acquisition_patience_ms = 60000
1222                lock_acquisition_retry_cadence_ms = 500
1223            "#,
1224        )?;
1225        assert_eq!(
1226            config.store.lock_acquisition_patience_ms,
1227            Some(60_000),
1228            "the retired key parses instead of refusing as unknown"
1229        );
1230        assert_eq!(
1231            config.store.lock_acquisition_retry_cadence_ms,
1232            Some(500),
1233            "both retired keys parse; nothing consumes them"
1234        );
1235        Ok(())
1236    }
1237
1238    /// An omitted budget parses to `None` rather than to a guess. The refusal
1239    /// happens where the value is USED (the haematite boot path), which
1240    /// `state.rs`'s tests pin; here the point is that nothing invents one on the
1241    /// way through — and that `unlimited` is distinguishable from absent.
1242    #[test]
1243    fn an_omitted_node_cache_budget_stays_absent() -> Result<(), Box<dyn std::error::Error>> {
1244        let config = ServerConfig::from_slice(
1245            br"
1246                [runtime]
1247                query_timeout_ms = 10000
1248
1249                [store]
1250                shard_count = 8
1251            ",
1252        )?;
1253        assert_eq!(config.store.node_cache_budget, None);
1254        Ok(())
1255    }
1256
1257    /// An omitted flush policy parses to `None` rather than to a guess. The
1258    /// refusal happens where the value is used (the publisher build), which
1259    /// `state.rs`'s tests pin; here the point is that nothing invents a value on
1260    /// the way through.
1261    #[test]
1262    fn an_omitted_flush_policy_stays_absent() -> Result<(), Box<dyn std::error::Error>> {
1263        let config = ServerConfig::from_slice(
1264            br"
1265                [runtime]
1266                query_timeout_ms = 10000
1267
1268                [observability]
1269                max_event_bytes = 512
1270            ",
1271        )?;
1272        assert_eq!(config.observability.max_batch_events, None);
1273        assert_eq!(config.observability.max_batch_hold_ms, None);
1274        Ok(())
1275    }
1276
1277    #[test]
1278    fn zero_observability_max_event_bytes_fails_startup_validation() {
1279        let result = ServerConfig::from_slice(
1280            br"
1281                [runtime]
1282                query_timeout_ms = 10000
1283
1284                [websocket]
1285                event_broadcast_capacity = 64
1286                cluster_broadcast_capacity = 64
1287
1288                [observability]
1289                max_event_bytes = 0
1290            ",
1291        );
1292        let message = result
1293            .err()
1294            .map_or_else(String::new, |error| error.to_string());
1295        assert!(
1296            message.contains("observability.max_event_bytes"),
1297            "validation message must name the zero-valued key: {message}"
1298        );
1299    }
1300
1301    #[test]
1302    fn zero_observability_max_stream_events_fails_startup_validation() {
1303        let result = ServerConfig::from_slice(
1304            br"
1305                [runtime]
1306                query_timeout_ms = 10000
1307
1308                [websocket]
1309                event_broadcast_capacity = 64
1310                cluster_broadcast_capacity = 64
1311
1312                [observability]
1313                max_stream_events = 0
1314            ",
1315        );
1316        let message = result
1317            .err()
1318            .map_or_else(String::new, |error| error.to_string());
1319        assert!(
1320            message.contains("observability.max_stream_events"),
1321            "validation message must name the zero-valued key: {message}"
1322        );
1323    }
1324
1325    #[test]
1326    fn missing_query_timeout_uses_default() -> Result<(), Box<dyn std::error::Error>> {
1327        // The server unconditionally mounts /workflows/query, but the reply
1328        // deadline is a tuning knob: omitting it must resolve to the default and
1329        // boot, not fail startup.
1330        let config = ServerConfig::from_slice(
1331            br"
1332                [runtime]
1333                scheduler_threads = 1
1334
1335                [websocket]
1336                event_broadcast_capacity = 64
1337                cluster_broadcast_capacity = 64
1338            ",
1339        )?;
1340        assert_eq!(
1341            config.runtime.query_timeout_ms,
1342            Some(DEFAULT_QUERY_TIMEOUT_MS),
1343            "omitted query_timeout_ms must resolve to the default"
1344        );
1345        Ok(())
1346    }
1347
1348    #[test]
1349    fn empty_config_boots_on_operational_defaults() -> Result<(), Box<dyn std::error::Error>> {
1350        // The headline zero-config contract: an empty TOML must parse, fill every
1351        // operational tuning knob with its default, and validate — so `aion
1352        // server` runs with no hand-authored file. The durable default backend
1353        // (haematite under its default data_dir) carries the store side.
1354        let config = ServerConfig::from_slice(b"")?;
1355        assert_eq!(config.store.backend, StoreBackend::Haematite);
1356        assert_eq!(
1357            config.runtime.query_timeout_ms,
1358            Some(DEFAULT_QUERY_TIMEOUT_MS)
1359        );
1360        assert_eq!(
1361            config.websocket.event_broadcast_capacity,
1362            Some(DEFAULT_EVENT_BROADCAST_CAPACITY)
1363        );
1364        assert_eq!(
1365            config.websocket.cluster_broadcast_capacity,
1366            Some(DEFAULT_CLUSTER_BROADCAST_CAPACITY)
1367        );
1368        Ok(())
1369    }
1370
1371    #[test]
1372    fn zero_query_timeout_fails_startup_validation() {
1373        let result = ServerConfig::from_slice(
1374            br"
1375                [runtime]
1376                query_timeout_ms = 0
1377
1378                [websocket]
1379                event_broadcast_capacity = 64
1380                cluster_broadcast_capacity = 64
1381            ",
1382        );
1383
1384        let message = result
1385            .err()
1386            .map_or_else(String::new, |error| error.to_string());
1387        assert!(
1388            message.contains("runtime.query_timeout_ms"),
1389            "validation message must name the zero-valued key: {message}"
1390        );
1391    }
1392
1393    /// The deploy surface is commissioned explicitly: enabling it without
1394    /// the archive ceiling is a conservative security default, not a forced
1395    /// operator decision: enabling deploy without it must resolve the ceiling
1396    /// to [`DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES`] and boot, not fail startup.
1397    #[test]
1398    fn deploy_enabled_defaults_max_archive_bytes() -> Result<(), Box<dyn std::error::Error>> {
1399        let config = ServerConfig::from_slice(
1400            br"
1401                [runtime]
1402                query_timeout_ms = 10000
1403
1404                [websocket]
1405                event_broadcast_capacity = 64
1406                cluster_broadcast_capacity = 64
1407
1408                [deploy]
1409                enabled = true
1410            ",
1411        )?;
1412
1413        assert_eq!(
1414            config.deploy.max_archive_bytes,
1415            Some(DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES),
1416            "omitted max_archive_bytes must resolve to the conservative default"
1417        );
1418        assert_eq!(
1419            config.deploy.max_inflated_bytes,
1420            Some(DEFAULT_DEPLOY_MAX_INFLATED_BYTES),
1421            "omitted max_inflated_bytes must resolve to the conservative default"
1422        );
1423        Ok(())
1424    }
1425
1426    #[test]
1427    fn deploy_zero_max_archive_bytes_fails_startup_validation() {
1428        let result = ServerConfig::from_slice(
1429            br"
1430                [runtime]
1431                query_timeout_ms = 10000
1432
1433                [websocket]
1434                event_broadcast_capacity = 64
1435                cluster_broadcast_capacity = 64
1436
1437                [deploy]
1438                enabled = true
1439                max_archive_bytes = 0
1440            ",
1441        );
1442
1443        let message = result
1444            .err()
1445            .map_or_else(String::new, |error| error.to_string());
1446        assert!(
1447            message.contains("deploy.max_archive_bytes"),
1448            "validation message must name the zero-valued key: {message}"
1449        );
1450    }
1451
1452    /// The inflate ceiling defaults independently of an explicit archive
1453    /// ceiling: setting only `max_archive_bytes` must resolve the inflate
1454    /// ceiling to [`DEFAULT_DEPLOY_MAX_INFLATED_BYTES`] (which exceeds a 16 MiB
1455    /// archive, so the invariant holds) and boot.
1456    #[test]
1457    fn deploy_enabled_defaults_max_inflated_bytes() -> Result<(), Box<dyn std::error::Error>> {
1458        let config = ServerConfig::from_slice(
1459            br"
1460                [runtime]
1461                query_timeout_ms = 10000
1462
1463                [websocket]
1464                event_broadcast_capacity = 64
1465                cluster_broadcast_capacity = 64
1466
1467                [deploy]
1468                enabled = true
1469                max_archive_bytes = 16777216
1470            ",
1471        )?;
1472
1473        assert_eq!(
1474            config.deploy.max_archive_bytes,
1475            Some(16_777_216),
1476            "explicit max_archive_bytes must be left untouched"
1477        );
1478        assert_eq!(
1479            config.deploy.max_inflated_bytes,
1480            Some(DEFAULT_DEPLOY_MAX_INFLATED_BYTES),
1481            "omitted max_inflated_bytes must resolve to the conservative default"
1482        );
1483        Ok(())
1484    }
1485
1486    #[test]
1487    fn deploy_zero_max_inflated_bytes_fails_startup_validation() {
1488        let result = ServerConfig::from_slice(
1489            br"
1490                [runtime]
1491                query_timeout_ms = 10000
1492
1493                [websocket]
1494                event_broadcast_capacity = 64
1495                cluster_broadcast_capacity = 64
1496
1497                [deploy]
1498                enabled = true
1499                max_archive_bytes = 16777216
1500                max_inflated_bytes = 0
1501            ",
1502        );
1503
1504        let message = result
1505            .err()
1506            .map_or_else(String::new, |error| error.to_string());
1507        assert!(
1508            message.contains("deploy.max_inflated_bytes"),
1509            "validation message must name the zero-valued key: {message}"
1510        );
1511    }
1512
1513    /// An inflate ceiling below the upload ceiling is incoherent: archives
1514    /// the upload ceiling admits would be refused even stored uncompressed.
1515    #[test]
1516    fn deploy_max_inflated_below_max_archive_fails_startup_validation() {
1517        let result = ServerConfig::from_slice(
1518            br"
1519                [runtime]
1520                query_timeout_ms = 10000
1521
1522                [websocket]
1523                event_broadcast_capacity = 64
1524                cluster_broadcast_capacity = 64
1525
1526                [deploy]
1527                enabled = true
1528                max_archive_bytes = 16777216
1529                max_inflated_bytes = 16777215
1530            ",
1531        );
1532
1533        let message = result
1534            .err()
1535            .map_or_else(String::new, |error| error.to_string());
1536        assert!(
1537            message.contains("deploy.max_inflated_bytes")
1538                && message.contains("deploy.max_archive_bytes"),
1539            "validation message must name both ceilings: {message}"
1540        );
1541    }
1542
1543    /// An absent `[deploy]` section means the surface stays dark and the
1544    /// ceilings are not required.
1545    #[test]
1546    fn deploy_disabled_requires_no_archive_ceiling() -> Result<(), Box<dyn std::error::Error>> {
1547        let config = ServerConfig::from_slice(
1548            br"
1549                [runtime]
1550                query_timeout_ms = 10000
1551
1552                [websocket]
1553                event_broadcast_capacity = 64
1554                cluster_broadcast_capacity = 64
1555            ",
1556        )?;
1557
1558        assert!(!config.deploy.enabled);
1559        assert_eq!(config.deploy.max_archive_bytes, None);
1560        assert_eq!(config.deploy.max_inflated_bytes, None);
1561        Ok(())
1562    }
1563
1564    #[test]
1565    fn deploy_section_parses_enabled_with_ceilings() -> Result<(), Box<dyn std::error::Error>> {
1566        let config = ServerConfig::from_slice(
1567            br"
1568                [runtime]
1569                query_timeout_ms = 10000
1570
1571                [websocket]
1572                event_broadcast_capacity = 64
1573                cluster_broadcast_capacity = 64
1574
1575                [deploy]
1576                enabled = true
1577                max_archive_bytes = 16777216
1578                max_inflated_bytes = 67108864
1579            ",
1580        )?;
1581
1582        assert!(config.deploy.enabled);
1583        assert_eq!(config.deploy.max_archive_bytes, Some(16_777_216));
1584        assert_eq!(config.deploy.max_inflated_bytes, Some(67_108_864));
1585        Ok(())
1586    }
1587
1588    /// With no `[server] cors_allowed_origins` the list is empty: the secure
1589    /// default, where no cross-origin request is permitted and no `CorsLayer`
1590    /// is installed.
1591    #[test]
1592    fn cors_allowed_origins_default_empty() -> Result<(), Box<dyn std::error::Error>> {
1593        let config = ServerConfig::from_slice(
1594            br"
1595                [runtime]
1596                query_timeout_ms = 10000
1597
1598                [websocket]
1599                event_broadcast_capacity = 64
1600                cluster_broadcast_capacity = 64
1601            ",
1602        )?;
1603
1604        assert!(config.server.cors_allowed_origins.is_empty());
1605        let (_, runtime) = config.into_parts();
1606        assert!(runtime.cors_allowed_origins.is_empty());
1607        Ok(())
1608    }
1609
1610    /// A configured `[server] cors_allowed_origins` list parses and round-trips
1611    /// into `RuntimeConfig` (the value the `CorsLayer` is built from).
1612    #[test]
1613    fn cors_allowed_origins_parse_and_round_trip() -> Result<(), Box<dyn std::error::Error>> {
1614        let config = ServerConfig::from_slice(
1615            br#"
1616                [server]
1617                cors_allowed_origins = ["http://localhost:5173", "http://127.0.0.1:5173"]
1618
1619                [runtime]
1620                query_timeout_ms = 10000
1621
1622                [websocket]
1623                event_broadcast_capacity = 64
1624                cluster_broadcast_capacity = 64
1625            "#,
1626        )?;
1627
1628        assert_eq!(
1629            config.server.cors_allowed_origins,
1630            vec![
1631                "http://localhost:5173".to_owned(),
1632                "http://127.0.0.1:5173".to_owned()
1633            ]
1634        );
1635        let (_, runtime) = config.into_parts();
1636        assert_eq!(
1637            runtime.cors_allowed_origins,
1638            vec![
1639                "http://localhost:5173".to_owned(),
1640                "http://127.0.0.1:5173".to_owned()
1641            ]
1642        );
1643        Ok(())
1644    }
1645
1646    /// A malformed CORS origin (no scheme, or a trailing path) can never match a
1647    /// browser `Origin` header, so it fails startup validation rather than
1648    /// silently never matching.
1649    #[test]
1650    fn cors_allowed_origins_reject_malformed() {
1651        for bad in ["", "localhost:5173", "http://localhost:5173/"] {
1652            let toml = format!(
1653                "[server]\ncors_allowed_origins = [\"{bad}\"]\n\n[runtime]\nquery_timeout_ms = 10000\n\n[websocket]\nevent_broadcast_capacity = 64\n"
1654            );
1655            let result = ServerConfig::from_slice(toml.as_bytes());
1656            let message = result
1657                .err()
1658                .map_or_else(String::new, |error| error.to_string());
1659            assert!(
1660                message.contains("cors_allowed_origins"),
1661                "malformed origin `{bad}` must be rejected naming the key: {message}"
1662            );
1663        }
1664    }
1665
1666    /// An absent `[dev]` section leaves the dev surface dark.
1667    #[test]
1668    fn dev_absent_leaves_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
1669        let config = ServerConfig::from_slice(
1670            br"
1671                [runtime]
1672                query_timeout_ms = 10000
1673
1674                [websocket]
1675                event_broadcast_capacity = 64
1676                cluster_broadcast_capacity = 64
1677            ",
1678        )?;
1679
1680        assert!(!config.dev.enabled);
1681        Ok(())
1682    }
1683
1684    /// `[dev] enabled = true` commissions the dev surface; it adds no other
1685    /// knobs (ADR-001: the only setting is the on/off gate).
1686    #[test]
1687    fn dev_section_parses_enabled() -> Result<(), Box<dyn std::error::Error>> {
1688        let config = ServerConfig::from_slice(
1689            br"
1690                [runtime]
1691                query_timeout_ms = 10000
1692
1693                [websocket]
1694                event_broadcast_capacity = 64
1695                cluster_broadcast_capacity = 64
1696
1697                [dev]
1698                enabled = true
1699            ",
1700        )?;
1701
1702        assert!(config.dev.enabled);
1703        Ok(())
1704    }
1705
1706    /// An absent `[authoring]` section commissions the AWL workspace while the
1707    /// separate Gleam loop remains dark and requires neither of its paths.
1708    #[test]
1709    fn authoring_absent_defaults_awl_workspace_but_keeps_gleam_dark()
1710    -> Result<(), Box<dyn std::error::Error>> {
1711        let home = crate::test_support::private_tempdir()?;
1712        let config = ServerConfig::from_slice_with_home(b"", home.path())?;
1713
1714        assert_eq!(config.authoring.gleam_path, None);
1715        assert_eq!(config.authoring.project_root, None);
1716        assert_eq!(
1717            config.authoring.workspace_dir.as_deref(),
1718            Some(home.path().join(DEFAULT_AUTHORING_WORKSPACE_DIR).as_path())
1719        );
1720        Ok(())
1721    }
1722
1723    /// An explicit `workspace_dir` wins over the out-of-box relative default.
1724    #[test]
1725    fn authoring_explicit_workspace_is_honored() -> Result<(), Box<dyn std::error::Error>> {
1726        let config = ServerConfig::from_slice(
1727            br#"
1728                [authoring]
1729                workspace_dir = "/srv/aion/studio"
1730            "#,
1731        )?;
1732
1733        assert_eq!(
1734            config.authoring.workspace_dir.as_deref(),
1735            Some(std::path::Path::new("/srv/aion/studio"))
1736        );
1737        Ok(())
1738    }
1739
1740    /// A configured `[authoring]` section with both `gleam_path` and
1741    /// `project_root` parses and round-trips into `RuntimeConfig`.
1742    #[test]
1743    fn authoring_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
1744        let config = ServerConfig::from_slice(
1745            br#"
1746                [runtime]
1747                query_timeout_ms = 10000
1748
1749                [websocket]
1750                event_broadcast_capacity = 64
1751                cluster_broadcast_capacity = 64
1752
1753                [authoring]
1754                gleam_path = "/usr/local/bin/gleam"
1755                project_root = "/srv/aion/authoring"
1756            "#,
1757        )?;
1758
1759        assert_eq!(
1760            config.authoring.gleam_path.as_deref(),
1761            Some(std::path::Path::new("/usr/local/bin/gleam"))
1762        );
1763        let (_, runtime) = config.into_parts();
1764        assert_eq!(
1765            runtime.authoring.gleam_path.as_deref(),
1766            Some(std::path::Path::new("/usr/local/bin/gleam"))
1767        );
1768        assert_eq!(
1769            runtime.authoring.project_root.as_deref(),
1770            Some(std::path::Path::new("/srv/aion/authoring"))
1771        );
1772        Ok(())
1773    }
1774
1775    /// Commissioning the authoring loop (a `gleam_path`) without a
1776    /// `project_root` must fail startup naming the key and the environment
1777    /// override (the deploy required-config pattern).
1778    #[test]
1779    fn authoring_gleam_path_without_project_root_fails_naming_key_and_env() {
1780        let result = ServerConfig::from_slice(
1781            br#"
1782                [runtime]
1783                query_timeout_ms = 10000
1784
1785                [websocket]
1786                event_broadcast_capacity = 64
1787                cluster_broadcast_capacity = 64
1788
1789                [authoring]
1790                gleam_path = "/usr/local/bin/gleam"
1791            "#,
1792        );
1793
1794        let message = result
1795            .err()
1796            .map_or_else(String::new, |error| error.to_string());
1797        assert!(
1798            message.contains("authoring.project_root"),
1799            "validation message must name the missing key: {message}"
1800        );
1801        assert!(
1802            message.contains("AION_AUTHORING_PROJECT_ROOT"),
1803            "validation message must name the environment override: {message}"
1804        );
1805    }
1806
1807    /// An empty `gleam_path` is a misconfiguration, not "dark": it must fail
1808    /// startup naming the key and the environment override.
1809    #[test]
1810    fn authoring_empty_gleam_path_fails_naming_key_and_env() {
1811        let result = ServerConfig::from_slice(
1812            br#"
1813                [runtime]
1814                query_timeout_ms = 10000
1815
1816                [websocket]
1817                event_broadcast_capacity = 64
1818                cluster_broadcast_capacity = 64
1819
1820                [authoring]
1821                gleam_path = ""
1822            "#,
1823        );
1824
1825        let message = result
1826            .err()
1827            .map_or_else(String::new, |error| error.to_string());
1828        assert!(
1829            message.contains("authoring.gleam_path"),
1830            "validation message must name the empty key: {message}"
1831        );
1832        assert!(
1833            message.contains("AION_AUTHORING_GLEAM_PATH"),
1834            "validation message must name the environment override: {message}"
1835        );
1836    }
1837
1838    /// CLI overrides commission the authoring loop after file/env merge.
1839    #[test]
1840    fn cli_overrides_set_authoring_paths() -> Result<(), Box<dyn std::error::Error>> {
1841        let mut config = ServerConfig::from_slice(
1842            br"
1843                [runtime]
1844                query_timeout_ms = 10000
1845
1846                [websocket]
1847                event_broadcast_capacity = 64
1848                cluster_broadcast_capacity = 64
1849            ",
1850        )?;
1851        let cli = CliOverrides {
1852            gleam_path: Some(std::path::PathBuf::from("/opt/gleam")),
1853            authoring_project_root: Some(std::path::PathBuf::from("/opt/project")),
1854            ..CliOverrides::default()
1855        };
1856
1857        config.apply_cli_overrides(&cli);
1858        config.validate()?;
1859
1860        assert_eq!(
1861            config.authoring.gleam_path.as_deref(),
1862            Some(std::path::Path::new("/opt/gleam"))
1863        );
1864        assert_eq!(
1865            config.authoring.project_root.as_deref(),
1866            Some(std::path::Path::new("/opt/project"))
1867        );
1868        Ok(())
1869    }
1870
1871    /// The config field renamed `dashboard` -> `ops_console` carries a serde
1872    /// alias so existing `[dashboard]` TOML still parses (non-breaking rename).
1873    /// R1: the written queue-service settings reach `RuntimeConfig`, and an
1874    /// operator who writes nothing gets `strict` with no clocks.
1875    #[test]
1876    fn queue_service_settings_are_read_from_the_worker_section()
1877    -> Result<(), Box<dyn std::error::Error>> {
1878        use crate::worker::QueueServicePolicy;
1879        use std::time::Duration;
1880
1881        let bare = ServerConfig::from_slice(
1882            br"
1883                [websocket]
1884                event_broadcast_capacity = 64
1885                cluster_broadcast_capacity = 64
1886            ",
1887        )?;
1888        assert_eq!(
1889            bare.worker.queue_service.default_policy,
1890            QueueServicePolicy::Strict,
1891            "strict is the default with nothing written"
1892        );
1893        assert_eq!(
1894            bare.worker.queue_service.service_availability_deadline,
1895            None
1896        );
1897        assert_eq!(bare.worker.queue_service.schedule_to_start_timeout, None);
1898
1899        let written = ServerConfig::from_slice(
1900            br#"
1901                [websocket]
1902                event_broadcast_capacity = 64
1903                cluster_broadcast_capacity = 64
1904
1905                [worker.queue_service]
1906                service_availability_deadline = 45000
1907                schedule_to_start_timeout = 5000
1908
1909                [[worker.queue_service.overrides]]
1910                task_queue = "general"
1911                policy = "durable_pending"
1912            "#,
1913        )?;
1914        assert_eq!(
1915            written.worker.queue_service.service_availability_deadline,
1916            Some(Duration::from_secs(45))
1917        );
1918        assert_eq!(
1919            written.worker.queue_service.schedule_to_start_timeout,
1920            Some(Duration::from_secs(5))
1921        );
1922        assert_eq!(
1923            written
1924                .worker
1925                .queue_service
1926                .policy_for("default", "general"),
1927            QueueServicePolicy::DurablePending,
1928            "the written opt-in must reach the dispatch seam"
1929        );
1930        assert_eq!(
1931            written
1932                .worker
1933                .queue_service
1934                .policy_for("default", "billing"),
1935            QueueServicePolicy::Strict,
1936            "an override must not leak onto other queues"
1937        );
1938
1939        // The same settings survive the runtime projection the server boots from.
1940        let (_store, runtime) = written.into_parts();
1941        assert_eq!(
1942            runtime
1943                .worker
1944                .queue_service
1945                .policy_for("default", "general"),
1946            QueueServicePolicy::DurablePending
1947        );
1948        Ok(())
1949    }
1950
1951    #[test]
1952    fn legacy_dashboard_section_alias_still_parses() -> Result<(), Box<dyn std::error::Error>> {
1953        let config = ServerConfig::from_slice(
1954            br#"
1955                [runtime]
1956                query_timeout_ms = 10000
1957
1958                [websocket]
1959                event_broadcast_capacity = 64
1960                cluster_broadcast_capacity = 64
1961
1962                [dashboard]
1963                source = { FileSystem = { asset_path = "/srv/aion/ui" } }
1964            "#,
1965        )?;
1966        match &config.ops_console.source {
1967            OpsConsoleAssetSource::FileSystem { asset_path } => {
1968                assert_eq!(asset_path.as_os_str(), "/srv/aion/ui");
1969            }
1970            OpsConsoleAssetSource::Embedded => {
1971                return Err("legacy [dashboard] section must map to ops_console".into());
1972            }
1973        }
1974        Ok(())
1975    }
1976
1977    /// The new `[ops_console]` section name also parses.
1978    #[test]
1979    fn ops_console_section_parses() -> Result<(), Box<dyn std::error::Error>> {
1980        let config = ServerConfig::from_slice(
1981            br#"
1982                [runtime]
1983                query_timeout_ms = 10000
1984
1985                [websocket]
1986                event_broadcast_capacity = 64
1987                cluster_broadcast_capacity = 64
1988
1989                [ops_console]
1990                source = { FileSystem = { asset_path = "/srv/aion/ui" } }
1991            "#,
1992        )?;
1993        assert!(matches!(
1994            config.ops_console.source,
1995            OpsConsoleAssetSource::FileSystem { .. }
1996        ));
1997        Ok(())
1998    }
1999
2000    #[test]
2001    fn invalid_values_name_problematic_field() {
2002        let result = ServerConfig::from_slice(
2003            br"
2004                [runtime]
2005                scheduler_threads = 0
2006            ",
2007        );
2008
2009        let message = result
2010            .err()
2011            .map_or_else(String::new, |error| error.to_string());
2012        assert!(message.contains("runtime.scheduler_threads"));
2013    }
2014
2015    const RETIRED_STORE_REMEDY: &str = "backend = \"haematite\" with store.data_dir and store.node_cache_budget (all three are required — haematite refuses to start without the budget); existing libsql database files are not read by this build";
2016
2017    fn assert_retired_store_refusal(
2018        result: Result<ServerConfig, crate::error::ServerError>,
2019        found: &str,
2020    ) {
2021        assert!(result.is_err(), "retired libsql input must be refused");
2022        let message = result
2023            .err()
2024            .map_or_else(String::new, |error| error.to_string());
2025        assert!(
2026            message.contains(found),
2027            "refusal did not name `{found}`: {message}"
2028        );
2029        assert!(
2030            message.contains(RETIRED_STORE_REMEDY),
2031            "refusal omitted the operator remedy: {message}"
2032        );
2033    }
2034
2035    #[test]
2036    fn retired_libsql_backend_is_refused_with_remedy() {
2037        assert_retired_store_refusal(
2038            ServerConfig::from_slice(
2039                br#"
2040                    [store]
2041                    backend = "libsql"
2042                "#,
2043            ),
2044            "backend = \"libsql\"",
2045        );
2046    }
2047
2048    #[test]
2049    fn retired_store_url_key_is_refused_with_remedy() {
2050        assert_retired_store_refusal(
2051            ServerConfig::from_slice(
2052                br#"
2053                    [store]
2054                    backend = "haematite"
2055                    url = "old.db"
2056                "#,
2057            ),
2058            "store.url",
2059        );
2060    }
2061
2062    /// The FIFTH door: `AION_STORE_BACKEND=libsql` names the retired backend as
2063    /// squarely as `backend = "libsql"` in the file does, and must reach the same
2064    /// named refusal rather than the parser's "must be one of" list, which names
2065    /// no remedy and reads like a typo.
2066    #[test]
2067    fn retired_libsql_backend_environment_is_refused_with_remedy() {
2068        let mut config = ServerConfig::default();
2069        let result = super::env::overlay_vars(
2070            &mut config,
2071            [("AION_STORE_BACKEND".to_owned(), "libsql".to_owned())],
2072        )
2073        .and_then(|()| config.validate().map(|()| config));
2074        assert_retired_store_refusal(result, "AION_STORE_BACKEND=libsql");
2075    }
2076
2077    #[test]
2078    fn retired_store_url_environment_is_refused_with_remedy() {
2079        let mut config = ServerConfig::default();
2080        let result = super::env::overlay_vars(
2081            &mut config,
2082            [("AION_STORE_URL".to_owned(), "old.db".to_owned())],
2083        )
2084        .and_then(|()| config.validate().map(|()| config));
2085        assert_retired_store_refusal(result, "AION_STORE_URL");
2086    }
2087
2088    #[test]
2089    fn retired_store_url_flag_is_refused_with_remedy() {
2090        let mut config = ServerConfig::default();
2091        config.apply_cli_overrides(&CliOverrides {
2092            store_url: Some("old.db".to_owned()),
2093            ..CliOverrides::default()
2094        });
2095        let result = config.validate().map(|()| config);
2096        assert_retired_store_refusal(result, "--store-url");
2097    }
2098
2099    /// A CLI flag beats the value the config file loaded. Only the `--store-url`
2100    /// arm of the original pin died with libSQL; the precedence claim itself is
2101    /// backend-agnostic and is the reason `apply_cli_overrides` exists.
2102    #[test]
2103    fn cli_overrides_win_over_loaded_values() -> Result<(), Box<dyn std::error::Error>> {
2104        let mut config = ServerConfig::from_slice(
2105            br#"
2106                [store]
2107                backend = "haematite"
2108                data_dir = "from-the-file"
2109
2110                [runtime]
2111                scheduler_threads = 1
2112                query_timeout_ms = 10000
2113
2114                [websocket]
2115                event_broadcast_capacity = 64
2116                cluster_broadcast_capacity = 64
2117
2118                [observability]
2119                max_batch_events = 64
2120                max_batch_hold_ms = 0
2121            "#,
2122        )?;
2123        assert_eq!(
2124            config.runtime.scheduler_threads, 1,
2125            "the file's value must be what the flag then beats"
2126        );
2127        let cli = CliOverrides {
2128            scheduler_threads: Some(
2129                std::num::NonZeroUsize::new(3)
2130                    .ok_or("3 is not zero")?
2131                    .into(),
2132            ),
2133            ..CliOverrides::default()
2134        };
2135
2136        config.apply_cli_overrides(&cli);
2137        config.validate()?;
2138
2139        assert_eq!(config.runtime.scheduler_threads, 3);
2140        assert_eq!(
2141            config.store.data_dir.as_deref(),
2142            Some("from-the-file"),
2143            "a value the CLI did not override keeps the file's value"
2144        );
2145        Ok(())
2146    }
2147
2148    #[test]
2149    fn haematite_store_config_remains_accepted() -> Result<(), Box<dyn std::error::Error>> {
2150        let config = ServerConfig::from_slice(
2151            br#"
2152                [store]
2153                backend = "haematite"
2154                data_dir = "aion-data"
2155            "#,
2156        )?;
2157        assert_eq!(config.store.backend, StoreBackend::Haematite);
2158        assert_eq!(config.store.data_dir.as_deref(), Some("aion-data"));
2159        Ok(())
2160    }
2161
2162    #[test]
2163    fn default_config_defaults() -> Result<(), Box<dyn std::error::Error>> {
2164        let mut config = ServerConfig::default();
2165
2166        // Raw serde defaults defer user-level paths until the merged loader has
2167        // resolved Aion home and applied every operator override.
2168        assert_eq!(config.store.backend, StoreBackend::Haematite);
2169        assert_eq!(config.store.data_dir, None);
2170        // 64 pending #187: 4096 defeated its own lazy-materialization premise
2171        // (boot scan_prefix materializes all shards; commit then fsyncs per
2172        // shard and blows the actor timeout). See StoreConfig::default().
2173        assert_eq!(config.store.shard_count, 64);
2174        assert_eq!(config.server.grpc_address.to_string(), "127.0.0.1:50051");
2175        assert_eq!(config.server.listen_address.to_string(), "127.0.0.1:8080");
2176        assert_eq!(config.namespaces.default, "default");
2177        // Minted-on-use is OPEN by default to preserve the zero-config,
2178        // no-pre-provision model: a namespace comes into being on first
2179        // worker reference.
2180        assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
2181        // The cluster-wide in-flight ceiling defaults to the generous platform
2182        // headroom value (P2-Q1); nothing enforces it yet.
2183        assert_eq!(
2184            config.namespaces.max_in_flight_activities,
2185            DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
2186        );
2187        assert_eq!(config.namespaces.max_in_flight_activities, 1024);
2188        assert!(!config.auth.enabled);
2189        assert!(config.metrics.enabled);
2190        // event_broadcast_capacity and query_timeout_ms are the deliberately
2191        // defaultless values: defaults validate only once the operator
2192        // supplies them.
2193        assert_eq!(config.websocket.event_broadcast_capacity, None);
2194        assert_eq!(config.websocket.cluster_broadcast_capacity, None);
2195        assert_eq!(config.runtime.query_timeout_ms, None);
2196        config.websocket.event_broadcast_capacity = Some(64);
2197        config.websocket.cluster_broadcast_capacity = Some(64);
2198        config.runtime.query_timeout_ms = Some(10_000);
2199        let home = crate::test_support::private_tempdir()?;
2200        let working_dir = crate::test_support::private_tempdir()?;
2201        super::fill_home_defaults(
2202            &mut config,
2203            home.path(),
2204            HomeSource::Derived,
2205            working_dir.path(),
2206        )?;
2207        assert_eq!(
2208            config.store.data_dir.as_deref(),
2209            home.path().join("data").to_str()
2210        );
2211        config.validate()?;
2212        Ok(())
2213    }
2214
2215    #[test]
2216    fn outbox_is_disabled_by_default_and_needs_no_knobs() -> Result<(), Box<dyn std::error::Error>>
2217    {
2218        let mut config = ServerConfig::default();
2219        config.store.data_dir = Some("test-data".to_owned());
2220        config.websocket.event_broadcast_capacity = Some(64);
2221        config.websocket.cluster_broadcast_capacity = Some(64);
2222        config.runtime.query_timeout_ms = Some(10_000);
2223
2224        // The dispatcher is dark by default and its operational knobs are all
2225        // absent — yet validation passes, because a disabled dispatcher never
2226        // reads them (no assumed defaults behind the gate).
2227        assert!(!config.outbox.enabled);
2228        assert_eq!(config.outbox.poll_interval_ms, None);
2229        assert_eq!(config.outbox.batch_size, None);
2230        assert_eq!(config.outbox.max_attempts, None);
2231        assert_eq!(config.outbox.backoff_base_ms, None);
2232        assert_eq!(config.outbox.backoff_multiplier, None);
2233        assert_eq!(config.outbox.backoff_max_ms, None);
2234        assert_eq!(config.outbox.reconcile_interval_ms, None);
2235        assert_eq!(config.outbox.reconcile_stale_after_ms, None);
2236        config.validate()?;
2237        Ok(())
2238    }
2239
2240    fn outbox_enabled_base() -> ServerConfig {
2241        let mut config = ServerConfig::default();
2242        config.store.data_dir = Some("test-data".to_owned());
2243        config.websocket.event_broadcast_capacity = Some(64);
2244        config.websocket.cluster_broadcast_capacity = Some(64);
2245        config.runtime.query_timeout_ms = Some(10_000);
2246        config.outbox.enabled = true;
2247        config.outbox.poll_interval_ms = Some(250);
2248        config.outbox.batch_size = Some(64);
2249        config.outbox.max_attempts = Some(5);
2250        config.outbox.backoff_base_ms = Some(100);
2251        config.outbox.backoff_multiplier = Some(2);
2252        config.outbox.backoff_max_ms = Some(30_000);
2253        config.outbox.reconcile_interval_ms = Some(1_000);
2254        config.outbox.reconcile_stale_after_ms = Some(60_000);
2255        config
2256    }
2257
2258    #[test]
2259    fn outbox_enabled_with_all_knobs_validates() -> Result<(), Box<dyn std::error::Error>> {
2260        outbox_enabled_base().validate()?;
2261        Ok(())
2262    }
2263
2264    #[test]
2265    fn outbox_enabled_defaults_poll_interval() -> Result<(), Box<dyn std::error::Error>> {
2266        // Enabling the dispatcher but omitting the poll cadence must resolve it
2267        // to the default and boot, not fail startup — the cadence is pure
2268        // tuning, not a forced operator decision.
2269        let config = ServerConfig::from_slice(
2270            br"
2271                [runtime]
2272                query_timeout_ms = 10000
2273
2274                [websocket]
2275                event_broadcast_capacity = 64
2276                cluster_broadcast_capacity = 64
2277
2278                [outbox]
2279                enabled = true
2280            ",
2281        )?;
2282        assert_eq!(
2283            config.outbox.poll_interval_ms,
2284            Some(DEFAULT_OUTBOX_POLL_INTERVAL_MS),
2285            "omitted poll_interval_ms must resolve to the default"
2286        );
2287        Ok(())
2288    }
2289
2290    #[test]
2291    fn outbox_enabled_defaults_max_attempts() -> Result<(), Box<dyn std::error::Error>> {
2292        // Setting a tuning knob explicitly but omitting the retry budget must
2293        // leave the explicit knob untouched and default only the omitted one.
2294        let config = ServerConfig::from_slice(
2295            br"
2296                [runtime]
2297                query_timeout_ms = 10000
2298
2299                [websocket]
2300                event_broadcast_capacity = 64
2301                cluster_broadcast_capacity = 64
2302
2303                [outbox]
2304                enabled = true
2305                poll_interval_ms = 250
2306            ",
2307        )?;
2308        assert_eq!(
2309            config.outbox.poll_interval_ms,
2310            Some(250),
2311            "explicit poll_interval_ms must be left untouched"
2312        );
2313        assert_eq!(
2314            config.outbox.max_attempts,
2315            Some(DEFAULT_OUTBOX_MAX_ATTEMPTS),
2316            "omitted max_attempts must resolve to the default"
2317        );
2318        Ok(())
2319    }
2320
2321    #[test]
2322    fn outbox_enabled_with_only_enabled_flag_uses_all_defaults()
2323    -> Result<(), Box<dyn std::error::Error>> {
2324        // Headline conditional-default contract: an outbox section with nothing
2325        // but `enabled = true` validates with every tuning knob resolved to its
2326        // default. The reconciliation pair stays dark (both absent), as before.
2327        let config = ServerConfig::from_slice(
2328            br"
2329                [runtime]
2330                query_timeout_ms = 10000
2331
2332                [websocket]
2333                event_broadcast_capacity = 64
2334                cluster_broadcast_capacity = 64
2335
2336                [outbox]
2337                enabled = true
2338            ",
2339        )?;
2340        assert!(config.outbox.enabled);
2341        assert_eq!(
2342            config.outbox.poll_interval_ms,
2343            Some(DEFAULT_OUTBOX_POLL_INTERVAL_MS)
2344        );
2345        assert_eq!(config.outbox.batch_size, Some(DEFAULT_OUTBOX_BATCH_SIZE));
2346        assert_eq!(
2347            config.outbox.max_attempts,
2348            Some(DEFAULT_OUTBOX_MAX_ATTEMPTS)
2349        );
2350        assert_eq!(
2351            config.outbox.backoff_base_ms,
2352            Some(DEFAULT_OUTBOX_BACKOFF_BASE_MS)
2353        );
2354        assert_eq!(
2355            config.outbox.backoff_multiplier,
2356            Some(DEFAULT_OUTBOX_BACKOFF_MULTIPLIER)
2357        );
2358        assert_eq!(
2359            config.outbox.backoff_max_ms,
2360            Some(DEFAULT_OUTBOX_BACKOFF_MAX_MS)
2361        );
2362        // Reconciliation is not force-defaulted: both knobs stay absent so the
2363        // live sweep remains dark.
2364        assert_eq!(config.outbox.reconcile_interval_ms, None);
2365        assert_eq!(config.outbox.reconcile_stale_after_ms, None);
2366        Ok(())
2367    }
2368
2369    #[test]
2370    fn outbox_enabled_zero_poll_interval_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
2371        // An explicit zero is a misconfiguration the default never masks:
2372        // `get_or_insert` leaves `Some(0)` untouched and validate rejects it.
2373        let mut config = outbox_enabled_base();
2374        config.outbox.poll_interval_ms = Some(0);
2375        let error = config
2376            .validate()
2377            .err()
2378            .ok_or("enabled outbox with zero poll interval must fail")?;
2379        assert!(
2380            error.to_string().contains("outbox.poll_interval_ms"),
2381            "error must name the zero-valued key: {error}"
2382        );
2383        Ok(())
2384    }
2385
2386    #[test]
2387    fn outbox_enabled_zero_max_attempts_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
2388        let mut config = outbox_enabled_base();
2389        config.outbox.max_attempts = Some(0);
2390        let error = config
2391            .validate()
2392            .err()
2393            .ok_or("enabled outbox with zero max attempts must fail")?;
2394        assert!(
2395            error.to_string().contains("outbox.max_attempts"),
2396            "error must name the zero-valued key: {error}"
2397        );
2398        Ok(())
2399    }
2400
2401    #[test]
2402    fn outbox_backoff_max_below_base_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
2403        let mut config = outbox_enabled_base();
2404        config.outbox.backoff_base_ms = Some(1_000);
2405        config.outbox.backoff_max_ms = Some(500);
2406        let error = config
2407            .validate()
2408            .err()
2409            .ok_or("backoff_max below backoff_base must fail")?;
2410        assert!(
2411            error.to_string().contains("outbox.backoff_max_ms"),
2412            "error must name the offending key: {error}"
2413        );
2414        Ok(())
2415    }
2416
2417    #[test]
2418    fn outbox_enabled_can_leave_reconciliation_dark() -> Result<(), Box<dyn std::error::Error>> {
2419        let mut config = outbox_enabled_base();
2420        config.outbox.reconcile_interval_ms = None;
2421        config.outbox.reconcile_stale_after_ms = None;
2422        config.validate()?;
2423        Ok(())
2424    }
2425
2426    #[test]
2427    fn outbox_reconciliation_requires_interval_when_partially_enabled()
2428    -> Result<(), Box<dyn std::error::Error>> {
2429        let mut config = outbox_enabled_base();
2430        config.outbox.reconcile_interval_ms = None;
2431        let error = config
2432            .validate()
2433            .err()
2434            .ok_or("reconciliation without interval must fail")?;
2435        assert!(error.to_string().contains("outbox.reconcile_interval_ms"));
2436        Ok(())
2437    }
2438
2439    #[test]
2440    fn outbox_reconciliation_requires_stale_threshold_when_partially_enabled()
2441    -> Result<(), Box<dyn std::error::Error>> {
2442        let mut config = outbox_enabled_base();
2443        config.outbox.reconcile_stale_after_ms = None;
2444        let error = config
2445            .validate()
2446            .err()
2447            .ok_or("reconciliation without stale threshold must fail")?;
2448        assert!(
2449            error
2450                .to_string()
2451                .contains("outbox.reconcile_stale_after_ms")
2452        );
2453        Ok(())
2454    }
2455
2456    #[test]
2457    fn package_discovery_is_sorted() -> Result<(), Box<dyn std::error::Error>> {
2458        let temp_dir = crate::test_support::private_tempdir()?;
2459        std::fs::write(temp_dir.path().join("zeta.aion"), b"package")?;
2460        std::fs::write(temp_dir.path().join("alpha.aion"), b"package")?;
2461        std::fs::write(temp_dir.path().join("ignored.txt"), b"package")?;
2462        std::fs::create_dir(temp_dir.path().join("nested"))?;
2463        std::fs::write(
2464            temp_dir.path().join("nested").join("nested.aion"),
2465            b"package",
2466        )?;
2467
2468        let packages = discover_workflow_packages(temp_dir.path())?;
2469
2470        assert_eq!(
2471            packages,
2472            vec![
2473                temp_dir.path().join("alpha.aion"),
2474                temp_dir.path().join("zeta.aion"),
2475            ]
2476        );
2477        Ok(())
2478    }
2479
2480    #[test]
2481    fn workflow_package_merge_is_additive_and_deduplicated() {
2482        let mut packages = vec!["config.aion".into(), "shared.aion".into()];
2483        let discovered = vec!["auto.aion".into(), "shared.aion".into()];
2484        let cli = vec!["cli.aion".into(), "auto.aion".into()];
2485
2486        merge_workflow_packages(&mut packages, discovered, &cli);
2487
2488        assert_eq!(
2489            packages,
2490            vec![
2491                std::path::PathBuf::from("config.aion"),
2492                std::path::PathBuf::from("shared.aion"),
2493                std::path::PathBuf::from("auto.aion"),
2494                std::path::PathBuf::from("cli.aion"),
2495            ]
2496        );
2497    }
2498
2499    #[test]
2500    fn package_merge_deduplicates_canonical_files() -> Result<(), Box<dyn std::error::Error>> {
2501        let temp_dir = crate::test_support::private_tempdir()?;
2502        let package = temp_dir.path().join("hello.aion");
2503        std::fs::write(&package, b"package")?;
2504        let mut packages = vec![package.clone()];
2505        let discovered = vec![temp_dir.path().join(".").join("hello.aion")];
2506
2507        merge_workflow_packages(&mut packages, discovered, &[]);
2508
2509        assert_eq!(packages, vec![package]);
2510        Ok(())
2511    }
2512
2513    #[test]
2514    fn zero_config_cli_workflow_package_uses_in_memory_defaults()
2515    -> Result<(), Box<dyn std::error::Error>> {
2516        let temp_dir = crate::test_support::private_tempdir()?;
2517
2518        let cli = CliOverrides {
2519            workflow_packages: vec!["hello-world.aion".into()],
2520            ..CliOverrides::default()
2521        };
2522        let mut config = ServerConfig::default();
2523        // This test exercises CLI workflow-package discovery against the ephemeral
2524        // in-memory store, so it opts OUT of the new durable haematite default
2525        // explicitly (the default would otherwise carry a haematite data_dir).
2526        config.store.backend = StoreBackend::Memory;
2527        config.store.data_dir = None;
2528        // Even zero-config development runs must size event streaming and the
2529        // query reply deadline explicitly (config keys or the
2530        // AION_WEBSOCKET_EVENT_BROADCAST_CAPACITY /
2531        // AION_RUNTIME_QUERY_TIMEOUT_MS environment overrides).
2532        config.websocket.event_broadcast_capacity = Some(64);
2533        config.websocket.cluster_broadcast_capacity = Some(64);
2534        config.runtime.query_timeout_ms = Some(10_000);
2535        config.load_discovered_workflow_packages(&cli, temp_dir.path())?;
2536
2537        config.validate()?;
2538
2539        assert_eq!(config.store.backend, StoreBackend::Memory);
2540        assert_eq!(
2541            config.workflow_packages,
2542            vec![std::path::PathBuf::from("hello-world.aion")]
2543        );
2544        Ok(())
2545    }
2546
2547    #[test]
2548    fn cli_packages_are_additive() -> Result<(), Box<dyn std::error::Error>> {
2549        let mut config = ServerConfig::from_slice(
2550            br#"
2551                workflow_packages = ["config.aion"]
2552
2553                [runtime]
2554                query_timeout_ms = 10000
2555
2556                [websocket]
2557                event_broadcast_capacity = 64
2558                cluster_broadcast_capacity = 64
2559            "#,
2560        )?;
2561        let cli = CliOverrides {
2562            workflow_packages: vec!["cli-one.aion".into(), "cli-two.aion".into()],
2563            ..CliOverrides::default()
2564        };
2565
2566        merge_workflow_packages(
2567            &mut config.workflow_packages,
2568            Vec::new(),
2569            &cli.workflow_packages,
2570        );
2571
2572        assert_eq!(
2573            config.workflow_packages,
2574            vec![
2575                std::path::PathBuf::from("config.aion"),
2576                std::path::PathBuf::from("cli-one.aion"),
2577                std::path::PathBuf::from("cli-two.aion"),
2578            ]
2579        );
2580        Ok(())
2581    }
2582}