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