x0x 0.32.0

//! Daemon configuration, server handle, and shared application state.
//!
//! Extracted from `server/mod.rs` (#125 / WS1.4) as a mechanical move.
//! Public API items are re-exported from the parent module; `AppState` and
//! the internal config/cache types are `pub(super)` — internal to `server`.

use std::collections::{BTreeMap, HashMap, VecDeque};
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex as StdMutex;
use std::time::{Duration, Instant};

use axum::http::StatusCode;
use serde::{Deserialize, Serialize};
use sha2::Sha256;
use tokio::sync::{broadcast, mpsc, watch, Mutex, RwLock};

use crate as x0x;
use crate::contacts::ContactStore;
use crate::{Agent, KvStoreHandle, TaskListHandle};

// Local types that stay in `mod.rs` (groups/files domain). A child module can
// name private items of its parent, so no `pub(super)` is needed on them —
// they are imported here and claimed by their own submodules later.
use super::auth::SessionStore;
use super::sse::SseEvent;
use super::ws::{SharedTopicState, WsOutboundStats, WsSession};
use super::{
    ExpectedJoinResultInviter, FileChunkAckSlot, NamedGroupMetadataEvent, PendingJoinResult,
    PendingTreeKemMetadataEvent, PendingWelcome, PendingWelcomeReceive, RestSubscription,
    WelcomeFetchWaiter,
};

fn validate_instance_name_grammar(name: &str) -> anyhow::Result<()> {
    if name.is_empty() || name.len() > 64 {
        anyhow::bail!("instance name must be 1-64 characters");
    }
    let valid = name
        .chars()
        .next()
        .is_some_and(|c| c.is_ascii_alphanumeric())
        && name.chars().all(|c| c.is_ascii_alphanumeric() || c == '-');
    if !valid {
        anyhow::bail!(
            "instance name must start with alphanumeric and contain only alphanumeric or hyphens"
        );
    }
    Ok(())
}

/// Validate a daemon instance name without taking ownership or allocating on
/// the success path.
///
/// New path-derivation code should construct [`InstanceName`] so invalid state
/// is unrepresentable. This function remains the stable borrowed validation API.
pub fn validate_instance_name(name: &str) -> anyhow::Result<()> {
    validate_instance_name_grammar(name)
}

/// Validated daemon instance name used for every instance-scoped path.
///
/// Construction enforces the shared CLI/config grammar, so path derivation
/// cannot receive separators, traversal components, or empty names.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InstanceName(String);

impl InstanceName {
    /// Borrow the validated instance name.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }

    /// Consume the validated name and return its owned representation.
    #[must_use]
    pub fn into_string(self) -> String {
        self.0
    }
}

impl TryFrom<String> for InstanceName {
    type Error = anyhow::Error;

    fn try_from(name: String) -> Result<Self, Self::Error> {
        validate_instance_name_grammar(&name)?;
        Ok(Self(name))
    }
}

/// Carries the CLI-derived flags that the server-bringup path consumes.
/// Phase 1: minimal — do not redesign config here.
#[derive(Default)]
pub struct ServeOptions {
    /// Skip the startup GitHub update check.
    pub skip_update_check: bool,
    /// Disable ant-quic UPnP IGD port mapping for this invocation.
    pub cli_no_port_mapping: bool,
    /// Do not load or save the cached peer set.
    pub cli_disable_peer_cache: bool,
    /// Active instance name (`--name`), if any.
    pub instance_name: Option<String>,
    /// Loaded exec ACL policy.
    pub exec_policy: x0x::exec::ExecPolicy,
    /// Loaded connect ACL policy.
    ///
    /// `Default` is [`x0x::connect::ConnectPolicy::Disabled`], so embedders
    /// that build `ServeOptions` without supplying a connect ACL get
    /// default-deny for free.
    pub connect_policy: x0x::connect::ConnectPolicy,
    /// Whether the self-update install/restart paths are allowed to run.
    ///
    /// AND-ed with `config.update.enabled`. The daemon binary sets this to
    /// `config.update.enabled` so its behaviour is unchanged. The public
    /// [`crate::server::serve`] entrypoint defaults it to `false` — an embedded library must
    /// never replace or restart the host application. Manifest *propagation*
    /// (broadcast/listen for informational purposes) is unaffected; only the
    /// paths that download + install + restart are gated.
    pub self_update_enabled: bool,
}

/// Handle to a running, in-process x0x server.
///
/// Returned by [`crate::server::serve`] / [`crate::server::serve_with_options`]. The server runs on a
/// detached supervisor task; the handle owns its lifecycle. All synchronous,
/// fallible startup (data-dir create, identity load/gen, listener bind, state
/// and router build, `api.port` write) has already completed by the time the
/// handle is returned, so [`local_addr`](ServerHandle::local_addr) is readable
/// immediately, which matters when binding `127.0.0.1:0` for tests.
///
/// Dropping the handle requests shutdown (the supervisor is cancelled) but does
/// not block; await [`wait`](ServerHandle::wait) or
/// [`shutdown_and_wait`](ServerHandle::shutdown_and_wait) to observe completion.
pub struct ServerHandle {
    pub(super) local_addr: SocketAddr,
    pub(super) cancel: tokio_util::sync::CancellationToken,
    // `Option` so the consuming `wait`/`shutdown_and_wait` can take the join
    // handle out without conflicting with the `Drop` impl (which only cancels).
    pub(super) task: Option<tokio::task::JoinHandle<anyhow::Result<()>>>,
}

impl ServerHandle {
    /// The actual bound API address. Readable immediately after the handle is
    /// returned, including the resolved port when the caller bound to port 0.
    #[must_use]
    pub fn local_addr(&self) -> SocketAddr {
        self.local_addr
    }

    /// Request graceful shutdown. Idempotent and non-consuming — safe to call
    /// repeatedly and from a `&self` reference. Returns immediately; await the
    /// handle to observe run-to-completion.
    pub fn shutdown(&self) {
        self.cancel.cancel();
    }

    /// Await the server's run-to-completion, returning its supervisor result.
    pub async fn wait(mut self) -> anyhow::Result<()> {
        // `task` is always `Some` here — it is only taken by this consuming
        // method, so a single `wait`/`shutdown_and_wait` call sees it set.
        let Some(task) = self.task.take() else {
            return Err(anyhow::anyhow!("server handle already consumed"));
        };
        match task.await {
            Ok(res) => res,
            Err(e) => Err(anyhow::Error::new(e).context("server supervisor task failed")),
        }
    }

    /// A clone of the cancellation token that drives shutdown. Lets a host
    /// `select!` over its own signal handling and cancel without holding the
    /// handle (the daemon binary uses this for Ctrl-C). Cancelling the returned
    /// token is equivalent to calling [`shutdown`](ServerHandle::shutdown).
    #[must_use]
    pub fn cancellation_token(&self) -> tokio_util::sync::CancellationToken {
        self.cancel.clone()
    }

    /// Request shutdown, then await run-to-completion.
    pub async fn shutdown_and_wait(self) -> anyhow::Result<()> {
        self.cancel.cancel();
        self.wait().await
    }
}

impl Drop for ServerHandle {
    fn drop(&mut self) {
        // No detached daemon: dropping the handle requests shutdown. Drop does
        // not block — callers that need to observe completion use `wait`.
        self.cancel.cancel();
    }
}

// ---------------------------------------------------------------------------
// Configuration
// ---------------------------------------------------------------------------

/// Daemon configuration loaded from TOML.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DaemonConfig {
    /// QUIC bind address for gossip (default 0.0.0.0:0 = random).
    #[serde(default = "default_bind_address")]
    pub bind_address: SocketAddr,

    /// HTTP API address (default 127.0.0.1:12700).
    #[serde(default = "default_api_address")]
    pub api_address: SocketAddr,

    /// Data directory for persistent storage.
    #[serde(default = "default_data_dir")]
    pub data_dir: PathBuf,

    /// Log level (trace, debug, info, warn, error).
    #[serde(default = "default_log_level")]
    pub log_level: String,

    /// Log format ("text" or "json").
    #[serde(default = "default_log_format")]
    pub log_format: String,

    /// Bootstrap peers to connect on startup.
    ///
    /// `None` (the TOML key absent, or `DaemonConfig::default()`) resolves to
    /// the hardcoded global bootstrap network via [`Self::resolved_bootstrap_peers`].
    /// `Some(vec)` honors the operator's explicit list verbatim — including an
    /// explicit `bootstrap_peers = []`, which means "no seed peers at all".
    ///
    /// This three-valued distinction lets `--no-hard-coded-bootstrap` clear
    /// *only* the embedded fallback (`None` → `Some([])`) without clobbering
    /// peers an operator deliberately listed in their config file.
    #[serde(default)]
    pub bootstrap_peers: Option<Vec<SocketAddr>>,

    /// X0X-0062 reviewer P2 #2: enable or disable ant-quic's best-effort
    /// UPnP IGD port-mapping. Default `true` (matches ant-quic). Set to
    /// `false` in the daemon TOML (`port_mapping_enabled = false`) or via
    /// the `--no-port-mapping` CLI flag on networks without IGD support
    /// or where unsolicited router port mappings are policy-forbidden.
    #[serde(default = "default_port_mapping_enabled")]
    pub(super) port_mapping_enabled: bool,

    /// X0X-0070b: peer-relay fallback configuration (TOML `[peer_relay]`).
    /// Defaults to disabled — opt in by setting `peer_relay.enabled = true`
    /// and listing relay-candidate hex agent IDs under
    /// `peer_relay.candidates`. The relay path only activates when a direct
    /// DM crosses the failure threshold; the happy path allocates nothing
    /// extra.
    #[serde(default)]
    pub(super) peer_relay: x0x::network::PeerRelayConfig,

    /// Update configuration.
    #[serde(default)]
    pub(super) update: DaemonUpdateConfig,

    /// Gossip overlay configuration (TOML: `[gossip]`).
    #[serde(default)]
    pub gossip: x0x::gossip::GossipConfig,

    /// How often to re-announce identity (seconds).
    #[serde(default = "default_heartbeat_interval")]
    pub(super) heartbeat_interval_secs: u64,

    /// How long before a discovered agent entry is considered stale (seconds).
    #[serde(default = "default_identity_ttl")]
    pub(super) identity_ttl_secs: u64,

    /// Optional path to a user keypair file for human identity.
    /// When set, the agent can announce with `include_user_identity: true`.
    #[serde(default)]
    pub(super) user_key_path: Option<PathBuf>,

    /// Enable rendezvous `ProviderSummary` advertisements for global findability.
    #[serde(default = "default_rendezvous_enabled")]
    pub(super) rendezvous_enabled: bool,

    /// Validity period (milliseconds) for each rendezvous advertisement.
    /// The daemon re-advertises every `validity_ms / 2` so that the record
    /// is always fresh before it expires.
    #[serde(default = "default_rendezvous_validity_ms")]
    pub(super) rendezvous_validity_ms: u64,

    /// Override the presence beacon interval (seconds) for tests / embeddings.
    #[serde(default)]
    pub(super) presence_beacon_interval_secs: Option<u64>,

    /// Override the presence event poll interval (seconds) for tests / embeddings.
    #[serde(default)]
    pub(super) presence_event_poll_interval_secs: Option<u64>,

    /// Override the fallback offline timeout used by presence events (seconds).
    #[serde(default)]
    pub(super) presence_offline_timeout_secs: Option<u64>,

    /// Instance name for multi-agent support.
    /// When set, identity and data are scoped to this name.
    #[serde(default)]
    pub instance_name: Option<String>,

    /// Explicit directory for identity material (machine/agent/user/cert keys).
    ///
    /// When set, ALL identity keys derive from this directory and the daemon
    /// never falls back to `~/.x0x`. This is the storage boundary required for
    /// in-process embedding ([`crate::server::serve`]): the host supplies its own directory so
    /// nothing is written under the user's home. When unset (the default for
    /// the daemon binary), identity falls back to the existing behaviour
    /// (`~/.x0x`, or the `--name`-scoped `~/.x0x-<name>` directory).
    #[serde(default)]
    pub identity_dir: Option<PathBuf>,

    /// Override the shard digest anti-entropy interval (seconds) for tests.
    #[serde(default)]
    pub(super) directory_digest_interval_secs: Option<u64>,

    /// Override discoverable group card republish interval (seconds).
    /// `Some(0)` disables the periodic republish loop for tests.
    #[serde(default)]
    pub(super) group_card_republish_interval_secs: Option<u64>,

    /// Override startup shard resubscribe jitter window (milliseconds)
    /// for restart-persistence tests.
    #[serde(default)]
    pub(super) directory_resubscribe_jitter_ms: Option<u64>,

    /// Forward (tailnet) configuration (TOML: `[forward]`).
    /// Default: `require_attestation = true` — inbound ForwardV1 streams are
    /// denied. Set `require_attestation = false` for mixed-version deployments
    /// (#204 must-fix 1).
    #[serde(default)]
    pub(super) forward: x0x::forward::ForwardConfig,
}

/// Default QUIC port: 5483 (LIVE on a phone keypad).
/// Every x0x node uses the same well-known port by default.
pub const DEFAULT_QUIC_PORT: u16 = 5483;

fn default_bootstrap_peers() -> Vec<SocketAddr> {
    x0x::network::DEFAULT_BOOTSTRAP_PEERS
        .iter()
        .filter_map(|s| s.parse().ok())
        .collect()
}

fn default_port_mapping_enabled() -> bool {
    true
}

pub fn default_bind_address() -> SocketAddr {
    // Bind to IPv6 unspecified ([::]) which accepts both IPv4 and IPv6
    // via dual-stack sockets. This avoids port conflicts on macOS where
    // binding 0.0.0.0:port prevents a subsequent [::]:port bind.
    SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 0], DEFAULT_QUIC_PORT))
}

pub fn default_api_address() -> SocketAddr {
    SocketAddr::from(([127, 0, 0, 1], 12700))
}

pub fn default_data_dir() -> PathBuf {
    dirs::data_dir()
        .map(|d| d.join("x0x"))
        .unwrap_or_else(|| PathBuf::from("/var/lib/x0x"))
}

/// Shared cache directory used by ALL instances (not per-instance).
/// This is always the base `x0x` dir, never `x0x-<name>`.
pub(super) fn shared_cache_dir() -> PathBuf {
    let dir = dirs::data_dir()
        .map(|d| d.join("x0x"))
        .unwrap_or_else(|| PathBuf::from("/var/lib/x0x"));
    // Ensure it exists
    let _ = std::fs::create_dir_all(&dir);
    dir
}

fn default_log_level() -> String {
    // Privacy by default for operators outside our fleet (issue #85): without
    // an explicit RUST_LOG or config log_level override, x0xd logs warn/error
    // only. Opt in to verbose logging with RUST_LOG=info.
    "warn".to_string()
}

fn default_log_format() -> String {
    "text".to_string()
}

/// Update configuration for x0xd daemon.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(super) struct DaemonUpdateConfig {
    /// Enable listening for release manifests via gossip and the GitHub fallback poll.
    #[serde(default = "default_true")]
    pub(super) enabled: bool,

    /// Maximum rollout window in minutes. Default: 0 (immediate — no delay).
    /// Set to a positive value (e.g. 1440 for 24h) to spread upgrades across
    /// the fleet using a deterministic hash of the node MachineId.
    #[serde(default = "default_rollout_window_minutes")]
    pub(super) rollout_window_minutes: u64,

    /// Exit cleanly for service manager restart instead of spawning.
    /// Default: true — the daemon stops with exit code 0 so that systemd
    /// (or any supervisor with Restart=always) picks up the new binary.
    /// Set to false to use `exec()` in-place replacement instead.
    #[serde(default = "default_true")]
    pub(super) stop_on_upgrade: bool,

    /// GitHub fallback poll interval in minutes. Default: 2880 (48 hours).
    /// Set to 0 to disable the fallback entirely (gossip-only mode).
    #[serde(default = "default_fallback_check_interval_minutes")]
    pub(super) fallback_check_interval_minutes: u64,

    /// GitHub repo for update discovery.
    #[serde(default = "default_update_repo")]
    pub(super) repo: String,

    /// Include pre-releases in update checks (default: false).
    #[serde(default)]
    pub(super) include_prereleases: bool,

    /// Enable gossip-based release manifest propagation (default: true).
    /// Set to false to only use the GitHub fallback poll.
    #[serde(default = "default_true")]
    pub(super) gossip_updates: bool,
}

impl Default for DaemonUpdateConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            rollout_window_minutes: 0,
            stop_on_upgrade: true,
            fallback_check_interval_minutes: 2880,
            repo: default_update_repo(),
            include_prereleases: false,
            gossip_updates: true,
        }
    }
}

fn default_true() -> bool {
    true
}

fn default_rollout_window_minutes() -> u64 {
    0
}

fn default_fallback_check_interval_minutes() -> u64 {
    2880
}

fn default_update_repo() -> String {
    "saorsa-labs/x0x".to_string()
}

fn default_heartbeat_interval() -> u64 {
    x0x::IDENTITY_HEARTBEAT_INTERVAL_SECS
}

fn default_identity_ttl() -> u64 {
    x0x::IDENTITY_TTL_SECS
}

fn default_rendezvous_enabled() -> bool {
    true
}

fn default_rendezvous_validity_ms() -> u64 {
    3_600_000 // 1 hour
}

impl DaemonConfig {
    /// Whether self-update is enabled in this config. The daemon binary uses
    /// this to set [`ServeOptions::self_update_enabled`] so its behaviour is
    /// unchanged by the embed-path default of `false`.
    #[must_use]
    pub fn update_enabled(&self) -> bool {
        self.update.enabled
    }

    /// Resolve `bootstrap_peers` to a concrete dial list.
    ///
    /// - `Some(v)` → the operator's explicit peers, verbatim (including `[]`).
    /// - `None` → the embedded global bootstrap network
    ///   (`x0x::network::DEFAULT_BOOTSTRAP_PEERS`).
    ///
    /// The daemon applies `--no-hard-coded-bootstrap` *before* serving by
    /// flipping `None` to `Some([])`, so the embedded fallback never reaches
    /// this method under that flag while an explicit operator list is
    /// preserved untouched.
    #[must_use]
    pub fn resolved_bootstrap_peers(&self) -> Vec<SocketAddr> {
        self.bootstrap_peers
            .clone()
            .unwrap_or_else(default_bootstrap_peers)
    }
}

impl Default for DaemonConfig {
    fn default() -> Self {
        Self {
            bind_address: default_bind_address(),
            api_address: default_api_address(),
            data_dir: default_data_dir(),
            log_level: default_log_level(),
            log_format: default_log_format(),
            bootstrap_peers: None,
            port_mapping_enabled: default_port_mapping_enabled(),
            peer_relay: x0x::network::PeerRelayConfig::default(),
            update: DaemonUpdateConfig::default(),
            gossip: x0x::gossip::GossipConfig::default(),
            heartbeat_interval_secs: default_heartbeat_interval(),
            identity_ttl_secs: default_identity_ttl(),
            user_key_path: None,
            rendezvous_enabled: default_rendezvous_enabled(),
            rendezvous_validity_ms: default_rendezvous_validity_ms(),
            presence_beacon_interval_secs: None,
            presence_event_poll_interval_secs: None,
            presence_offline_timeout_secs: None,
            instance_name: None,
            identity_dir: None,
            directory_digest_interval_secs: None,
            group_card_republish_interval_secs: None,
            directory_resubscribe_jitter_ms: None,
            forward: x0x::forward::ForwardConfig::default(),
        }
    }
}

/// Shared state accessible from all route handlers.
pub(super) struct AppState {
    pub(super) agent: Arc<Agent>,
    pub(super) subscriptions: RwLock<HashMap<String, RestSubscription>>,
    pub(super) task_lists: RwLock<HashMap<String, TaskListHandle>>,
    pub(super) kv_stores: RwLock<HashMap<String, KvStoreHandle>>,
    /// Persisted task-list/kv-store subscription manifest so registrations
    /// survive a daemon restart (rehydrated after `join_network`; see
    /// `crdt_subscriptions`).
    pub(super) crdt_subscriptions: RwLock<super::crdt_subscriptions::CrdtSubscriptionManifest>,
    /// Disk location for the CRDT subscription manifest (instance data dir,
    /// alongside `directory-subscriptions.json`).
    pub(super) crdt_subscriptions_path: PathBuf,
    /// Serializes snapshot-and-write of `crdt-subscriptions.json` so an older
    /// in-memory snapshot cannot rename over a newer one after a concurrent
    /// `crdt_subscriptions::record` (the snapshot-after-unlock lost update).
    /// Mirrors `named_groups_persistence_lock`.
    pub(super) crdt_subscriptions_persistence_lock: Mutex<()>,
    /// Per-`(kind,id)` reservation locks serialising the full
    /// create/join → insert-handle → persist-manifest transaction for CRDT
    /// subscriptions. Shared by REST handlers and rehydration so concurrent
    /// same-`(kind,id)` requests cannot interleave handle insertion with
    /// failure rollback, and REST-vs-rehydrate cannot spawn duplicate
    /// long-lived sync listeners. Keyed by `"{kind}:{id}"`.
    pub(super) crdt_handle_locks: RwLock<HashMap<String, Arc<Mutex<()>>>>,
    pub(super) named_groups: RwLock<HashMap<String, x0x::groups::GroupInfo>>,
    pub(super) named_groups_path: PathBuf,
    /// Serializes snapshot-and-write of `named_groups.json` so an older
    /// snapshot cannot rename over a newer recovered KeyPackage.
    pub(super) named_groups_persistence_lock: Mutex<()>,
    /// Background metadata listeners for named groups (one per group id).
    pub(super) group_metadata_tasks: RwLock<HashMap<String, tokio::task::JoinHandle<()>>>,
    /// Cached group cards discovered via gossip or imported from peers.
    pub(super) group_card_cache: RwLock<HashMap<String, x0x::groups::GroupCard>>,
    /// Phase C.2: per-shard cache of signed cards received via
    /// `x0x.directory.{tag|name|id}.{N}` gossip topics.
    pub(super) directory_cache: RwLock<x0x::groups::DirectoryShardCache>,
    /// Phase C.2: persistent set of shard subscriptions. Survives
    /// daemon restart (see `directory_subscriptions_path`).
    pub(super) directory_subscriptions: RwLock<x0x::groups::SubscriptionSet>,
    /// Phase C.2: disk location for subscription persistence.
    pub(super) directory_subscriptions_path: PathBuf,
    /// Phase C.2: background shard-listener tasks, keyed by (kind, shard).
    pub(super) directory_tasks:
        RwLock<HashMap<(x0x::groups::ShardKind, u32), tokio::task::JoinHandle<()>>>,
    /// Phase C.2: digest anti-entropy interval in seconds.
    pub(super) directory_digest_interval_secs: u64,
    /// Phase C.2: startup shard resubscribe jitter window in milliseconds.
    pub(super) directory_resubscribe_jitter_ms: u64,
    /// Phase E: per-group ring buffer of validated public messages.
    /// Keyed by `group_id`. Bounded by `PUBLIC_MESSAGE_HISTORY_CAP`.
    pub(super) public_messages: RwLock<HashMap<String, Vec<x0x::groups::GroupPublicMessage>>>,
    /// Phase E: background listener tasks on public-chat topics.
    pub(super) public_message_tasks: RwLock<HashMap<String, tokio::task::JoinHandle<()>>>,
    /// Per-daemon ML-KEM-768 keypair used to open `SecureShareDelivered`
    /// envelopes addressed to this agent. Public half is published in the
    /// `/agent` response and in `JoinRequestCreated` so other daemons can
    /// seal to us. Replaces the earlier publicly-derivable envelope key.
    pub(super) agent_kem_keypair: Arc<x0x::groups::kem_envelope::AgentKemKeypair>,
    pub(super) contacts: Arc<RwLock<ContactStore>>,
    pub(super) mls_groups: RwLock<HashMap<String, x0x::mls::MlsGroup>>,
    #[allow(dead_code)]
    pub(super) mls_groups_path: PathBuf,
    /// Authority-signed MemberAdded results staged by an anchor for joiner polling.
    pub(super) pending_join_results: RwLock<HashMap<String, PendingJoinResult>>,
    /// Expected inviter for a pending TreeKEM join-result response, keyed by
    /// stable group id + joining member id. Transient process-local state.
    pub(super) expected_join_result_inviters: StdMutex<HashMap<String, ExpectedJoinResultInviter>>,
    /// TreeKEM Welcome blobs staged by an anchor for pull-based delivery.
    pub(super) pending_welcomes: RwLock<HashMap<String, PendingWelcome>>,
    /// In-progress pulled TreeKEM Welcome blob receives, keyed by blake3 id.
    pub(super) pending_welcome_receives: RwLock<HashMap<String, PendingWelcomeReceive>>,
    /// Waiters blocked on a Welcome blob receive completing.
    pub(super) pending_welcome_waiters: RwLock<HashMap<String, Vec<WelcomeFetchWaiter>>>,
    /// Per-active Welcome blob transfer ack slots.
    pub(super) pending_welcome_acks: RwLock<HashMap<String, Arc<FileChunkAckSlot>>>,
    /// Bounded per-group queue for verified TreeKEM membership events that
    /// arrived before local TreeKEM readiness or ahead of our state frontier.
    pub(super) treekem_pending_events:
        RwLock<HashMap<String, VecDeque<PendingTreeKemMetadataEvent>>>,
    /// Bounded per-group log of locally authored/applied TreeKEM membership
    /// events used to satisfy explicit catch-up requests.
    pub(super) treekem_event_log: RwLock<HashMap<String, VecDeque<NamedGroupMetadataEvent>>>,
    /// Bounded member-keyed cache of verified TreeKEM key-package-bearing
    /// `MemberJoined` recovery records. It retains inviter-authority-attested
    /// records and independently witnessed provisional records under the same
    /// global count/byte limits, and owns lifecycle pruning, serialized
    /// persistence, dirty retry state, and diagnostics.
    pub(super) treekem_member_key_packages: super::TreeKemMemberKeyPackageCache,
    /// Anti-spam throttle for outbound catch-up requests.
    pub(super) treekem_catchup_throttle: RwLock<HashMap<String, Instant>>,
    /// Per-group serialization for authoritative membership mutations. The
    /// owner-side `MemberJoined`→`MemberAdded` add (and every other membership
    /// apply) is a read-modify-write that loads `info`, mutates a clone, mutates
    /// the live MLS tree, then commits the roster — across several locks, not
    /// one. The gossip metadata listener and the direct-channel listener call
    /// `apply_named_group_metadata_event` for the same group concurrently, so
    /// without serialization two stale clones can both pass the
    /// `has_active_member` check, both consume the bearer invite, and double-add
    /// to the MLS tree (the second add fails "already a member") while the
    /// roster is clobbered or never committed — leaving tree and roster
    /// permanently diverged. This per-group mutex serializes those applies so
    /// the second observes the committed add and cleanly no-ops.
    pub(super) group_membership_locks: RwLock<HashMap<String, Arc<Mutex<()>>>>,
    /// Live real-TreeKEM groups (ADR-0012), keyed by group-id hex. Each is
    /// wrapped in its own async mutex so a group's encrypt/decrypt/commit op
    /// and the snapshot-persist that follows it are serialized per group
    /// without blocking other groups (and without holding the map lock across
    /// disk IO). Snapshots persist under [`Self::treekem_dir`].
    pub(super) treekem_groups:
        RwLock<HashMap<String, Arc<tokio::sync::Mutex<x0x::mls::TreeKemMlsGroup>>>>,
    /// Directory holding `<group_id>.snap` snapshots and `<group_id>.journal`
    /// TreeKEM persistence journals (mode 0600).
    pub(super) treekem_dir: PathBuf,
    /// Active WebSocket sessions.
    pub(super) ws_sessions: RwLock<HashMap<String, WsSession>>,
    /// Shared WS topic state (single lock for channel + subscribers + forwarder per topic).
    pub(super) ws_topics: RwLock<HashMap<String, SharedTopicState>>,
    /// Per-WS-outbound-queue observability (drop / slow-consumer-close counters).
    pub(super) ws_outbound_stats: Arc<WsOutboundStats>,
    pub(super) api_address: SocketAddr,
    pub(super) start_time: Instant,
    pub(super) broadcast_tx: broadcast::Sender<SseEvent>,
    /// Active file transfers.
    pub(super) file_transfers: RwLock<HashMap<String, x0x::files::TransferState>>,
    /// Incremental SHA-256 hashers for receiving transfers.
    pub(super) receive_hashers: RwLock<HashMap<String, Sha256>>,
    /// Out-of-order decoded chunks buffered until their predecessors arrive.
    pub(super) pending_file_chunks: RwLock<HashMap<String, BTreeMap<u64, Vec<u8>>>>,
    /// Per-active-send-transfer chunk-ack slots used to apply windowed
    /// back-pressure. Sender registers a slot at the start of
    /// `stream_file_chunks`; receiver replies with `FileMessage::ChunkAck`
    /// after each chunk is persisted; sender waits before exceeding the
    /// in-flight window. Removed when the transfer terminates. See
    /// `FILE_CHUNK_WINDOW` and `FileChunkAckSlot`.
    pub(super) file_chunk_acks: RwLock<HashMap<String, Arc<FileChunkAckSlot>>>,
    /// Directory for received file data.
    pub(super) transfers_dir: PathBuf,
    /// Channel to trigger graceful shutdown from the /shutdown endpoint.
    pub(super) shutdown_tx: mpsc::Sender<()>,
    /// Broadcasts daemon shutdown so long-lived SSE/WS connections can close.
    pub(super) shutdown_notify: watch::Sender<bool>,
    /// Update configuration honored by manual API-triggered update checks.
    pub(super) update_config: DaemonUpdateConfig,
    /// Whether install/restart-capable self-update is allowed. `false` on the
    /// embed path so `/upgrade/apply` cannot replace/restart the host process.
    pub(super) self_update_enabled: bool,
    /// Cached `/upgrade` response so polling clients cannot hammer GitHub.
    pub(super) upgrade_check_cache: Mutex<Option<CachedUpgradeCheck>>,
    /// Serializes all destructive binary replacement attempts.
    pub(super) upgrade_apply_lock: Arc<Mutex<()>>,
    /// API bearer token for authenticating local clients.
    pub(super) api_token: String,
    /// Short-lived browser session tokens (#127 / WS1.6). The only tokens
    /// accepted via `?token=` query strings on WS/SSE endpoints.
    pub(super) sessions: SessionStore,
    /// Tier-1 remote exec service.
    pub(super) exec_service: Arc<x0x::exec::ExecService>,
    /// Per-group ingest diagnostics surfaced via `/diagnostics/groups`.
    pub(super) groups_diagnostics: Arc<x0x::groups::GroupsDiagnostics>,
    /// Connect-ACL allow/deny counters + policy summary for
    /// `/diagnostics/connect`. Counters reflect live forwards when connect is
    /// enabled (the forwarder calls `record_allowed`/`record_denied`) and read
    /// 0 when it is disabled (the default).
    pub(super) connect_diagnostics: Arc<x0x::connect::ConnectDiagnostics>,
    /// Tailnet forwarder service (#132 T4/T6): owns the inbound connect-gated
    /// consumer + outbound local-port listeners. `None` when connect is
    /// disabled (no policy) so the daemon runs zero forwarder tasks.
    pub(super) forward_service: Option<Arc<x0x::forward::ForwardService>>,
}

#[derive(Clone)]
pub(super) struct CachedUpgradeCheck {
    pub(super) checked_at: Instant,
    pub(super) status: StatusCode,
    pub(super) body: serde_json::Value,
    pub(super) ttl: Duration,
}

#[cfg(test)]
mod tests {
    use super::*;

    // The two canonical messages enforced by `InstanceName::try_from`.
    // `x0xd::resolve_instance_startup` propagates these verbatim (bare `?`,
    // no added context), so an invalid CLI or config name surfaces one
    // identical, established message at startup.
    const GRAMMAR_ERR: &str =
        "instance name must start with alphanumeric and contain only alphanumeric or hyphens";
    const LENGTH_ERR: &str = "instance name must be 1-64 characters";

    #[test]
    fn try_from_rejects_path_traversal_separators_and_invalid_grammar() {
        // Each row is a named boundary a path-injection or grammar bug would
        // ride in on. The separator/traversal cluster is security-critical:
        // these are the exact strings that must never reach path derivation.
        let overlength = "a".repeat(65); // 65 > 64 max
        let cases: &[(&str, &str)] = &[
            // separators — forward slash + backslash, bare and embedded
            ("/", GRAMMAR_ERR),
            ("\\", GRAMMAR_ERR),
            ("a/b", GRAMMAR_ERR),
            ("a\\b", GRAMMAR_ERR),
            // traversal components / absolute paths
            ("..", GRAMMAR_ERR),
            ("../etc/passwd", GRAMMAR_ERR),
            ("/etc/passwd", GRAMMAR_ERR),
            // emptiness (length rule, checked before grammar)
            ("", LENGTH_ERR),
            // whitespace
            ("   ", GRAMMAR_ERR),
            ("ab cd", GRAMMAR_ERR),
            // leading hyphen (also a CLI-flag-injection hazard)
            ("-lead", GRAMMAR_ERR),
            // non-ASCII
            ("café", GRAMMAR_ERR),
            // overlength boundary (length rule)
            (overlength.as_str(), LENGTH_ERR),
        ];

        for &(raw, expected) in cases {
            let err = InstanceName::try_from(raw.to_owned())
                .expect_err("invalid instance name must be rejected");
            let msg = err.to_string();
            assert!(
                msg.contains(expected),
                "{raw:?}: expected error containing {expected:?}, got {msg:?}"
            );
        }
    }

    #[test]
    fn try_from_accepts_valid_names_and_preserves_bytes() {
        // Boundary pair with the 65-char case above: 64 passes, 65 fails.
        let max_len = "a".repeat(64);
        let cases: &[&str] = &[
            "a",              // single char — lower length boundary
            "testnet",        // lowercase alphanumeric
            "x0x-443",        // alphanumeric + hyphen
            "ProdNode1",      // mixed case + digits
            "build-",         // trailing hyphen allowed (hyphen barred only at position 0)
            max_len.as_str(), // exactly 64 chars — upper length boundary
        ];

        for &raw in cases {
            let name = InstanceName::try_from(raw.to_owned())
                .unwrap_or_else(|e| panic!("{raw:?}: expected valid, got {e}"));
            assert_eq!(
                name.as_str(),
                raw,
                "an accepted name must be preserved byte-for-byte (no trim/lowercase)"
            );
        }
    }

    /// Backcompat / parity guard for the stable borrowed validator.
    ///
    /// `x0x::server::validate_instance_name` is the documented
    /// `fn(&str) -> anyhow::Result<()>` surface that pre-dates the typed
    /// [`InstanceName`] constructor. It must reach the SAME verdict as
    /// [`InstanceName::try_from`] and emit the SAME error bytes for every
    /// grammar/length boundary, so a caller cannot observe a name that one path
    /// accepts but the other rejects. The two `try_from_*` tests above pin the
    /// constructor's absolute behavior against the canonical messages; this test
    /// pins the *relationship* between the two surfaces and deliberately does
    /// not re-encode the grammar (it derives expectations from the sibling API),
    /// so it stays correct if the messages are ever reworded.
    #[test]
    fn borrowed_validator_matches_typed_constructor_outcomes_and_messages() {
        // Bind the restored public signature as a typed fn pointer. If the
        // `fn(&str) -> anyhow::Result<()>` surface ever changes arity, argument
        // type, or return type, this binding fails to COMPILE — a
        // backcompat guard stronger than any runtime assertion.
        let validate: fn(&str) -> anyhow::Result<()> = x0x::server::validate_instance_name;

        let max = "a".repeat(64); // exactly 64 — upper length boundary (valid)
        let over = "a".repeat(65); // 65 — one past the boundary (invalid)

        // One representative row per documented category: valid boundaries on
        // the success side; every documented rejection class on the failure
        // side. No grammar is re-encoded below — each row is compared against
        // the typed constructor's actual output.
        let cases: &[&str] = &[
            // --- valid boundaries ---
            "a",          // single char — lower length boundary
            "testnet",    // lowercase alphanumeric
            "x0x-443",    // alphanumeric + hyphen (embedded hyphen allowed)
            max.as_str(), // exactly 64 chars — upper length boundary
            // --- invalid: empty (length rule) ---
            "",
            // --- invalid: whitespace ---
            "   ",   // whitespace only
            "ab cd", // embedded space
            "\t",    // tab
            // --- invalid: path separators ---
            "/",    // forward slash
            "\\",   // backslash
            "a/b",  // embedded forward slash
            "a\\b", // embedded backslash
            // --- invalid: traversal + absolute & platform path forms ---
            "..",                // parent traversal
            "../etc/passwd",     // traversal with target
            "/etc/passwd",       // absolute POSIX path
            "C:\\Users",         // absolute Windows drive path (platform form)
            "\\\\server\\share", // UNC path form
            // --- invalid: leading hyphen (also a CLI-flag-injection hazard) ---
            "-lead",
            // --- invalid: non-ASCII ---
            "café", // accented Latin
            "测试", // CJK
            // --- invalid: overlength (length rule) ---
            over.as_str(),
        ];

        for &raw in cases {
            let borrowed = validate(raw);
            let typed = InstanceName::try_from(raw.to_owned());
            match (&borrowed, &typed) {
                (Ok(()), Ok(_)) => {}
                (Err(borrowed_err), Err(typed_err)) => assert_eq!(
                    borrowed_err.to_string(),
                    typed_err.to_string(),
                    "borrowed and typed validators disagree on error message for {raw:?}"
                ),
                (Ok(()), Err(typed_err)) => panic!(
                    "parity break: borrowed validator ACCEPTED {raw:?} \
                     but typed constructor rejected it: {typed_err}"
                ),
                (Err(borrowed_err), Ok(_)) => panic!(
                    "parity break: borrowed validator REJECTED {raw:?} \
                     but typed constructor accepted it: {borrowed_err}"
                ),
            }
        }
    }
    // Mirrors the exact logic in `src/bin/x0xd.rs` so the unit tests below
    // exercise the real flag application, not an approximation.
    fn apply_no_hard_coded_bootstrap(config: &mut DaemonConfig) {
        if config.bootstrap_peers.is_none() {
            config.bootstrap_peers = Some(Vec::new());
        }
    }

    #[test]
    fn flag_with_explicit_config_peers_keeps_them() {
        // (1) Flag + explicit `bootstrap_peers` in config → operator peers
        // kept verbatim, embedded defaults absent. This is the regression we
        // are fixing: previously the flag wiped operator peers too.
        let mut config: DaemonConfig =
            toml::from_str(r#"bootstrap_peers = ["127.0.0.1:6483", "[::1]:6483"]"#)
                .expect("explicit peers parse");
        assert_eq!(
            config.bootstrap_peers,
            Some(vec![
                "127.0.0.1:6483".parse().unwrap(),
                "[::1]:6483".parse().unwrap(),
            ])
        );

        apply_no_hard_coded_bootstrap(&mut config);

        // Unchanged — the operator's list survives the flag.
        let resolved = config.resolved_bootstrap_peers();
        assert_eq!(resolved.len(), 2);
        assert!(resolved.contains(&"127.0.0.1:6483".parse().unwrap()));
        assert!(resolved.contains(&"[::1]:6483".parse().unwrap()));
    }

    #[test]
    fn flag_without_config_key_clears_embedded_defaults() {
        // (2) Flag + no `bootstrap_peers` key in config → the embedded
        // global bootstrap network is skipped (resolves to empty).
        let mut config: DaemonConfig = toml::from_str("").expect("empty config parses");
        assert!(
            config.bootstrap_peers.is_none(),
            "absent key must deserialize to None, not the embedded defaults"
        );

        apply_no_hard_coded_bootstrap(&mut config);

        assert_eq!(config.resolved_bootstrap_peers(), Vec::<SocketAddr>::new());
    }

    #[test]
    fn no_flag_without_config_key_uses_embedded_defaults() {
        // (3) No flag + no `bootstrap_peers` key → embedded global bootstrap
        // network resolves (unchanged pre-existing behaviour).
        let config: DaemonConfig = toml::from_str("").expect("empty config parses");
        let resolved = config.resolved_bootstrap_peers();
        let embedded: Vec<SocketAddr> = x0x::network::DEFAULT_BOOTSTRAP_PEERS
            .iter()
            .filter_map(|s| s.parse().ok())
            .collect();
        assert!(!embedded.is_empty(), "sanity: embedded list is non-empty");
        assert_eq!(resolved, embedded);
    }

    #[test]
    fn explicit_empty_list_is_honored_not_treated_as_default() {
        // An explicit `bootstrap_peers = []` means "the operator wants no seed
        // peers". It must deserialize to Some([]) — distinct from the absent
        // key (None) — and survive the flag untouched.
        let mut config: DaemonConfig =
            toml::from_str("bootstrap_peers = []").expect("empty list parses");
        assert_eq!(config.bootstrap_peers, Some(Vec::new()));

        apply_no_hard_coded_bootstrap(&mut config);

        // Still Some([]) — explicit-empty is an operator choice, not a default.
        assert_eq!(config.resolved_bootstrap_peers(), Vec::<SocketAddr>::new());
    }

    #[test]
    fn default_config_resolves_to_embedded_network() {
        // DaemonConfig::default() (no config file at all) must resolve to the
        // embedded bootstrap network — same as an absent TOML key.
        let config = DaemonConfig::default();
        assert!(config.bootstrap_peers.is_none());
        let embedded: Vec<SocketAddr> = x0x::network::DEFAULT_BOOTSTRAP_PEERS
            .iter()
            .filter_map(|s| s.parse().ok())
            .collect();
        assert_eq!(config.resolved_bootstrap_peers(), embedded);
    }
}