aion_server/config/

load.rs

//! [`ServerConfig`]: the complete merged server configuration, its load /
//! merge / validate pipeline, and the workflow-package discovery helpers.
//!
//! This is the assembly point of the config surface: [`ServerConfig`] holds one
//! field per `[section]` (defined in [`super::sections`]), and its impls carry
//! the load/merge/CLI-override/default-fill/validate pipeline plus
//! [`ServerConfig::into_parts`], which splits the durable store config from the
//! non-secret [`RuntimeConfig`] runtime view. It is re-exported from the
//! `config` module so every existing `crate::config::X` path resolves
//! identically.

use std::{
    collections::HashSet,
    fs,
    path::{Path, PathBuf},
    time::Duration,
};

use serde::Deserialize;

use crate::error::ServerError;

use super::{
    AUTHORING_GLEAM_PATH_EMPTY, AUTHORING_PROJECT_ROOT_REQUIRED, AuthConfig, AuthoringConfig,
    CORS_ALLOWED_ORIGIN_INVALID, CliOverrides, ClusterConfig, ConfigResolution,
    DEFAULT_CLUSTER_BROADCAST_CAPACITY, DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES,
    DEFAULT_DEPLOY_MAX_INFLATED_BYTES, DEFAULT_EVENT_BROADCAST_CAPACITY,
    DEFAULT_OUTBOX_BACKOFF_BASE_MS, DEFAULT_OUTBOX_BACKOFF_MAX_MS,
    DEFAULT_OUTBOX_BACKOFF_MULTIPLIER, DEFAULT_OUTBOX_BATCH_SIZE, DEFAULT_OUTBOX_MAX_ATTEMPTS,
    DEFAULT_OUTBOX_POLL_INTERVAL_MS, DEFAULT_QUERY_TIMEOUT_MS, DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED,
    DEPLOY_MAX_INFLATED_BYTES_REQUIRED, DeployConfig, DevConfig, DrainConfig, HomeSource,
    ListenConfig, McpConfig, MetricsConfig, NamespaceConfig, NamespaceMode, NamespacesConfig,
    OUTBOX_BACKOFF_BASE_REQUIRED, OUTBOX_BACKOFF_MAX_REQUIRED, OUTBOX_BACKOFF_MULTIPLIER_REQUIRED,
    OUTBOX_BATCH_SIZE_REQUIRED, OUTBOX_MAX_ATTEMPTS_REQUIRED, OUTBOX_POLL_INTERVAL_REQUIRED,
    OUTBOX_RECONCILE_INTERVAL_REQUIRED, OUTBOX_RECONCILE_STALE_AFTER_REQUIRED, ObservabilityConfig,
    OpsConsoleAssetSource, OpsConsoleConfig, OutboxConfig, QUERY_TIMEOUT_REQUIRED, RuntimeConfig,
    RuntimeSection, ServerSection, StoreBackend, StoreConfig, TlsConfig, WebSocketConfig,
    WorkerConfig, WorkerSupervisionConfig, aion_home, config_error, env, file, home,
    resolution::fill_home_defaults, sections::RetiredStoreInput,
};

/// Complete merged server configuration.
#[derive(Clone, Debug, Deserialize)]
#[serde(default, deny_unknown_fields)]
#[derive(Default)]
pub struct ServerConfig {
    /// Public listener and transport addresses.
    pub server: ServerSection,
    /// Event-store backend configuration.
    pub store: StoreConfig,
    /// Engine runtime settings.
    pub runtime: RuntimeSection,
    /// Shutdown drain settings.
    pub drain: DrainConfig,
    /// Authentication settings defined by the operations config surface.
    pub auth: AuthConfig,
    /// Metrics endpoint settings.
    pub metrics: MetricsConfig,
    /// Namespace defaults.
    pub namespaces: NamespacesConfig,
    /// Optional TLS material for transports that require it.
    pub tls: Option<TlsConfig>,
    /// Static ops-console asset bundle location.
    #[serde(alias = "dashboard")]
    pub ops_console: OpsConsoleConfig,
    /// Namespace resolver construction mode retained for existing transports.
    pub namespace: NamespaceConfig,
    /// Remote-worker heartbeat policy.
    pub worker: WorkerConfig,
    /// WebSocket event streaming policy.
    pub websocket: WebSocketConfig,
    /// Workflow package archives loaded into the engine at startup.
    pub workflow_packages: Vec<PathBuf>,
    /// Operator deploy API settings.
    pub deploy: DeployConfig,
    /// AWL studio and optional server-side Gleam authoring settings.
    pub authoring: AuthoringConfig,
    /// Local dev-server surface settings.
    pub dev: DevConfig,
    /// Durable-outbox fan-out dispatcher settings.
    pub outbox: OutboxConfig,
    /// Agent-observability transcript retention bounds.
    pub observability: ObservabilityConfig,
    /// Model Context Protocol surface settings.
    pub mcp: McpConfig,
    /// Managed-worker supervision policy. Absent = supervision uncommissioned:
    /// the server runs no worker processes and says so with the remedy.
    pub worker_supervision: WorkerSupervisionConfig,
}

pub(crate) struct LoadedConfig {
    pub(crate) config: ServerConfig,
    pub(crate) resolution: ConfigResolution,
}

/// Expand a leading `~` in the operator-supplied filesystem paths.
///
/// #180 review m10: `~` already expands for `AION_HOME` and `--config`, so an
/// operator naturally writes it in the path VALUES too — where it used to pass
/// through untouched and silently become a literal `./~` directory. The same
/// [`home::expand_tilde`] rule applies to `store.data_dir` and
/// `authoring.workspace_dir` (the two home-adjacent state roots): `~`/`~/…`
/// expand against `$HOME`, `~user` is refused by name, and a path with no
/// tilde passes through byte-for-byte.
fn expand_operator_path_tildes(config: &mut ServerConfig) -> Result<(), ServerError> {
    if let Some(data_dir) = config.store.data_dir.as_deref() {
        let expanded = home::expand_tilde(Path::new(data_dir))?;
        let expanded = expanded.to_str().ok_or_else(|| ServerError::Config {
            message: format!(
                "store.data_dir `{data_dir}` expands to a path that is not valid UTF-8"
            ),
        })?;
        config.store.data_dir = Some(expanded.to_owned());
    }
    if let Some(workspace_dir) = config.authoring.workspace_dir.as_deref() {
        config.authoring.workspace_dir = Some(home::expand_tilde(workspace_dir)?);
    }
    Ok(())
}

impl ServerConfig {
    /// Load and merge config from defaults, optional TOML file, environment, and CLI overrides.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::Config`] when file discovery, parsing, environment parsing, CLI
    /// values, or validation fail.
    pub fn load(cli: &CliOverrides) -> Result<Self, ServerError> {
        Ok(Self::load_resolved(cli)?.config)
    }

    pub(crate) fn load_resolved(cli: &CliOverrides) -> Result<LoadedConfig, ServerError> {
        let home = aion_home()?;
        let working_dir = std::env::current_dir().map_err(|source| ServerError::Config {
            message: format!(
                "failed to resolve the current directory for config discovery: {source}"
            ),
        })?;
        Self::load_in(cli, &home.path, home.source, &working_dir, true)
    }

    fn load_in(
        cli: &CliOverrides,
        home: &Path,
        home_source: HomeSource,
        working_dir: &Path,
        overlay_environment: bool,
    ) -> Result<LoadedConfig, ServerError> {
        // #152: `--config ~/aion.toml` reaches this point unexpanded whenever a
        // shell did not get to it (quoted values, env files, process managers);
        // expand it here so the operator's spelling and the loader agree.
        let explicit = cli
            .config_path
            .as_deref()
            .map(super::home::expand_tilde)
            .transpose()?;
        let discovered = file::discover(explicit.as_deref(), home, working_dir)?;
        let mut config = match discovered.bytes {
            Some(bytes) => Self::parse_unresolved(&bytes).map_err(|error| ServerError::Config {
                message: format!("failed to parse {}: {error}", discovered.source),
            })?,
            None => Self::default(),
        };
        if overlay_environment {
            env::overlay(&mut config)?;
        }
        config.apply_cli_overrides(cli);
        expand_operator_path_tildes(&mut config)?;
        #[cfg(not(unix))]
        let home_explicit = !overlay_environment || std::env::var_os("AION_HOME").is_some();
        #[cfg(not(unix))]
        let data_dir_explicit = config.store.data_dir.is_some();
        #[cfg(not(unix))]
        let data_root_required = matches!(config.store.backend, StoreBackend::Haematite);
        #[cfg(not(unix))]
        let authoring_workspace_explicit = config.authoring.workspace_dir.is_some();
        config.load_discovered_workflow_packages(cli, working_dir)?;
        let legacy_notices = fill_home_defaults(&mut config, home, home_source, working_dir)?;
        config.fill_operational_defaults();
        config.validate()?;
        let resolution = ConfigResolution {
            home: home.to_owned(),
            source: discovered.source,
            data_dir: config.store.data_dir.clone(),
            authoring_workspace: config.authoring.workspace_dir.clone(),
            legacy_notices,
            #[cfg(not(unix))]
            home_explicit,
            #[cfg(not(unix))]
            data_dir_explicit,
            #[cfg(not(unix))]
            data_root_required,
            #[cfg(not(unix))]
            authoring_workspace_explicit,
        };
        Ok(LoadedConfig { config, resolution })
    }

    /// Load with an explicitly supplied home, without environment or CLI
    /// overlays.
    ///
    /// `home_source` is a parameter rather than a constant because the two
    /// worlds it selects between are exactly what the legacy-directory guard
    /// turns on, and a test must be able to say which world it is standing in.
    /// A test that could only ever simulate one of them would pin half the
    /// behaviour and call it covered.
    #[cfg(test)]
    pub(super) fn load_for_test(
        cli: &CliOverrides,
        home: &Path,
        home_source: HomeSource,
        working_dir: &Path,
    ) -> Result<LoadedConfig, ServerError> {
        Self::load_in(cli, home, home_source, working_dir, false)
    }

    /// Fill operational tuning knobs that have a sane default when omitted, so a
    /// minimal or empty config boots without forcing the operator to hand-author
    /// values that are pure tuning. Uses `get_or_insert`, so an explicitly set
    /// value (including a misconfigured `0`, which [`Self::validate`] still
    /// rejects) is left untouched; only an absent (`None`) field is defaulted.
    fn fill_operational_defaults(&mut self) {
        self.runtime
            .query_timeout_ms
            .get_or_insert(DEFAULT_QUERY_TIMEOUT_MS);
        self.websocket
            .event_broadcast_capacity
            .get_or_insert(DEFAULT_EVENT_BROADCAST_CAPACITY);
        self.websocket
            .cluster_broadcast_capacity
            .get_or_insert(DEFAULT_CLUSTER_BROADCAST_CAPACITY);
        self.fill_outbox_defaults();
        self.fill_deploy_defaults();
    }

    /// Fill the durable-outbox tuning knobs with sane defaults when the
    /// dispatcher is enabled but a knob was omitted, so turning the feature on
    /// does not force hand-authoring pure tuning. Inert while `outbox.enabled`
    /// is false (the knobs are never read behind the gate). The reconciliation
    /// pair is intentionally NOT defaulted: when both are absent reconciliation
    /// stays dark, so forcing a default would silently commission a sweep.
    /// `get_or_insert` leaves any explicit value (including a misconfigured `0`,
    /// which [`Self::validate_outbox`] still rejects) untouched.
    fn fill_outbox_defaults(&mut self) {
        if !self.outbox.enabled {
            return;
        }
        self.outbox
            .poll_interval_ms
            .get_or_insert(DEFAULT_OUTBOX_POLL_INTERVAL_MS);
        self.outbox
            .batch_size
            .get_or_insert(DEFAULT_OUTBOX_BATCH_SIZE);
        self.outbox
            .max_attempts
            .get_or_insert(DEFAULT_OUTBOX_MAX_ATTEMPTS);
        self.outbox
            .backoff_base_ms
            .get_or_insert(DEFAULT_OUTBOX_BACKOFF_BASE_MS);
        self.outbox
            .backoff_multiplier
            .get_or_insert(DEFAULT_OUTBOX_BACKOFF_MULTIPLIER);
        self.outbox
            .backoff_max_ms
            .get_or_insert(DEFAULT_OUTBOX_BACKOFF_MAX_MS);
    }

    /// Fill the deploy decompression-bomb ceilings with conservative defaults
    /// when the deploy surface is enabled but a ceiling was omitted, so turning
    /// the feature on boots rather than refusing for want of a security knob.
    /// Inert while `deploy.enabled` is false (the ceilings are never read with
    /// the surface dark). `get_or_insert` leaves any explicit value (including a
    /// misconfigured `0` or an inflate ceiling below the archive ceiling, both
    /// still rejected by [`Self::validate`]) untouched.
    fn fill_deploy_defaults(&mut self) {
        if !self.deploy.enabled {
            return;
        }
        self.deploy
            .max_archive_bytes
            .get_or_insert(DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES);
        self.deploy
            .max_inflated_bytes
            .get_or_insert(DEFAULT_DEPLOY_MAX_INFLATED_BYTES);
    }

    fn load_discovered_workflow_packages(
        &mut self,
        cli: &CliOverrides,
        directory: &Path,
    ) -> Result<(), ServerError> {
        let discovered_packages = discover_workflow_packages(directory)?;
        merge_workflow_packages(
            &mut self.workflow_packages,
            discovered_packages,
            &cli.workflow_packages,
        );
        Ok(())
    }

    /// Parse server configuration from TOML bytes and validate it.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::Config`] when parsing fails or values are invalid.
    pub fn from_slice(bytes: &[u8]) -> Result<Self, ServerError> {
        let home = aion_home()?;
        Self::from_slice_with_home(bytes, &home.path)
    }

    /// Parse server configuration using an explicitly supplied Aion home.
    ///
    /// This is useful for embedded callers and tests that must not consult or
    /// touch the process user's real home. It applies the same dynamic defaults
    /// as [`Self::load`], but does not apply environment or CLI overlays, and
    /// naming the home here suppresses the working-directory legacy fallback
    /// exactly as setting `AION_HOME` does — an embedded caller that supplies a
    /// home is isolating this server's state, not asking to inherit whatever
    /// directory the process happens to have been started in.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::Config`] when parsing, current-directory
    /// resolution, dynamic path defaulting, or validation fails.
    pub fn from_slice_with_home(bytes: &[u8], home: &Path) -> Result<Self, ServerError> {
        let working_dir = std::env::current_dir().map_err(|source| ServerError::Config {
            message: format!(
                "failed to resolve the current directory for config defaults: {source}"
            ),
        })?;
        let mut config = Self::parse_unresolved(bytes)?;
        // The caller named the home in the call itself, which is the same
        // assertion `AION_HOME` makes and is treated identically: a legacy
        // directory in the working directory must not override it.
        fill_home_defaults(&mut config, home, HomeSource::Explicit, &working_dir)?;
        config.fill_operational_defaults();
        config.validate()?;
        Ok(config)
    }

    fn parse_unresolved(bytes: &[u8]) -> Result<Self, ServerError> {
        toml::from_slice(bytes).map_err(|source| ServerError::Config {
            message: format!("invalid server config: {source}"),
        })
    }

    /// Load server configuration from an explicit TOML file path.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::Config`] when the file is missing, unreadable, unparsable, or invalid.
    pub fn load_from_path(path: impl Into<PathBuf>) -> Result<Self, ServerError> {
        file::load_required(&path.into())
    }

    /// Split store configuration from non-secret runtime settings.
    #[must_use]
    pub fn into_parts(self) -> (StoreConfig, RuntimeConfig) {
        let runtime = RuntimeConfig {
            listen: ListenConfig {
                grpc: self.server.grpc_address,
                http: self.server.listen_address,
            },
            tls: self.tls,
            auth: self.auth,
            ops_console: self.ops_console,
            namespace: self.namespace,
            worker: self.worker,
            websocket: self.websocket,
            workflow_packages: self.workflow_packages,
            deploy: self.deploy,
            authoring: self.authoring,
            dev: self.dev,
            outbox: self.outbox,
            observability: self.observability,
            mcp: self.mcp.resolved(),
            scheduler_threads: self.runtime.scheduler_threads,
            query_timeout: self.runtime.query_timeout_ms.map(Duration::from_millis),
            default_namespace: self.namespaces.default,
            auto_create: self.namespaces.auto_create,
            max_in_flight_activities: self.namespaces.max_in_flight_activities,
            drain_timeout: Duration::from_secs(self.drain.timeout_seconds),
            metrics: self.metrics,
            owned_shards: self.store.owned_shards.clone(),
            cors_allowed_origins: self.server.cors_allowed_origins.clone(),
        };
        (self.store, runtime)
    }

    fn apply_cli_overrides(&mut self, cli: &CliOverrides) {
        if let Some(address) = cli.listen_address {
            self.server.listen_address = address;
        }
        if let Some(address) = cli.grpc_address {
            self.server.grpc_address = address;
        }
        if cli.store_url.is_some() {
            self.store.retired_input = Some(RetiredStoreInput::Flag);
        }
        if let Some(threads) = cli.scheduler_threads {
            self.runtime.scheduler_threads = threads;
        }
        if let Some(timeout) = cli.drain_timeout_seconds {
            self.drain.timeout_seconds = timeout;
        }
        if let Some(gleam_path) = &cli.gleam_path {
            self.authoring.gleam_path = Some(gleam_path.clone());
        }
        if let Some(project_root) = &cli.authoring_project_root {
            self.authoring.project_root = Some(project_root.clone());
        }
    }

    fn validate(&self) -> Result<(), ServerError> {
        if let Some(input) = self.store.retired_input {
            let found = match input {
                RetiredStoreInput::Backend => "backend = \"libsql\"",
                RetiredStoreInput::BackendEnvironment => "AION_STORE_BACKEND=libsql",
                RetiredStoreInput::Url => "store.url",
                RetiredStoreInput::Environment => "AION_STORE_URL",
                RetiredStoreInput::Flag => "--store-url",
            };
            return config_error(format!(
                "found retired {found}; use backend = \"haematite\" with store.data_dir and store.node_cache_budget (all three are required — haematite refuses to start without the budget); existing libsql database files are not read by this build"
            ));
        }
        if self.server.listen_address.port() == 0 {
            return config_error("server.listen_address must use an explicit non-zero port");
        }
        if self.server.grpc_address.port() == 0 {
            return config_error("server.grpc_address must use an explicit non-zero port");
        }
        validate_cors_origins(&self.server.cors_allowed_origins)?;
        if self.runtime.scheduler_threads == 0 {
            return config_error("runtime.scheduler_threads must be greater than zero");
        }
        if self.drain.timeout_seconds == 0 {
            return config_error("drain.timeout_seconds must be greater than zero");
        }
        if self.auth.enabled && self.auth.jwks_url.as_deref().is_none_or(str::is_empty) {
            return config_error("auth.jwks_url must not be empty when auth.enabled is true");
        }
        if self.auth.jwks_refresh_seconds == 0 {
            return config_error("auth.jwks_refresh_seconds must be greater than zero");
        }
        if self.namespaces.default.is_empty() {
            return config_error("namespaces.default must not be empty");
        }
        if matches!(self.store.backend, StoreBackend::Haematite) {
            if self.store.data_dir.as_deref().is_none_or(str::is_empty) {
                return config_error(
                    "store.data_dir must not be empty when store.backend is haematite",
                );
            }
            if self.store.shard_count == 0 {
                return config_error("store.shard_count must be greater than zero");
            }
            if let Some(cluster) = &self.store.cluster {
                validate_cluster(cluster)?;
            }
        } else if self.store.cluster.is_some() {
            return config_error("store.cluster is only valid when store.backend is haematite");
        }
        if let OpsConsoleAssetSource::FileSystem { asset_path } = &self.ops_console.source {
            if asset_path.as_os_str().is_empty() {
                return config_error("ops_console.source.FileSystem.asset_path must not be empty");
            }
        }
        if let NamespaceMode::SingleTenant { namespace } = &self.namespace.mode {
            if namespace.is_empty() {
                return config_error("namespace.mode.SingleTenant.namespace must not be empty");
            }
        }
        if self.worker.heartbeat_window.is_zero() {
            return config_error("worker.heartbeat_window must be greater than zero");
        }
        self.websocket.validate()?;
        self.observability.validate()?;
        self.mcp.validate()?;
        match self.runtime.query_timeout_ms {
            None | Some(0) => return config_error(QUERY_TIMEOUT_REQUIRED),
            Some(_) => {}
        }
        if self.deploy.enabled {
            let max_archive_bytes = match self.deploy.max_archive_bytes {
                None | Some(0) => return config_error(DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED),
                Some(value) => value,
            };
            let max_inflated_bytes = match self.deploy.max_inflated_bytes {
                None | Some(0) => return config_error(DEPLOY_MAX_INFLATED_BYTES_REQUIRED),
                Some(value) => value,
            };
            // Both ceilings size in-memory buffers, so they must be
            // addressable on this platform (32-bit targets).
            ensure_fits_usize("deploy.max_archive_bytes", max_archive_bytes)?;
            ensure_fits_usize("deploy.max_inflated_bytes", max_inflated_bytes)?;
            if max_inflated_bytes < max_archive_bytes {
                return config_error(format!(
                    "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"
                ));
            }
        }
        if let Some(gleam_path) = &self.authoring.gleam_path {
            // The authoring surface is commissioned by a non-empty gleam_path;
            // an empty value is a misconfiguration, not "dark".
            if gleam_path.as_os_str().is_empty() {
                return config_error(AUTHORING_GLEAM_PATH_EMPTY);
            }
            // Commissioning the loop requires a project root with no default
            // (a Gleam project cannot be invented; the operator provisions it).
            match &self.authoring.project_root {
                Some(root) if !root.as_os_str().is_empty() => {}
                _ => return config_error(AUTHORING_PROJECT_ROOT_REQUIRED),
            }
        }
        self.validate_outbox()?;
        // Resolution IS the validation: a partial or nonsensical
        // `[worker_supervision]` section refuses here, at load, rather than at
        // the first `aion worker start` hours later.
        self.worker_supervision.resolve()?;
        Ok(())
    }

    /// Validate the durable-outbox dispatcher knobs.
    ///
    /// All knobs are inert while `outbox.enabled` is false (the dispatcher is
    /// never spawned), so they are only required — and only checked — once the
    /// operator commissions the dispatcher. This mirrors the dark-by-default
    /// `deploy` surface: the on/off gate carries no defaults, and every
    /// operational value behind it is an explicit operator decision.
    fn validate_outbox(&self) -> Result<(), ServerError> {
        if !self.outbox.enabled {
            return Ok(());
        }
        match self.outbox.poll_interval_ms {
            None | Some(0) => return config_error(OUTBOX_POLL_INTERVAL_REQUIRED),
            Some(_) => {}
        }
        match self.outbox.batch_size {
            None | Some(0) => return config_error(OUTBOX_BATCH_SIZE_REQUIRED),
            Some(_) => {}
        }
        match self.outbox.max_attempts {
            None | Some(0) => return config_error(OUTBOX_MAX_ATTEMPTS_REQUIRED),
            Some(_) => {}
        }
        let backoff_base_ms = match self.outbox.backoff_base_ms {
            None | Some(0) => return config_error(OUTBOX_BACKOFF_BASE_REQUIRED),
            Some(value) => value,
        };
        match self.outbox.backoff_multiplier {
            None | Some(0) => return config_error(OUTBOX_BACKOFF_MULTIPLIER_REQUIRED),
            Some(_) => {}
        }
        match self.outbox.backoff_max_ms {
            Some(max) if max >= backoff_base_ms => {}
            _ => return config_error(OUTBOX_BACKOFF_MAX_REQUIRED),
        }
        match (
            self.outbox.reconcile_interval_ms,
            self.outbox.reconcile_stale_after_ms,
        ) {
            (None, None) => {}
            (None | Some(0), _) => return config_error(OUTBOX_RECONCILE_INTERVAL_REQUIRED),
            (_, None | Some(0)) => return config_error(OUTBOX_RECONCILE_STALE_AFTER_REQUIRED),
            (Some(_), Some(_)) => {}
        }
        Ok(())
    }
}

/// Validate a `[store.cluster]` section: a non-empty node id, and every member /
/// peer name non-empty. A cluster of one (no peers, members empty or `[node_id]`)
/// is valid.
fn validate_cluster(cluster: &ClusterConfig) -> Result<(), ServerError> {
    if cluster.node_id.is_empty() {
        return config_error("store.cluster.node_id must not be empty");
    }
    if cluster.members.iter().any(String::is_empty) {
        return config_error("store.cluster.members entries must not be empty");
    }
    if cluster.peers.iter().any(|peer| peer.name.is_empty()) {
        return config_error("store.cluster.peers entries must name a non-empty node");
    }
    if matches!(cluster.failover_poll_interval_ms, Some(0)) {
        return config_error(
            "store.cluster.failover_poll_interval_ms must be greater than zero when set",
        );
    }
    if matches!(cluster.failover_confirmations, Some(0)) {
        return config_error("store.cluster.failover_confirmations must be at least one when set");
    }
    Ok(())
}

/// Validate every `[server] cors_allowed_origins` entry.
fn validate_cors_origins(origins: &[String]) -> Result<(), ServerError> {
    for origin in origins {
        validate_cors_origin(origin)?;
    }
    Ok(())
}

/// Validate one `[server] cors_allowed_origins` entry: it must be a non-empty,
/// parseable HTTP origin so the `CorsLayer` can match it against the browser's
/// `Origin` header. A malformed origin can never match a real request, so it is
/// a misconfiguration caught at startup rather than silently never matching.
fn validate_cors_origin(origin: &str) -> Result<(), ServerError> {
    if origin.is_empty() {
        return config_error(CORS_ALLOWED_ORIGIN_INVALID);
    }
    // An origin is scheme + host + optional port and carries no path: reject a
    // trailing slash or any path segment, which would never equal a browser
    // `Origin` header value.
    let scheme_split = origin.split_once("://");
    let Some((scheme, authority)) = scheme_split else {
        return config_error(CORS_ALLOWED_ORIGIN_INVALID);
    };
    if scheme.is_empty() || authority.is_empty() || authority.contains('/') {
        return config_error(CORS_ALLOWED_ORIGIN_INVALID);
    }
    // It must parse as an HTTP header value (the form the CorsLayer compares).
    if origin.parse::<axum::http::HeaderValue>().is_err() {
        return config_error(CORS_ALLOWED_ORIGIN_INVALID);
    }
    Ok(())
}

/// Refuses byte-ceiling values that cannot index memory on this platform.
fn ensure_fits_usize(key: &str, value: u64) -> Result<(), ServerError> {
    if usize::try_from(value).is_err() {
        return config_error(format!(
            "{key} ({value}) exceeds this platform's addressable memory; set it to at most {}",
            usize::MAX
        ));
    }
    Ok(())
}

fn discover_workflow_packages(directory: &Path) -> Result<Vec<PathBuf>, ServerError> {
    let mut packages = Vec::new();
    let entries = fs::read_dir(directory).map_err(|source| ServerError::Config {
        message: format!(
            "failed to scan workflow packages in `{}`: {source}",
            directory.display()
        ),
    })?;

    for entry in entries {
        let entry = entry.map_err(|source| ServerError::Config {
            message: format!(
                "failed to read workflow package entry in `{}`: {source}",
                directory.display()
            ),
        })?;
        let path = entry.path();
        let has_aion_extension = path
            .extension()
            .is_some_and(|extension| extension == "aion");
        if path.is_file() && has_aion_extension {
            packages.push(path);
        }
    }

    packages.sort_by(|left, right| left.as_os_str().cmp(right.as_os_str()));
    Ok(packages)
}

fn merge_workflow_packages(
    workflow_packages: &mut Vec<PathBuf>,
    discovered_packages: Vec<PathBuf>,
    cli_packages: &[PathBuf],
) {
    let mut seen: HashSet<PathBuf> = workflow_packages
        .iter()
        .map(|package| deduplicated_package_key(package))
        .collect();
    for package in discovered_packages
        .into_iter()
        .chain(cli_packages.iter().cloned())
    {
        if seen.insert(deduplicated_package_key(&package)) {
            workflow_packages.push(package);
        }
    }
}

fn deduplicated_package_key(path: &Path) -> PathBuf {
    path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
}

#[cfg(test)]
#[path = "load_home_tests.rs"]
mod home_tests;

#[cfg(test)]
mod tests {
    use crate::config::{
        AutoCreate, DEFAULT_AUTHORING_WORKSPACE_DIR, DEFAULT_MAX_IN_FLIGHT_ACTIVITIES, HomeSource,
        OpsConsoleAssetSource,
    };

    use super::{
        CliOverrides, DEFAULT_CLUSTER_BROADCAST_CAPACITY, DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES,
        DEFAULT_DEPLOY_MAX_INFLATED_BYTES, DEFAULT_EVENT_BROADCAST_CAPACITY,
        DEFAULT_OUTBOX_BACKOFF_BASE_MS, DEFAULT_OUTBOX_BACKOFF_MAX_MS,
        DEFAULT_OUTBOX_BACKOFF_MULTIPLIER, DEFAULT_OUTBOX_BATCH_SIZE, DEFAULT_OUTBOX_MAX_ATTEMPTS,
        DEFAULT_OUTBOX_POLL_INTERVAL_MS, DEFAULT_QUERY_TIMEOUT_MS, ServerConfig, StoreBackend,
        discover_workflow_packages, merge_workflow_packages,
    };

    #[test]
    fn valid_toml_is_parsed_into_typed_config() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br#"
                [server]
                listen_address = "127.0.0.1:18080"
                grpc_address = "127.0.0.1:15051"

                [store]
                backend = "haematite"
                data_dir = "aion-data"

                [runtime]
                scheduler_threads = 2
                query_timeout_ms = 10000

                [drain]
                timeout_seconds = 45

                [auth]
                enabled = true
                jwks_url = "https://issuer.example.com/.well-known/jwks.json"
                jwks_refresh_seconds = 60

                [metrics]
                enabled = true

                [namespaces]
                default = "production"

                [websocket]
                outbound_buffer_bound = 16
                event_broadcast_capacity = 1024
                cluster_broadcast_capacity = 1024
            "#,
        )?;

        assert_eq!(config.store.backend, StoreBackend::Haematite);
        assert_eq!(config.store.data_dir.as_deref(), Some("aion-data"));
        assert_eq!(config.runtime.scheduler_threads, 2);
        assert_eq!(config.runtime.query_timeout_ms, Some(10_000));
        assert_eq!(config.namespaces.default, "production");
        // `auto_create` is omitted above, so it resolves to the Open default.
        assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
        // `max_in_flight_activities` is omitted above, so it resolves to the
        // generous platform default.
        assert_eq!(
            config.namespaces.max_in_flight_activities,
            DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
        );
        assert_eq!(config.websocket.outbound_buffer_bound, 16);
        assert_eq!(config.websocket.event_broadcast_capacity, Some(1024));
        Ok(())
    }

    #[test]
    fn namespaces_auto_create_closed_parses() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br#"
                [namespaces]
                default = "production"
                auto_create = "closed"
            "#,
        )?;
        assert_eq!(config.namespaces.default, "production");
        assert_eq!(config.namespaces.auto_create, AutoCreate::Closed);
        Ok(())
    }

    #[test]
    fn namespaces_auto_create_open_parses() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br#"
                [namespaces]
                auto_create = "open"
            "#,
        )?;
        assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
        Ok(())
    }

    #[test]
    fn namespaces_max_in_flight_activities_override_parses()
    -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br#"
                [namespaces]
                default = "production"
                max_in_flight_activities = 32
            "#,
        )?;
        assert_eq!(config.namespaces.max_in_flight_activities, 32);
        // The override also propagates into the runtime view.
        let (_store, runtime) = config.into_parts();
        assert_eq!(runtime.max_in_flight_activities, 32);
        Ok(())
    }

    #[test]
    fn namespaces_max_in_flight_activities_defaults_when_omitted()
    -> Result<(), Box<dyn std::error::Error>> {
        // An old/minimal config that predates the field omits it entirely and
        // resolves to the generous platform default (additive, not a migration).
        let config = ServerConfig::from_slice(
            br#"
                [namespaces]
                default = "production"
            "#,
        )?;
        assert_eq!(
            config.namespaces.max_in_flight_activities,
            DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
        );
        Ok(())
    }

    #[test]
    fn namespaces_auto_create_rejects_unknown_variant() {
        let result = ServerConfig::from_slice(
            br#"
                [namespaces]
                auto_create = "sometimes"
            "#,
        );
        assert!(
            result.is_err(),
            "an unknown auto_create variant must fail to parse"
        );
    }

    #[test]
    fn missing_event_broadcast_capacity_uses_default() -> Result<(), Box<dyn std::error::Error>> {
        // The server unconditionally mounts /events/stream, but the channel
        // capacity is a tuning knob: omitting it must resolve to the default and
        // boot, not fail startup.
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                cluster_broadcast_capacity = 64
            ",
        )?;
        assert_eq!(
            config.websocket.event_broadcast_capacity,
            Some(DEFAULT_EVENT_BROADCAST_CAPACITY),
            "omitted event_broadcast_capacity must resolve to the default"
        );
        Ok(())
    }

    #[test]
    fn zero_event_broadcast_capacity_fails_startup_validation() {
        let result = ServerConfig::from_slice(
            br"
                [websocket]
                event_broadcast_capacity = 0
            ",
        );

        let message = result
            .err()
            .map_or_else(String::new, |error| error.to_string());
        assert!(
            message.contains("websocket.event_broadcast_capacity"),
            "validation message must name the zero-valued key: {message}"
        );
    }

    #[test]
    fn missing_cluster_broadcast_capacity_uses_default() -> Result<(), Box<dyn std::error::Error>> {
        // A config that sizes the workflow channel but omits the low-rate cluster
        // channel must resolve the cluster capacity to its default and boot, not
        // fail loudly.
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                scheduler_threads = 1
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
            ",
        )?;
        assert_eq!(
            config.websocket.cluster_broadcast_capacity,
            Some(DEFAULT_CLUSTER_BROADCAST_CAPACITY),
            "omitted cluster_broadcast_capacity must resolve to the default"
        );
        Ok(())
    }

    #[test]
    fn zero_cluster_broadcast_capacity_fails_startup_validation() {
        let result = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 0
            ",
        );

        let message = result
            .err()
            .map_or_else(String::new, |error| error.to_string());
        assert!(
            message.contains("websocket.cluster_broadcast_capacity"),
            "validation message must name the zero-valued cluster key: {message}"
        );
    }

    /// An omitted `[observability]` section resolves both retention bounds to
    /// their defaults and boots — retention is on out of the box, never a
    /// forced operator decision.
    #[test]
    fn missing_observability_section_uses_defaults() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64
            ",
        )?;
        assert_eq!(
            config.observability.max_event_bytes,
            crate::config::DEFAULT_OBSERVABILITY_MAX_EVENT_BYTES
        );
        assert_eq!(
            config.observability.max_stream_events,
            crate::config::DEFAULT_OBSERVABILITY_MAX_STREAM_EVENTS
        );
        // The bounds also ride into the runtime view the server state reads.
        let (_store, runtime) = config.into_parts();
        assert_eq!(
            runtime.observability.max_event_bytes,
            crate::config::DEFAULT_OBSERVABILITY_MAX_EVENT_BYTES
        );
        Ok(())
    }

    /// Explicit `[observability]` values parse and round-trip into the runtime
    /// view.
    #[test]
    fn observability_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [observability]
                max_event_bytes = 512
                max_stream_events = 3
            ",
        )?;
        assert_eq!(config.observability.max_event_bytes, 512);
        assert_eq!(config.observability.max_stream_events, 3);
        let (_store, runtime) = config.into_parts();
        assert_eq!(runtime.observability.max_event_bytes, 512);
        assert_eq!(runtime.observability.max_stream_events, 3);
        Ok(())
    }

    /// An omitted `[mcp]` section leaves the surface DARK and every lifetime at
    /// its documented default. A config that never mentions MCP must not end up
    /// serving one.
    #[test]
    fn missing_mcp_section_leaves_the_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64
            ",
        )?;
        let (_store, runtime) = config.into_parts();
        assert!(!runtime.mcp.enabled);
        assert!(runtime.mcp.allowed_origins.is_empty());
        assert_eq!(
            runtime.mcp.discover_ttl_ms,
            crate::config::DEFAULT_MCP_DISCOVER_TTL_MS
        );
        assert_eq!(
            runtime.mcp.task_poll_interval_ms,
            crate::config::DEFAULT_MCP_TASK_POLL_INTERVAL_MS
        );
        assert_eq!(
            runtime.mcp.task_ttl_ms,
            Some(crate::config::DEFAULT_MCP_TASK_TTL_MS)
        );
        Ok(())
    }

    /// Explicit `[mcp]` values parse and round-trip, and `task_ttl_ms = 0` is
    /// the operator's way of saying "unlimited" — which the wire spells `null`.
    #[test]
    fn mcp_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br#"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [mcp]
                enabled = true
                allowed_origins = ["http://localhost:5173"]
                discover_ttl_ms = 1000
                tools_list_ttl_ms = 2000
                task_ttl_ms = 0
                task_poll_interval_ms = 250
                await_poll_interval_ms = 125
            "#,
        )?;
        let (_store, runtime) = config.into_parts();
        assert!(runtime.mcp.enabled);
        assert_eq!(runtime.mcp.allowed_origins, vec!["http://localhost:5173"]);
        assert_eq!(runtime.mcp.discover_ttl_ms, 1_000);
        assert_eq!(runtime.mcp.tools_list_ttl_ms, 2_000);
        assert_eq!(runtime.mcp.task_ttl_ms, None, "zero means unlimited");
        assert_eq!(runtime.mcp.task_poll_interval_ms, 250);
        assert_eq!(runtime.mcp.await_poll_interval_ms, 125);
        Ok(())
    }

    /// A zero poll interval is a busy loop, not a fast one. It is refused at
    /// startup with an operator-facing message rather than accepted.
    #[test]
    fn zero_mcp_poll_intervals_fail_startup_validation() {
        for key in ["task_poll_interval_ms", "await_poll_interval_ms"] {
            let source = format!(
                "
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [mcp]
                enabled = true
                {key} = 0
            "
            );
            let result = ServerConfig::from_slice(source.as_bytes());
            let message = match result {
                Err(crate::ServerError::Config { message }) => message,
                _ => String::new(),
            };
            assert!(message.contains(key), "{key} must be refused: {message}");
        }
    }

    /// The same zero on a DARK surface is not a boot failure: nothing reads it,
    /// so refusing the boot would be refusing over a value that has no effect.
    #[test]
    fn a_dark_mcp_surface_does_not_validate_its_unused_knobs()
    -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [mcp]
                task_poll_interval_ms = 0
            ",
        )?;
        let (_store, runtime) = config.into_parts();
        assert!(!runtime.mcp.enabled);
        assert_eq!(
            runtime.mcp.task_poll_interval_ms,
            crate::config::DEFAULT_MCP_TASK_POLL_INTERVAL_MS,
            "an unused zero resolves to the default rather than to a busy loop"
        );
        Ok(())
    }

    /// The flush policy parses and round-trips into the runtime view. Both keys
    /// are `Option` with NO default, so what a config states is exactly what the
    /// boot path sees — including a hold of zero, which is a real setting
    /// ("never wait"), not an absence.
    #[test]
    fn observability_flush_policy_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>>
    {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [observability]
                max_batch_events = 32
                max_batch_hold_ms = 0
            ",
        )?;
        assert_eq!(config.observability.max_batch_events, Some(32));
        assert_eq!(config.observability.max_batch_hold_ms, Some(0));
        let (_store, runtime) = config.into_parts();
        assert_eq!(runtime.observability.max_batch_events, Some(32));
        assert_eq!(
            runtime.observability.max_batch_hold_ms,
            Some(0),
            "a stated zero hold survives as a stated zero, never collapsing to absent"
        );
        Ok(())
    }

    /// The node cache budget parses in BOTH of haematite's spellings and
    /// round-trips into the store view unchanged. There is no default, so what
    /// a config states is exactly what the boot path sees.
    #[test]
    fn node_cache_budget_parses_both_spellings() -> Result<(), Box<dyn std::error::Error>> {
        let bounded = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [store]
                node_cache_budget = { bytes = 1073741824 }
            ",
        )?;
        assert_eq!(
            bounded.store.node_cache_budget,
            Some(haematite::NodeCacheBudget::bytes(1 << 30)?),
            "a 1 GiB ceiling arrives as a 1 GiB ceiling"
        );

        let unlimited = ServerConfig::from_slice(
            br#"
                [runtime]
                query_timeout_ms = 10000

                [store]
                node_cache_budget = "unlimited"
            "#,
        )?;
        assert_eq!(
            unlimited.store.node_cache_budget,
            Some(haematite::NodeCacheBudget::Unlimited),
            "`unlimited` is a stated choice, never collapsed to absent"
        );
        Ok(())
    }

    /// An omitted budget parses to `None` rather than to a guess. The refusal
    /// happens where the value is USED (the haematite boot path), which
    /// `state.rs`'s tests pin; here the point is that nothing invents one on the
    /// way through — and that `unlimited` is distinguishable from absent.
    #[test]
    fn an_omitted_node_cache_budget_stays_absent() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [store]
                shard_count = 8
            ",
        )?;
        assert_eq!(config.store.node_cache_budget, None);
        Ok(())
    }

    /// An omitted flush policy parses to `None` rather than to a guess. The
    /// refusal happens where the value is used (the publisher build), which
    /// `state.rs`'s tests pin; here the point is that nothing invents a value on
    /// the way through.
    #[test]
    fn an_omitted_flush_policy_stays_absent() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [observability]
                max_event_bytes = 512
            ",
        )?;
        assert_eq!(config.observability.max_batch_events, None);
        assert_eq!(config.observability.max_batch_hold_ms, None);
        Ok(())
    }

    #[test]
    fn zero_observability_max_event_bytes_fails_startup_validation() {
        let result = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [observability]
                max_event_bytes = 0
            ",
        );
        let message = result
            .err()
            .map_or_else(String::new, |error| error.to_string());
        assert!(
            message.contains("observability.max_event_bytes"),
            "validation message must name the zero-valued key: {message}"
        );
    }

    #[test]
    fn zero_observability_max_stream_events_fails_startup_validation() {
        let result = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [observability]
                max_stream_events = 0
            ",
        );
        let message = result
            .err()
            .map_or_else(String::new, |error| error.to_string());
        assert!(
            message.contains("observability.max_stream_events"),
            "validation message must name the zero-valued key: {message}"
        );
    }

    #[test]
    fn missing_query_timeout_uses_default() -> Result<(), Box<dyn std::error::Error>> {
        // The server unconditionally mounts /workflows/query, but the reply
        // deadline is a tuning knob: omitting it must resolve to the default and
        // boot, not fail startup.
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                scheduler_threads = 1

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64
            ",
        )?;
        assert_eq!(
            config.runtime.query_timeout_ms,
            Some(DEFAULT_QUERY_TIMEOUT_MS),
            "omitted query_timeout_ms must resolve to the default"
        );
        Ok(())
    }

    #[test]
    fn empty_config_boots_on_operational_defaults() -> Result<(), Box<dyn std::error::Error>> {
        // The headline zero-config contract: an empty TOML must parse, fill every
        // operational tuning knob with its default, and validate — so `aion
        // server` runs with no hand-authored file. The durable default backend
        // (haematite under its default data_dir) carries the store side.
        let config = ServerConfig::from_slice(b"")?;
        assert_eq!(config.store.backend, StoreBackend::Haematite);
        assert_eq!(
            config.runtime.query_timeout_ms,
            Some(DEFAULT_QUERY_TIMEOUT_MS)
        );
        assert_eq!(
            config.websocket.event_broadcast_capacity,
            Some(DEFAULT_EVENT_BROADCAST_CAPACITY)
        );
        assert_eq!(
            config.websocket.cluster_broadcast_capacity,
            Some(DEFAULT_CLUSTER_BROADCAST_CAPACITY)
        );
        Ok(())
    }

    #[test]
    fn zero_query_timeout_fails_startup_validation() {
        let result = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 0

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64
            ",
        );

        let message = result
            .err()
            .map_or_else(String::new, |error| error.to_string());
        assert!(
            message.contains("runtime.query_timeout_ms"),
            "validation message must name the zero-valued key: {message}"
        );
    }

    /// The deploy surface is commissioned explicitly: enabling it without
    /// the archive ceiling is a conservative security default, not a forced
    /// operator decision: enabling deploy without it must resolve the ceiling
    /// to [`DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES`] and boot, not fail startup.
    #[test]
    fn deploy_enabled_defaults_max_archive_bytes() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [deploy]
                enabled = true
            ",
        )?;

        assert_eq!(
            config.deploy.max_archive_bytes,
            Some(DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES),
            "omitted max_archive_bytes must resolve to the conservative default"
        );
        assert_eq!(
            config.deploy.max_inflated_bytes,
            Some(DEFAULT_DEPLOY_MAX_INFLATED_BYTES),
            "omitted max_inflated_bytes must resolve to the conservative default"
        );
        Ok(())
    }

    #[test]
    fn deploy_zero_max_archive_bytes_fails_startup_validation() {
        let result = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [deploy]
                enabled = true
                max_archive_bytes = 0
            ",
        );

        let message = result
            .err()
            .map_or_else(String::new, |error| error.to_string());
        assert!(
            message.contains("deploy.max_archive_bytes"),
            "validation message must name the zero-valued key: {message}"
        );
    }

    /// The inflate ceiling defaults independently of an explicit archive
    /// ceiling: setting only `max_archive_bytes` must resolve the inflate
    /// ceiling to [`DEFAULT_DEPLOY_MAX_INFLATED_BYTES`] (which exceeds a 16 MiB
    /// archive, so the invariant holds) and boot.
    #[test]
    fn deploy_enabled_defaults_max_inflated_bytes() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [deploy]
                enabled = true
                max_archive_bytes = 16777216
            ",
        )?;

        assert_eq!(
            config.deploy.max_archive_bytes,
            Some(16_777_216),
            "explicit max_archive_bytes must be left untouched"
        );
        assert_eq!(
            config.deploy.max_inflated_bytes,
            Some(DEFAULT_DEPLOY_MAX_INFLATED_BYTES),
            "omitted max_inflated_bytes must resolve to the conservative default"
        );
        Ok(())
    }

    #[test]
    fn deploy_zero_max_inflated_bytes_fails_startup_validation() {
        let result = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [deploy]
                enabled = true
                max_archive_bytes = 16777216
                max_inflated_bytes = 0
            ",
        );

        let message = result
            .err()
            .map_or_else(String::new, |error| error.to_string());
        assert!(
            message.contains("deploy.max_inflated_bytes"),
            "validation message must name the zero-valued key: {message}"
        );
    }

    /// An inflate ceiling below the upload ceiling is incoherent: archives
    /// the upload ceiling admits would be refused even stored uncompressed.
    #[test]
    fn deploy_max_inflated_below_max_archive_fails_startup_validation() {
        let result = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [deploy]
                enabled = true
                max_archive_bytes = 16777216
                max_inflated_bytes = 16777215
            ",
        );

        let message = result
            .err()
            .map_or_else(String::new, |error| error.to_string());
        assert!(
            message.contains("deploy.max_inflated_bytes")
                && message.contains("deploy.max_archive_bytes"),
            "validation message must name both ceilings: {message}"
        );
    }

    /// An absent `[deploy]` section means the surface stays dark and the
    /// ceilings are not required.
    #[test]
    fn deploy_disabled_requires_no_archive_ceiling() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64
            ",
        )?;

        assert!(!config.deploy.enabled);
        assert_eq!(config.deploy.max_archive_bytes, None);
        assert_eq!(config.deploy.max_inflated_bytes, None);
        Ok(())
    }

    #[test]
    fn deploy_section_parses_enabled_with_ceilings() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [deploy]
                enabled = true
                max_archive_bytes = 16777216
                max_inflated_bytes = 67108864
            ",
        )?;

        assert!(config.deploy.enabled);
        assert_eq!(config.deploy.max_archive_bytes, Some(16_777_216));
        assert_eq!(config.deploy.max_inflated_bytes, Some(67_108_864));
        Ok(())
    }

    /// With no `[server] cors_allowed_origins` the list is empty: the secure
    /// default, where no cross-origin request is permitted and no `CorsLayer`
    /// is installed.
    #[test]
    fn cors_allowed_origins_default_empty() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64
            ",
        )?;

        assert!(config.server.cors_allowed_origins.is_empty());
        let (_, runtime) = config.into_parts();
        assert!(runtime.cors_allowed_origins.is_empty());
        Ok(())
    }

    /// A configured `[server] cors_allowed_origins` list parses and round-trips
    /// into `RuntimeConfig` (the value the `CorsLayer` is built from).
    #[test]
    fn cors_allowed_origins_parse_and_round_trip() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br#"
                [server]
                cors_allowed_origins = ["http://localhost:5173", "http://127.0.0.1:5173"]

                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64
            "#,
        )?;

        assert_eq!(
            config.server.cors_allowed_origins,
            vec![
                "http://localhost:5173".to_owned(),
                "http://127.0.0.1:5173".to_owned()
            ]
        );
        let (_, runtime) = config.into_parts();
        assert_eq!(
            runtime.cors_allowed_origins,
            vec![
                "http://localhost:5173".to_owned(),
                "http://127.0.0.1:5173".to_owned()
            ]
        );
        Ok(())
    }

    /// A malformed CORS origin (no scheme, or a trailing path) can never match a
    /// browser `Origin` header, so it fails startup validation rather than
    /// silently never matching.
    #[test]
    fn cors_allowed_origins_reject_malformed() {
        for bad in ["", "localhost:5173", "http://localhost:5173/"] {
            let toml = format!(
                "[server]\ncors_allowed_origins = [\"{bad}\"]\n\n[runtime]\nquery_timeout_ms = 10000\n\n[websocket]\nevent_broadcast_capacity = 64\n"
            );
            let result = ServerConfig::from_slice(toml.as_bytes());
            let message = result
                .err()
                .map_or_else(String::new, |error| error.to_string());
            assert!(
                message.contains("cors_allowed_origins"),
                "malformed origin `{bad}` must be rejected naming the key: {message}"
            );
        }
    }

    /// An absent `[dev]` section leaves the dev surface dark.
    #[test]
    fn dev_absent_leaves_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64
            ",
        )?;

        assert!(!config.dev.enabled);
        Ok(())
    }

    /// `[dev] enabled = true` commissions the dev surface; it adds no other
    /// knobs (ADR-001: the only setting is the on/off gate).
    #[test]
    fn dev_section_parses_enabled() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [dev]
                enabled = true
            ",
        )?;

        assert!(config.dev.enabled);
        Ok(())
    }

    /// An absent `[authoring]` section commissions the AWL workspace while the
    /// separate Gleam loop remains dark and requires neither of its paths.
    #[test]
    fn authoring_absent_defaults_awl_workspace_but_keeps_gleam_dark()
    -> Result<(), Box<dyn std::error::Error>> {
        let home = crate::test_support::private_tempdir()?;
        let config = ServerConfig::from_slice_with_home(b"", home.path())?;

        assert_eq!(config.authoring.gleam_path, None);
        assert_eq!(config.authoring.project_root, None);
        assert_eq!(
            config.authoring.workspace_dir.as_deref(),
            Some(home.path().join(DEFAULT_AUTHORING_WORKSPACE_DIR).as_path())
        );
        Ok(())
    }

    /// An explicit `workspace_dir` wins over the out-of-box relative default.
    #[test]
    fn authoring_explicit_workspace_is_honored() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br#"
                [authoring]
                workspace_dir = "/srv/aion/studio"
            "#,
        )?;

        assert_eq!(
            config.authoring.workspace_dir.as_deref(),
            Some(std::path::Path::new("/srv/aion/studio"))
        );
        Ok(())
    }

    /// A configured `[authoring]` section with both `gleam_path` and
    /// `project_root` parses and round-trips into `RuntimeConfig`.
    #[test]
    fn authoring_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br#"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [authoring]
                gleam_path = "/usr/local/bin/gleam"
                project_root = "/srv/aion/authoring"
            "#,
        )?;

        assert_eq!(
            config.authoring.gleam_path.as_deref(),
            Some(std::path::Path::new("/usr/local/bin/gleam"))
        );
        let (_, runtime) = config.into_parts();
        assert_eq!(
            runtime.authoring.gleam_path.as_deref(),
            Some(std::path::Path::new("/usr/local/bin/gleam"))
        );
        assert_eq!(
            runtime.authoring.project_root.as_deref(),
            Some(std::path::Path::new("/srv/aion/authoring"))
        );
        Ok(())
    }

    /// Commissioning the authoring loop (a `gleam_path`) without a
    /// `project_root` must fail startup naming the key and the environment
    /// override (the deploy required-config pattern).
    #[test]
    fn authoring_gleam_path_without_project_root_fails_naming_key_and_env() {
        let result = ServerConfig::from_slice(
            br#"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [authoring]
                gleam_path = "/usr/local/bin/gleam"
            "#,
        );

        let message = result
            .err()
            .map_or_else(String::new, |error| error.to_string());
        assert!(
            message.contains("authoring.project_root"),
            "validation message must name the missing key: {message}"
        );
        assert!(
            message.contains("AION_AUTHORING_PROJECT_ROOT"),
            "validation message must name the environment override: {message}"
        );
    }

    /// An empty `gleam_path` is a misconfiguration, not "dark": it must fail
    /// startup naming the key and the environment override.
    #[test]
    fn authoring_empty_gleam_path_fails_naming_key_and_env() {
        let result = ServerConfig::from_slice(
            br#"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [authoring]
                gleam_path = ""
            "#,
        );

        let message = result
            .err()
            .map_or_else(String::new, |error| error.to_string());
        assert!(
            message.contains("authoring.gleam_path"),
            "validation message must name the empty key: {message}"
        );
        assert!(
            message.contains("AION_AUTHORING_GLEAM_PATH"),
            "validation message must name the environment override: {message}"
        );
    }

    /// CLI overrides commission the authoring loop after file/env merge.
    #[test]
    fn cli_overrides_set_authoring_paths() -> Result<(), Box<dyn std::error::Error>> {
        let mut config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64
            ",
        )?;
        let cli = CliOverrides {
            gleam_path: Some(std::path::PathBuf::from("/opt/gleam")),
            authoring_project_root: Some(std::path::PathBuf::from("/opt/project")),
            ..CliOverrides::default()
        };

        config.apply_cli_overrides(&cli);
        config.validate()?;

        assert_eq!(
            config.authoring.gleam_path.as_deref(),
            Some(std::path::Path::new("/opt/gleam"))
        );
        assert_eq!(
            config.authoring.project_root.as_deref(),
            Some(std::path::Path::new("/opt/project"))
        );
        Ok(())
    }

    /// The config field renamed `dashboard` -> `ops_console` carries a serde
    /// alias so existing `[dashboard]` TOML still parses (non-breaking rename).
    /// R1: the written queue-service settings reach `RuntimeConfig`, and an
    /// operator who writes nothing gets `strict` with no clocks.
    #[test]
    fn queue_service_settings_are_read_from_the_worker_section()
    -> Result<(), Box<dyn std::error::Error>> {
        use crate::worker::QueueServicePolicy;
        use std::time::Duration;

        let bare = ServerConfig::from_slice(
            br"
                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64
            ",
        )?;
        assert_eq!(
            bare.worker.queue_service.default_policy,
            QueueServicePolicy::Strict,
            "strict is the default with nothing written"
        );
        assert_eq!(
            bare.worker.queue_service.service_availability_deadline,
            None
        );
        assert_eq!(bare.worker.queue_service.schedule_to_start_timeout, None);

        let written = ServerConfig::from_slice(
            br#"
                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [worker.queue_service]
                service_availability_deadline = 45000
                schedule_to_start_timeout = 5000

                [[worker.queue_service.overrides]]
                task_queue = "general"
                policy = "durable_pending"
            "#,
        )?;
        assert_eq!(
            written.worker.queue_service.service_availability_deadline,
            Some(Duration::from_secs(45))
        );
        assert_eq!(
            written.worker.queue_service.schedule_to_start_timeout,
            Some(Duration::from_secs(5))
        );
        assert_eq!(
            written
                .worker
                .queue_service
                .policy_for("default", "general"),
            QueueServicePolicy::DurablePending,
            "the written opt-in must reach the dispatch seam"
        );
        assert_eq!(
            written
                .worker
                .queue_service
                .policy_for("default", "billing"),
            QueueServicePolicy::Strict,
            "an override must not leak onto other queues"
        );

        // The same settings survive the runtime projection the server boots from.
        let (_store, runtime) = written.into_parts();
        assert_eq!(
            runtime
                .worker
                .queue_service
                .policy_for("default", "general"),
            QueueServicePolicy::DurablePending
        );
        Ok(())
    }

    #[test]
    fn legacy_dashboard_section_alias_still_parses() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br#"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [dashboard]
                source = { FileSystem = { asset_path = "/srv/aion/ui" } }
            "#,
        )?;
        match &config.ops_console.source {
            OpsConsoleAssetSource::FileSystem { asset_path } => {
                assert_eq!(asset_path.as_os_str(), "/srv/aion/ui");
            }
            OpsConsoleAssetSource::Embedded => {
                return Err("legacy [dashboard] section must map to ops_console".into());
            }
        }
        Ok(())
    }

    /// The new `[ops_console]` section name also parses.
    #[test]
    fn ops_console_section_parses() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br#"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [ops_console]
                source = { FileSystem = { asset_path = "/srv/aion/ui" } }
            "#,
        )?;
        assert!(matches!(
            config.ops_console.source,
            OpsConsoleAssetSource::FileSystem { .. }
        ));
        Ok(())
    }

    #[test]
    fn invalid_values_name_problematic_field() {
        let result = ServerConfig::from_slice(
            br"
                [runtime]
                scheduler_threads = 0
            ",
        );

        let message = result
            .err()
            .map_or_else(String::new, |error| error.to_string());
        assert!(message.contains("runtime.scheduler_threads"));
    }

    const RETIRED_STORE_REMEDY: &str = "backend = \"haematite\" with store.data_dir and store.node_cache_budget (all three are required — haematite refuses to start without the budget); existing libsql database files are not read by this build";

    fn assert_retired_store_refusal(
        result: Result<ServerConfig, crate::error::ServerError>,
        found: &str,
    ) {
        assert!(result.is_err(), "retired libsql input must be refused");
        let message = result
            .err()
            .map_or_else(String::new, |error| error.to_string());
        assert!(
            message.contains(found),
            "refusal did not name `{found}`: {message}"
        );
        assert!(
            message.contains(RETIRED_STORE_REMEDY),
            "refusal omitted the operator remedy: {message}"
        );
    }

    #[test]
    fn retired_libsql_backend_is_refused_with_remedy() {
        assert_retired_store_refusal(
            ServerConfig::from_slice(
                br#"
                    [store]
                    backend = "libsql"
                "#,
            ),
            "backend = \"libsql\"",
        );
    }

    #[test]
    fn retired_store_url_key_is_refused_with_remedy() {
        assert_retired_store_refusal(
            ServerConfig::from_slice(
                br#"
                    [store]
                    backend = "haematite"
                    url = "old.db"
                "#,
            ),
            "store.url",
        );
    }

    /// The FIFTH door: `AION_STORE_BACKEND=libsql` names the retired backend as
    /// squarely as `backend = "libsql"` in the file does, and must reach the same
    /// named refusal rather than the parser's "must be one of" list, which names
    /// no remedy and reads like a typo.
    #[test]
    fn retired_libsql_backend_environment_is_refused_with_remedy() {
        let mut config = ServerConfig::default();
        let result = super::env::overlay_vars(
            &mut config,
            [("AION_STORE_BACKEND".to_owned(), "libsql".to_owned())],
        )
        .and_then(|()| config.validate().map(|()| config));
        assert_retired_store_refusal(result, "AION_STORE_BACKEND=libsql");
    }

    #[test]
    fn retired_store_url_environment_is_refused_with_remedy() {
        let mut config = ServerConfig::default();
        let result = super::env::overlay_vars(
            &mut config,
            [("AION_STORE_URL".to_owned(), "old.db".to_owned())],
        )
        .and_then(|()| config.validate().map(|()| config));
        assert_retired_store_refusal(result, "AION_STORE_URL");
    }

    #[test]
    fn retired_store_url_flag_is_refused_with_remedy() {
        let mut config = ServerConfig::default();
        config.apply_cli_overrides(&CliOverrides {
            store_url: Some("old.db".to_owned()),
            ..CliOverrides::default()
        });
        let result = config.validate().map(|()| config);
        assert_retired_store_refusal(result, "--store-url");
    }

    /// A CLI flag beats the value the config file loaded. Only the `--store-url`
    /// arm of the original pin died with libSQL; the precedence claim itself is
    /// backend-agnostic and is the reason `apply_cli_overrides` exists.
    #[test]
    fn cli_overrides_win_over_loaded_values() -> Result<(), Box<dyn std::error::Error>> {
        let mut config = ServerConfig::from_slice(
            br#"
                [store]
                backend = "haematite"
                data_dir = "from-the-file"

                [runtime]
                scheduler_threads = 1
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [observability]
                max_batch_events = 64
                max_batch_hold_ms = 0
            "#,
        )?;
        assert_eq!(
            config.runtime.scheduler_threads, 1,
            "the file's value must be what the flag then beats"
        );
        let cli = CliOverrides {
            scheduler_threads: Some(
                std::num::NonZeroUsize::new(3)
                    .ok_or("3 is not zero")?
                    .into(),
            ),
            ..CliOverrides::default()
        };

        config.apply_cli_overrides(&cli);
        config.validate()?;

        assert_eq!(config.runtime.scheduler_threads, 3);
        assert_eq!(
            config.store.data_dir.as_deref(),
            Some("from-the-file"),
            "a value the CLI did not override keeps the file's value"
        );
        Ok(())
    }

    #[test]
    fn haematite_store_config_remains_accepted() -> Result<(), Box<dyn std::error::Error>> {
        let config = ServerConfig::from_slice(
            br#"
                [store]
                backend = "haematite"
                data_dir = "aion-data"
            "#,
        )?;
        assert_eq!(config.store.backend, StoreBackend::Haematite);
        assert_eq!(config.store.data_dir.as_deref(), Some("aion-data"));
        Ok(())
    }

    #[test]
    fn default_config_defaults() -> Result<(), Box<dyn std::error::Error>> {
        let mut config = ServerConfig::default();

        // Raw serde defaults defer user-level paths until the merged loader has
        // resolved Aion home and applied every operator override.
        assert_eq!(config.store.backend, StoreBackend::Haematite);
        assert_eq!(config.store.data_dir, None);
        // 64 pending #187: 4096 defeated its own lazy-materialization premise
        // (boot scan_prefix materializes all shards; commit then fsyncs per
        // shard and blows the actor timeout). See StoreConfig::default().
        assert_eq!(config.store.shard_count, 64);
        assert_eq!(config.server.grpc_address.to_string(), "127.0.0.1:50051");
        assert_eq!(config.server.listen_address.to_string(), "127.0.0.1:8080");
        assert_eq!(config.namespaces.default, "default");
        // Minted-on-use is OPEN by default to preserve the zero-config,
        // no-pre-provision model: a namespace comes into being on first
        // worker reference.
        assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
        // The cluster-wide in-flight ceiling defaults to the generous platform
        // headroom value (P2-Q1); nothing enforces it yet.
        assert_eq!(
            config.namespaces.max_in_flight_activities,
            DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
        );
        assert_eq!(config.namespaces.max_in_flight_activities, 1024);
        assert!(!config.auth.enabled);
        assert!(config.metrics.enabled);
        // event_broadcast_capacity and query_timeout_ms are the deliberately
        // defaultless values: defaults validate only once the operator
        // supplies them.
        assert_eq!(config.websocket.event_broadcast_capacity, None);
        assert_eq!(config.websocket.cluster_broadcast_capacity, None);
        assert_eq!(config.runtime.query_timeout_ms, None);
        config.websocket.event_broadcast_capacity = Some(64);
        config.websocket.cluster_broadcast_capacity = Some(64);
        config.runtime.query_timeout_ms = Some(10_000);
        let home = crate::test_support::private_tempdir()?;
        let working_dir = crate::test_support::private_tempdir()?;
        super::fill_home_defaults(
            &mut config,
            home.path(),
            HomeSource::Derived,
            working_dir.path(),
        )?;
        assert_eq!(
            config.store.data_dir.as_deref(),
            home.path().join("data").to_str()
        );
        config.validate()?;
        Ok(())
    }

    #[test]
    fn outbox_is_disabled_by_default_and_needs_no_knobs() -> Result<(), Box<dyn std::error::Error>>
    {
        let mut config = ServerConfig::default();
        config.store.data_dir = Some("test-data".to_owned());
        config.websocket.event_broadcast_capacity = Some(64);
        config.websocket.cluster_broadcast_capacity = Some(64);
        config.runtime.query_timeout_ms = Some(10_000);

        // The dispatcher is dark by default and its operational knobs are all
        // absent — yet validation passes, because a disabled dispatcher never
        // reads them (no assumed defaults behind the gate).
        assert!(!config.outbox.enabled);
        assert_eq!(config.outbox.poll_interval_ms, None);
        assert_eq!(config.outbox.batch_size, None);
        assert_eq!(config.outbox.max_attempts, None);
        assert_eq!(config.outbox.backoff_base_ms, None);
        assert_eq!(config.outbox.backoff_multiplier, None);
        assert_eq!(config.outbox.backoff_max_ms, None);
        assert_eq!(config.outbox.reconcile_interval_ms, None);
        assert_eq!(config.outbox.reconcile_stale_after_ms, None);
        config.validate()?;
        Ok(())
    }

    fn outbox_enabled_base() -> ServerConfig {
        let mut config = ServerConfig::default();
        config.store.data_dir = Some("test-data".to_owned());
        config.websocket.event_broadcast_capacity = Some(64);
        config.websocket.cluster_broadcast_capacity = Some(64);
        config.runtime.query_timeout_ms = Some(10_000);
        config.outbox.enabled = true;
        config.outbox.poll_interval_ms = Some(250);
        config.outbox.batch_size = Some(64);
        config.outbox.max_attempts = Some(5);
        config.outbox.backoff_base_ms = Some(100);
        config.outbox.backoff_multiplier = Some(2);
        config.outbox.backoff_max_ms = Some(30_000);
        config.outbox.reconcile_interval_ms = Some(1_000);
        config.outbox.reconcile_stale_after_ms = Some(60_000);
        config
    }

    #[test]
    fn outbox_enabled_with_all_knobs_validates() -> Result<(), Box<dyn std::error::Error>> {
        outbox_enabled_base().validate()?;
        Ok(())
    }

    #[test]
    fn outbox_enabled_defaults_poll_interval() -> Result<(), Box<dyn std::error::Error>> {
        // Enabling the dispatcher but omitting the poll cadence must resolve it
        // to the default and boot, not fail startup — the cadence is pure
        // tuning, not a forced operator decision.
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [outbox]
                enabled = true
            ",
        )?;
        assert_eq!(
            config.outbox.poll_interval_ms,
            Some(DEFAULT_OUTBOX_POLL_INTERVAL_MS),
            "omitted poll_interval_ms must resolve to the default"
        );
        Ok(())
    }

    #[test]
    fn outbox_enabled_defaults_max_attempts() -> Result<(), Box<dyn std::error::Error>> {
        // Setting a tuning knob explicitly but omitting the retry budget must
        // leave the explicit knob untouched and default only the omitted one.
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [outbox]
                enabled = true
                poll_interval_ms = 250
            ",
        )?;
        assert_eq!(
            config.outbox.poll_interval_ms,
            Some(250),
            "explicit poll_interval_ms must be left untouched"
        );
        assert_eq!(
            config.outbox.max_attempts,
            Some(DEFAULT_OUTBOX_MAX_ATTEMPTS),
            "omitted max_attempts must resolve to the default"
        );
        Ok(())
    }

    #[test]
    fn outbox_enabled_with_only_enabled_flag_uses_all_defaults()
    -> Result<(), Box<dyn std::error::Error>> {
        // Headline conditional-default contract: an outbox section with nothing
        // but `enabled = true` validates with every tuning knob resolved to its
        // default. The reconciliation pair stays dark (both absent), as before.
        let config = ServerConfig::from_slice(
            br"
                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64

                [outbox]
                enabled = true
            ",
        )?;
        assert!(config.outbox.enabled);
        assert_eq!(
            config.outbox.poll_interval_ms,
            Some(DEFAULT_OUTBOX_POLL_INTERVAL_MS)
        );
        assert_eq!(config.outbox.batch_size, Some(DEFAULT_OUTBOX_BATCH_SIZE));
        assert_eq!(
            config.outbox.max_attempts,
            Some(DEFAULT_OUTBOX_MAX_ATTEMPTS)
        );
        assert_eq!(
            config.outbox.backoff_base_ms,
            Some(DEFAULT_OUTBOX_BACKOFF_BASE_MS)
        );
        assert_eq!(
            config.outbox.backoff_multiplier,
            Some(DEFAULT_OUTBOX_BACKOFF_MULTIPLIER)
        );
        assert_eq!(
            config.outbox.backoff_max_ms,
            Some(DEFAULT_OUTBOX_BACKOFF_MAX_MS)
        );
        // Reconciliation is not force-defaulted: both knobs stay absent so the
        // live sweep remains dark.
        assert_eq!(config.outbox.reconcile_interval_ms, None);
        assert_eq!(config.outbox.reconcile_stale_after_ms, None);
        Ok(())
    }

    #[test]
    fn outbox_enabled_zero_poll_interval_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
        // An explicit zero is a misconfiguration the default never masks:
        // `get_or_insert` leaves `Some(0)` untouched and validate rejects it.
        let mut config = outbox_enabled_base();
        config.outbox.poll_interval_ms = Some(0);
        let error = config
            .validate()
            .err()
            .ok_or("enabled outbox with zero poll interval must fail")?;
        assert!(
            error.to_string().contains("outbox.poll_interval_ms"),
            "error must name the zero-valued key: {error}"
        );
        Ok(())
    }

    #[test]
    fn outbox_enabled_zero_max_attempts_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
        let mut config = outbox_enabled_base();
        config.outbox.max_attempts = Some(0);
        let error = config
            .validate()
            .err()
            .ok_or("enabled outbox with zero max attempts must fail")?;
        assert!(
            error.to_string().contains("outbox.max_attempts"),
            "error must name the zero-valued key: {error}"
        );
        Ok(())
    }

    #[test]
    fn outbox_backoff_max_below_base_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
        let mut config = outbox_enabled_base();
        config.outbox.backoff_base_ms = Some(1_000);
        config.outbox.backoff_max_ms = Some(500);
        let error = config
            .validate()
            .err()
            .ok_or("backoff_max below backoff_base must fail")?;
        assert!(
            error.to_string().contains("outbox.backoff_max_ms"),
            "error must name the offending key: {error}"
        );
        Ok(())
    }

    #[test]
    fn outbox_enabled_can_leave_reconciliation_dark() -> Result<(), Box<dyn std::error::Error>> {
        let mut config = outbox_enabled_base();
        config.outbox.reconcile_interval_ms = None;
        config.outbox.reconcile_stale_after_ms = None;
        config.validate()?;
        Ok(())
    }

    #[test]
    fn outbox_reconciliation_requires_interval_when_partially_enabled()
    -> Result<(), Box<dyn std::error::Error>> {
        let mut config = outbox_enabled_base();
        config.outbox.reconcile_interval_ms = None;
        let error = config
            .validate()
            .err()
            .ok_or("reconciliation without interval must fail")?;
        assert!(error.to_string().contains("outbox.reconcile_interval_ms"));
        Ok(())
    }

    #[test]
    fn outbox_reconciliation_requires_stale_threshold_when_partially_enabled()
    -> Result<(), Box<dyn std::error::Error>> {
        let mut config = outbox_enabled_base();
        config.outbox.reconcile_stale_after_ms = None;
        let error = config
            .validate()
            .err()
            .ok_or("reconciliation without stale threshold must fail")?;
        assert!(
            error
                .to_string()
                .contains("outbox.reconcile_stale_after_ms")
        );
        Ok(())
    }

    #[test]
    fn package_discovery_is_sorted() -> Result<(), Box<dyn std::error::Error>> {
        let temp_dir = crate::test_support::private_tempdir()?;
        std::fs::write(temp_dir.path().join("zeta.aion"), b"package")?;
        std::fs::write(temp_dir.path().join("alpha.aion"), b"package")?;
        std::fs::write(temp_dir.path().join("ignored.txt"), b"package")?;
        std::fs::create_dir(temp_dir.path().join("nested"))?;
        std::fs::write(
            temp_dir.path().join("nested").join("nested.aion"),
            b"package",
        )?;

        let packages = discover_workflow_packages(temp_dir.path())?;

        assert_eq!(
            packages,
            vec![
                temp_dir.path().join("alpha.aion"),
                temp_dir.path().join("zeta.aion"),
            ]
        );
        Ok(())
    }

    #[test]
    fn workflow_package_merge_is_additive_and_deduplicated() {
        let mut packages = vec!["config.aion".into(), "shared.aion".into()];
        let discovered = vec!["auto.aion".into(), "shared.aion".into()];
        let cli = vec!["cli.aion".into(), "auto.aion".into()];

        merge_workflow_packages(&mut packages, discovered, &cli);

        assert_eq!(
            packages,
            vec![
                std::path::PathBuf::from("config.aion"),
                std::path::PathBuf::from("shared.aion"),
                std::path::PathBuf::from("auto.aion"),
                std::path::PathBuf::from("cli.aion"),
            ]
        );
    }

    #[test]
    fn package_merge_deduplicates_canonical_files() -> Result<(), Box<dyn std::error::Error>> {
        let temp_dir = crate::test_support::private_tempdir()?;
        let package = temp_dir.path().join("hello.aion");
        std::fs::write(&package, b"package")?;
        let mut packages = vec![package.clone()];
        let discovered = vec![temp_dir.path().join(".").join("hello.aion")];

        merge_workflow_packages(&mut packages, discovered, &[]);

        assert_eq!(packages, vec![package]);
        Ok(())
    }

    #[test]
    fn zero_config_cli_workflow_package_uses_in_memory_defaults()
    -> Result<(), Box<dyn std::error::Error>> {
        let temp_dir = crate::test_support::private_tempdir()?;

        let cli = CliOverrides {
            workflow_packages: vec!["hello-world.aion".into()],
            ..CliOverrides::default()
        };
        let mut config = ServerConfig::default();
        // This test exercises CLI workflow-package discovery against the ephemeral
        // in-memory store, so it opts OUT of the new durable haematite default
        // explicitly (the default would otherwise carry a haematite data_dir).
        config.store.backend = StoreBackend::Memory;
        config.store.data_dir = None;
        // Even zero-config development runs must size event streaming and the
        // query reply deadline explicitly (config keys or the
        // AION_WEBSOCKET_EVENT_BROADCAST_CAPACITY /
        // AION_RUNTIME_QUERY_TIMEOUT_MS environment overrides).
        config.websocket.event_broadcast_capacity = Some(64);
        config.websocket.cluster_broadcast_capacity = Some(64);
        config.runtime.query_timeout_ms = Some(10_000);
        config.load_discovered_workflow_packages(&cli, temp_dir.path())?;

        config.validate()?;

        assert_eq!(config.store.backend, StoreBackend::Memory);
        assert_eq!(
            config.workflow_packages,
            vec![std::path::PathBuf::from("hello-world.aion")]
        );
        Ok(())
    }

    #[test]
    fn cli_packages_are_additive() -> Result<(), Box<dyn std::error::Error>> {
        let mut config = ServerConfig::from_slice(
            br#"
                workflow_packages = ["config.aion"]

                [runtime]
                query_timeout_ms = 10000

                [websocket]
                event_broadcast_capacity = 64
                cluster_broadcast_capacity = 64
            "#,
        )?;
        let cli = CliOverrides {
            workflow_packages: vec!["cli-one.aion".into(), "cli-two.aion".into()],
            ..CliOverrides::default()
        };

        merge_workflow_packages(
            &mut config.workflow_packages,
            Vec::new(),
            &cli.workflow_packages,
        );

        assert_eq!(
            config.workflow_packages,
            vec![
                std::path::PathBuf::from("config.aion"),
                std::path::PathBuf::from("cli-one.aion"),
                std::path::PathBuf::from("cli-two.aion"),
            ]
        );
        Ok(())
    }
}