Skip to main content

aion_server/config/
load.rs

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