bitrouter 1.0.0-alpha.34

//! Daemon control over a local IPC channel.
//!
//! A running `bro serve` listens on a control endpoint alongside the
//! HTTP API. The CLI's `stop` / `restart` / `reload` / `status` / `route`
//! subcommands are thin clients that connect, send one newline-delimited JSON
//! [`DaemonCommand`], and read one [`DaemonResponse`].
//!
//! The transport is platform-specific but the wire protocol is identical:
//! a **Unix domain socket** (mode `0600`) on Unix, and a **Windows named
//! pipe** (`\\.\pipe\bitrouter-…`, secured by its default owner-only DACL)
//! on Windows. Both are encapsulated by a private `transport` module so the
//! rest of this file — and every caller — is platform-agnostic.

use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};

use bitrouter_sdk::App;
use bitrouter_sdk::caller::CallerContext;
use bitrouter_sdk::language_model::RoutingPrefs;

use chrono::{DateTime, Utc};

use crate::acp_runtime::AcpRuntime;
use crate::actions::administration::{
    Administration, AgentsReport, ObserveReport, PolicyInput, PolicyReport, ProvidersReport,
};
use crate::metering::{MeteringStore, TimeWindow};
use crate::reload::{ReloadAdmissionError, ReloadReport, ReloadReservation, ReloadState};

/// Anything the daemon's `Reload` command (and SIGHUP) should re-read. The
/// runtime reloader fans out to every reloadable subsystem — routing table,
/// policy store, … — atomically per subsystem. A failure in one is reported
/// but does not abort the others. The trait uses `#[async_trait]` so it is
/// object-safe (`Arc<dyn DaemonReloader>`).
#[async_trait::async_trait]
pub trait DaemonReloader: Send + Sync {
    /// Reload every reloadable subsystem.
    async fn reload(&self) -> anyhow::Result<()>;

    /// Reload with owner-trusted local environment overrides. Implementations
    /// that own a coordinator install overrides after reserving exclusive
    /// reload ownership; a generic reloader cannot truthfully promise that.
    async fn reload_with_env(&self, env: Vec<(String, String)>) -> anyhow::Result<()> {
        if env.is_empty() {
            self.reload().await
        } else {
            Err(anyhow::anyhow!(
                "this daemon reloader does not support coordinated environment overrides"
            ))
        }
    }

    /// Return boot-local reload state when this implementation supports guarded
    /// reload admission. `None` deliberately means unsupported, rather than a
    /// fabricated instance identity or generation.
    fn reload_state(&self) -> Option<ReloadState> {
        None
    }

    /// Redaction-safe saved/running configuration state. The read is async
    /// because the server must inspect its own source files; callers must not
    /// substitute a client-local config.
    async fn configuration_state(&self) -> Option<crate::reload::ConfigurationState> {
        None
    }

    /// Atomically fence a remote operation against the current boot and
    /// generation. The reservation must be passed to [`Self::reload_reserved`]
    /// exactly once or dropped so its coordinator can clear the admission.
    fn reserve_remote(
        &self,
        _expected_instance: &str,
        _expected_generation: u64,
    ) -> Result<ReloadReservation, ReloadAdmissionError> {
        Err(ReloadAdmissionError::Unsupported)
    }

    /// Execute a previously admitted remote reload. The default is reachable
    /// only through an invalid implementation because [`Self::reserve_remote`]
    /// rejects every request, and still returns a truthful unsupported report.
    async fn reload_reserved(&self, _reservation: ReloadReservation) -> ReloadReport {
        ReloadReport::unsupported()
    }
}

/// A reloader that does nothing — useful for tests / minimal embeddings of the
/// daemon control surface that don't have anything to reload.
pub struct NoopReloader;

#[async_trait::async_trait]
impl DaemonReloader for NoopReloader {
    async fn reload(&self) -> anyhow::Result<()> {
        Ok(())
    }
}

/// A command sent from the CLI to a running daemon.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "cmd", rename_all = "snake_case")]
pub enum DaemonCommand {
    /// Stop the daemon — it finishes the response, then exits.
    Stop,
    /// Discover the daemon's handoff contract without changing its state.
    HandoffHello,
    /// Close admission if all daemon-owned work is idle.
    HandoffPrepare,
    /// Reopen admission when pre-stop checks fail.
    HandoffAbort { token: String },
    /// Stop only the daemon that issued this short-lived token.
    HandoffStop { token: String },
    /// Hot-reload the config / routing table. The CLI piggybacks a
    /// snapshot of API-key-style env vars from its own process so a
    /// `export OPENAI_API_KEY=…; bro reload` propagates the new
    /// value into the running daemon without requiring a restart.
    /// `env` is `#[serde(default)]` for wire-compat with the
    /// historical unit variant — older clients sending `{"cmd":"reload"}`
    /// still deserialise as an empty override list.
    Reload {
        /// `(name, value)` pairs to apply to the daemon's env-override
        /// map before reload. Empty list = no override changes.
        #[serde(default)]
        env: Vec<(String, String)>,
    },
    /// Read the reload coordinator's boot-local state without mutating it.
    ReloadState,
    /// Report daemon status.
    Status,
    /// List every model the live routing table can route, each with the
    /// providers that can serve it.
    ///
    /// Distinct from [`Self::Status`], which reports only the *count* and the
    /// distinct provider set: this is the catalog itself, and it is what lets
    /// `bro models` and the MCP `list_models` tool answer with the
    /// daemon's real view — subscription-backed providers and post-`reload`
    /// state included — instead of a static config projection.
    Models,
    /// Resolve a model name through the live routing table.
    Route {
        /// The model name to resolve.
        model: String,
    },
    /// Report the OTel exporter's current state — what's wired, current
    /// cardinality usage, in-flight span count. Returned as a JSON
    /// snapshot. The wire format is the same `ObserveStatusPayload` the
    /// CLI pretty-prints for `bro observe status`.
    ObserveStatus,
    /// Run one safe, typed administration read against the live daemon.
    ///
    /// This is intentionally a closed enum rather than a JSON method name or
    /// a serialized HTTP request.  The local control socket remains a
    /// host-local transport with its own contract.
    Inspect { inspection: DaemonInspection },
    /// Local, owner-scoped checkpoint evolution control.
    Evolution {
        operation: crate::evolution::operator::EvolutionOperation,
    },
    /// Acknowledge request inventory for a recorder before capture starts.
    AcpRecordingRegister {
        connection_id: String,
        api_principal: String,
        controller_instance_id: String,
    },
    /// Remove every route lease in one principal/controller namespace.
    AcpControllerCleanup {
        /// Opaque principal derived from the normal API credential, or local.
        api_principal: String,
        /// Caller-declared controller process identity.
        controller_instance_id: String,
    },
    /// Query routes and exact lease state for one native session.
    AcpRouteList {
        /// Opaque principal derived from the normal API credential, or local.
        api_principal: String,
        /// Caller-declared controller process identity.
        controller_instance_id: String,
        /// Harness-native ACP session identity.
        session_id: String,
    },
    /// Install or replace one native-session route lease.
    AcpRouteSet {
        /// Opaque principal derived from the normal API credential, or local.
        api_principal: String,
        /// Caller-declared controller process identity.
        controller_instance_id: String,
        /// Harness-native ACP session identity.
        session_id: String,
        /// BitRouter route selector.
        route: String,
    },
    /// Remove one native-session route lease.
    /// Read the spend metered under one controller for one native session.
    AcpSessionSpend {
        /// Opaque principal derived from the normal API credential, or local.
        api_principal: String,
        /// Caller-declared controller process identity.
        controller_instance_id: String,
        /// Harness-native ACP session identity.
        session_id: String,
    },
    AcpRouteReset {
        /// Opaque principal derived from the normal API credential, or local.
        api_principal: String,
        /// Caller-declared controller process identity.
        controller_instance_id: String,
        /// Harness-native ACP session identity.
        session_id: String,
    },
    /// Versioned daemon-owned ACP session control. This variant is accepted
    /// only on the owner-scoped local transport; the remote control server has
    /// no corresponding operation.
    Sessions {
        request: crate::supervisor::SessionRequest,
    },
}

/// Passive live inspection operations accepted over the local control socket.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "inspection", rename_all = "snake_case")]
pub enum DaemonInspection {
    Providers,
    Agents,
    Observe,
    Policy { input: PolicyInput },
}

/// Typed result of a [`DaemonInspection`] request.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind", content = "report", rename_all = "snake_case")]
pub enum DaemonInspectionReport {
    Providers(ProvidersReport),
    Agents(AgentsReport),
    Observe(ObserveReport),
    Policy(PolicyReport),
}

/// One resolved hop of a route chain.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RouteHop {
    /// Provider id.
    pub provider: String,
    /// Service / model id at the provider.
    pub service_id: String,
    /// The wire protocol for the hop.
    pub api_protocol: String,
}

/// Point-in-time activity observed by the local daemon. Missing counts mean
/// that subsystem could not prove idleness.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct HandoffActivity {
    pub operations: usize,
    pub supervised_runs: Option<usize>,
    pub route_leases: usize,
    pub acp_recordings: Option<u64>,
    pub drain_in_progress: bool,
}

/// The daemon's reply to a [`DaemonCommand`].
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "resp", rename_all = "snake_case")]
pub enum DaemonResponse {
    /// The command succeeded with no payload.
    Ok,
    HandoffHello {
        version: String,
        protocol: u32,
        instance_id: String,
        build_id: String,
        cli_owned: bool,
    },
    HandoffReady {
        token: String,
    },
    HandoffBusy {
        reason: String,
    },
    /// Status payload.
    Status {
        /// The daemon's process id.
        pid: u32,
        /// Binary version. Missing on legacy daemons.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        daemon_version: Option<String>,
        /// Handoff protocol; missing means idle cannot be proven.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        handoff_protocol: Option<u32>,
        /// Fingerprint of the daemon's handoff and migration implementation.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        handoff_build_id: Option<String>,
        /// Current activity. Missing on a legacy daemon.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        handoff_activity: Option<HandoffActivity>,
        /// Whether the detached `bro start` launcher owns this process.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        cli_owned: Option<bool>,
        /// The HTTP listen address.
        listen: String,
        /// Count of routable models.
        models: usize,
        /// The distinct providers behind those models, sorted.
        ///
        /// `#[serde(default)]` for wire-compat: a client talking to a daemon
        /// from before this field existed reads an empty list rather than
        /// failing the whole exchange.
        #[serde(default)]
        providers: Vec<String>,
        /// Router inventory from the configuration source owned by this
        /// daemon. A caller-local `--config` is never mixed into a live reply.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        saved_routers: Option<Vec<crate::actions::models::RouterStatus>>,
        /// Router inventory from the same running config. `None` means the
        /// daemon predates this field or has no inspectable app routing config.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        running_routers: Option<Vec<crate::actions::models::RouterStatus>>,
        /// Whether the saved canonical router map differs from the full map
        /// held by this daemon. Computed host-side so secret default values do
        /// not cross the control boundary.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        router_restart_required: Option<bool>,
        /// Whole saved/running configuration state computed by the daemon.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        config_state: Option<Box<crate::reload::ConfigurationState>>,
    },
    /// The live routing table's catalog.
    Models {
        /// Every routable model, each with all the providers declaring it.
        models: Vec<bitrouter_sdk::language_model::routing::ModelInfo>,
        /// Routers from the same live routing snapshot.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        routers: Option<Vec<crate::actions::models::RouterStatus>>,
    },
    /// A resolved route chain.
    Route {
        /// The ordered fallback chain.
        chain: Vec<RouteHop>,
        /// Stage-0 model used to build `chain`. Missing from older daemons.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        resolved_model: Option<String>,
        /// Stable named-router identity, when the selector resolved one.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        router: Option<bitrouter_sdk::language_model::routing::RouterRequestIdentity>,
        /// Canonical or legacy source of the named-router definition.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        router_source: Option<crate::actions::models::RouterSource>,
        /// Dynamic policy bound to the router. The control verb does not run
        /// that policy and therefore never reports a selected candidate.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        bound_policy: Option<String>,
        /// Redaction-safe possible targets from the active policy snapshot.
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        candidate_models: Vec<String>,
    },
    /// OTel exporter snapshot.
    ObserveStatus {
        /// The serialized exporter state.
        payload: ObserveStatusPayload,
    },
    /// Reload coordinator state for the current daemon boot.
    ReloadState {
        state: ReloadState,
    },
    /// A typed administration inspection result.
    Inspection {
        report: DaemonInspectionReport,
    },
    Evolution {
        report: Box<crate::evolution::operator::EvolutionReport>,
    },
    /// This daemon shares the recorder's database and records gateway requests.
    AcpRecordingRegistered {
        registration: crate::evolution::inventory::CoverageRegistration,
    },
    /// Daemon-confirmed route state for one native ACP session.
    AcpRouteState {
        /// Live model selectors accepted as logical routes.
        available: Vec<String>,
        /// Exact installed session route, if any.
        current: Option<String>,
        /// `session` for a lease and `default` otherwise.
        scope: String,
    },
    /// Spend metered under one controller for one native session and its
    /// descendants. Raw summary: the controller decides what reaches the
    /// wire, and unpriced rows are never a computed zero.
    AcpSessionSpend {
        /// Micro-USD across the rows that carry charge evidence.
        spend_micro_usd: u64,
        /// Rows attributed to the session, priced or not.
        requests: u64,
        /// Rows without charge evidence, absent from `spend_micro_usd`.
        unpriced: u64,
    },
    /// Versioned daemon-owned ACP session response.
    Sessions {
        response: crate::supervisor::SessionResponse,
    },
    /// The command failed.
    Error {
        /// Human-readable failure detail.
        message: String,
    },
}

/// Serializable snapshot of the OTel exporter's state, transported over
/// the daemon control socket. Fields mirror `bitrouter_telemetry::otel::OtelStatus`
/// — but this module deliberately re-states the wire format rather than
/// serializing the SDK type directly, so the JSON shape on the control
/// socket stays stable even if `OtelStatus` gains, drops, or renames a
/// field. The duplication is the point: it makes any wire-format change a
/// visible edit here.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ObserveStatusPayload {
    /// Whether the `otel` feature was compiled in.
    pub compiled_in: bool,
    /// Whether an exporter is actually wired.
    pub exporter_wired: bool,
    /// OTLP/HTTP+protobuf endpoint.
    pub endpoint: Option<String>,
    /// Number of additional headers (count only — no credential leak).
    pub header_count: usize,
    /// Service name reported on the resource.
    pub service_name: Option<String>,
    /// Number of `OTEL_RESOURCE_ATTRIBUTES` entries merged in.
    pub resource_attribute_count: usize,
    /// Sampler kind (e.g. `parentbased_always_on`).
    pub sampler: Option<String>,
    /// Sampler ratio argument (only for `*_traceidratio`).
    pub sampler_arg: Option<f64>,
    /// Whether metrics export is enabled.
    pub metrics_enabled: bool,
    /// Distinct `api_key_id` values seen by the cardinality limiter.
    pub api_key_count: usize,
    /// Cardinality cap for the `api_key_id` metric dimension.
    pub api_key_cap: usize,
    /// Distinct `user_id` values seen.
    pub user_id_count: usize,
    /// Cardinality cap for the `user_id` metric dimension.
    pub user_id_cap: usize,
    /// In-flight spans currently tracked.
    pub active_spans: usize,
}

impl ObserveStatusPayload {
    /// Payload for "feature compiled in but no exporter wired" — used by
    /// the noop provider and by the CLI's stopped-daemon fallback.
    pub fn unwired(compiled_in: bool) -> Self {
        Self {
            compiled_in,
            exporter_wired: false,
            endpoint: None,
            header_count: 0,
            service_name: None,
            resource_attribute_count: 0,
            sampler: None,
            sampler_arg: None,
            metrics_enabled: false,
            api_key_count: 0,
            api_key_cap: 0,
            user_id_count: 0,
            user_id_cap: 0,
            active_spans: 0,
        }
    }
}

/// Source of truth for the OTel exporter's current state, *and* the
/// lifecycle hook the daemon's shutdown path calls to flush pending
/// telemetry before the runtime tears down. Mirrors the
/// [`DaemonReloader`] pattern: the daemon module holds a trait object,
/// the app's assembly layer provides the concrete impl (which reads
/// from — and drives — the running `OtelExporter`).
///
/// Status and shutdown share a trait because they share a singleton:
/// each `serve()` has exactly one observe provider, with one address
/// and one set of background tasks. Splitting them would duplicate the
/// `Arc<dyn …>` plumbing without buying separation of concerns.
#[async_trait::async_trait]
pub trait ObserveStatusProvider: Send + Sync {
    /// Snapshot what the exporter looks like right now.
    fn status(&self) -> ObserveStatusPayload;

    /// Flush in-flight telemetry and tear down the exporter's background
    /// tasks. Must be driven from an async context that allows
    /// `spawn_blocking`, because the underlying SDK shutdown is
    /// synchronous and needs a separate thread to park on while the
    /// `rt-tokio` worker drains the SDK's internal channels.
    ///
    /// Default impl is a no-op so providers that own nothing
    /// (the noop / test variants) don't need to implement it.
    async fn shutdown(&self) {}
}

/// Provider that reports "not wired" — used when no exporter is built
/// (no YAML opt-in, no env var). `compiled_in` is plumbed through so the
/// caller can still distinguish "feature off" from "feature on but no
/// config."
pub struct NoopObserveStatus {
    /// Whether the `otel` feature was compiled into the binary.
    pub compiled_in: bool,
}

#[async_trait::async_trait]
impl ObserveStatusProvider for NoopObserveStatus {
    fn status(&self) -> ObserveStatusPayload {
        ObserveStatusPayload::unwired(self.compiled_in)
    }
    // `shutdown` falls back to the trait's default no-op — there is no
    // exporter to flush.
}

/// The default control-socket path when the config does not set one.
/// Stored as a relative path so [`resolve_socket_path`] places it next
/// to the config file rather than in the daemon's working directory.
pub const DEFAULT_CONTROL_SOCKET: &str = "./bitrouter.sock";

/// Resolve `cfg.server.control_socket` against the config file's
/// directory. Absolute paths are returned as-is; relative paths are
/// joined onto the parent of `config_path`. This is what keeps `start`
/// (which binds the socket) and `status` / `stop` / `reload` (which
/// connect to it) agreeing on a single path regardless of where each
/// invocation's CWD happens to be — the daemon launched from `~` and
/// the `status` run from `/tmp` both resolve to the same file under
/// `~/.bitrouter/`.
pub fn resolve_socket_path(config_path: &Path, cfg_socket: &str) -> PathBuf {
    let raw = Path::new(cfg_socket);
    if raw.is_absolute() {
        return raw.to_path_buf();
    }
    // Strip a leading `./` so the joined path doesn't render as
    // `/a/b/./c` — `Path::join` is faithful, not normalising.
    let stripped = raw.strip_prefix("./").unwrap_or(raw);
    let parent = config_path.parent().filter(|p| !p.as_os_str().is_empty());
    match parent {
        Some(dir) => dir.join(stripped),
        None => stripped.to_path_buf(),
    }
}

/// True when `err` came from connecting to a control endpoint that nothing
/// is listening on — i.e. the daemon is stopped, not that something
/// else went wrong. Inspects the chain for an io error of kind
/// `NotFound` (socket file / pipe absent) or `ConnectionRefused` (endpoint
/// present but no acceptor, e.g. stale after a crash). On Windows a missing
/// named pipe surfaces as `ERROR_FILE_NOT_FOUND`, which maps to `NotFound`.
pub fn is_not_reachable(err: &anyhow::Error) -> bool {
    err.chain().any(|cause| {
        cause
            .downcast_ref::<std::io::Error>()
            .map(|io| {
                matches!(
                    io.kind(),
                    std::io::ErrorKind::NotFound | std::io::ErrorKind::ConnectionRefused
                )
            })
            .unwrap_or(false)
    })
}

/// The daemon's self-reported readiness snapshot, returned by
/// [`probe_status`]. A successful probe means the control socket is bound
/// and the app is assembled (the model count is populated), which is exactly
/// the signal `bro status` reports.
#[derive(Debug, Clone)]
pub struct ReadyInfo {
    /// The daemon's process id.
    pub pid: u32,
    /// The HTTP listen address.
    pub listen: String,
    /// Count of routable models.
    pub models: usize,
}

/// Resolve the control-socket path for a `(source, cfg)` pair. For a `File`
/// source the configured `server.control_socket` is resolved against the
/// config file's directory; for a `Default` source the socket lives at
/// `<home>/bitrouter.sock`. Single source of truth shared by `serve`,
/// `start`, and `spawn`.
pub fn socket_path_for(
    source: &crate::paths::ConfigSource,
    cfg: &bitrouter_sdk::config::Config,
) -> PathBuf {
    match source {
        crate::paths::ConfigSource::File(path) => {
            resolve_socket_path(path, &cfg.server.control_socket)
        }
        crate::paths::ConfigSource::Default { home } => home.join("bitrouter.sock"),
    }
}

/// One-shot readiness probe: send `Status` and classify the reply.
/// `Ok(Some(_))` = the daemon answered (it is up); `Ok(None)` = nothing is
/// listening yet (not reachable — keep waiting); `Err` = a daemon is there but
/// the exchange failed (a real error worth surfacing).
pub async fn probe_status(socket: &Path) -> Result<Option<ReadyInfo>> {
    match send_command(socket, &DaemonCommand::Status).await {
        Ok(DaemonResponse::Status {
            pid,
            listen,
            models,
            ..
        }) => Ok(Some(ReadyInfo {
            pid,
            listen,
            models,
        })),
        Ok(DaemonResponse::Error { message }) => Err(anyhow::anyhow!(message)),
        Ok(other) => Err(anyhow::anyhow!("unexpected response: {other:?}")),
        Err(e) if is_not_reachable(&e) => Ok(None),
        Err(e) => Err(e),
    }
}

// ===== server side =====

/// Run the control listener until a `Stop` command is received (then it
/// returns) — run this alongside `App::serve` under a `tokio::select!`.
///
/// `listen` is the HTTP address (reported in `Status`); the endpoint is bound
/// at `socket_path` (a Unix socket file, or a named pipe derived from the path
/// on Windows) and torn down on return.
pub async fn run_control_socket(
    socket_path: PathBuf,
    app: Arc<App>,
    listen: String,
    reloader: Arc<dyn DaemonReloader>,
    observe: Arc<dyn ObserveStatusProvider>,
    metering: MeteringStore,
) -> Result<()> {
    run_control_socket_with_acp_runtime(
        socket_path,
        app,
        listen,
        reloader,
        observe,
        AcpControlPlane {
            runtime: Arc::new(AcpRuntime::new()),
            metering,
            inventory: None,
            evolution: None,
        },
    )
    .await
}

/// Run the control listener with the same ACP route runtime used by the model
/// request pipeline.
/// Authenticated ACP control state the control socket shares with the model
/// request pipeline.
///
/// One value rather than two parameters: `dispatch` already carries seven, and
/// an eighth trips `clippy::too_many_arguments`, which the house rules forbid
/// silencing.
#[derive(Clone)]
pub struct AcpControlPlane {
    /// In-memory route leases, keyed by principal and controller.
    pub runtime: Arc<AcpRuntime>,
    /// Settled-request store behind session-attributed spend.
    pub metering: MeteringStore,
    /// The inventory installed in this daemon's model request pipeline.
    pub inventory: Option<crate::evolution::inventory::GatewayInventory>,
    /// Live route validation and checkpoint worker status, when installed.
    pub evolution: Option<crate::evolution::runtime::EvolutionRuntime>,
}

pub async fn run_control_socket_with_acp_runtime(
    socket_path: PathBuf,
    app: Arc<App>,
    listen: String,
    reloader: Arc<dyn DaemonReloader>,
    observe: Arc<dyn ObserveStatusProvider>,
    acp: AcpControlPlane,
) -> Result<()> {
    run_control_socket_with_acp_runtime_and_administration(
        socket_path,
        app,
        listen,
        reloader,
        observe,
        acp,
        None,
    )
    .await
}

/// Run the local control socket with optional live administration ports.
///
/// Existing embeddings retain the historical control-socket surface through
/// [`run_control_socket_with_acp_runtime`].  The app assembly injects these
/// ports only after it has constructed the running routing and policy state.
pub async fn run_control_socket_with_acp_runtime_and_administration(
    socket_path: PathBuf,
    app: Arc<App>,
    listen: String,
    reloader: Arc<dyn DaemonReloader>,
    observe: Arc<dyn ObserveStatusProvider>,
    acp: AcpControlPlane,
    administration: Option<Administration>,
) -> Result<()> {
    let listener = bind_control_socket(&socket_path).await?;
    run_bound_control_socket(
        listener,
        app,
        listen,
        reloader,
        observe,
        acp,
        LocalControlOptions {
            administration,
            gate: crate::daemon_handoff::HandoffGate::default(),
            cli_owned: false,
        },
    )
    .await
}

/// Optional owner-scoped services shared with the HTTP listener.
pub struct LocalControlOptions {
    pub administration: Option<Administration>,
    pub gate: crate::daemon_handoff::HandoffGate,
    pub cli_owned: bool,
}

/// Bound local endpoint owned by one foreground startup attempt.
pub(crate) struct BoundControlSocket {
    listener: transport::ControlListener,
}

pub(crate) async fn bind_control_socket(path: &Path) -> Result<BoundControlSocket> {
    Ok(BoundControlSocket {
        listener: transport::bind(path).await?,
    })
}

pub(crate) async fn run_bound_control_socket(
    bound: BoundControlSocket,
    app: Arc<App>,
    listen: String,
    reloader: Arc<dyn DaemonReloader>,
    observe: Arc<dyn ObserveStatusProvider>,
    acp: AcpControlPlane,
    options: LocalControlOptions,
) -> Result<()> {
    let supervisor = match &options.administration {
        Some(administration) => Some(
            crate::supervisor::Supervisor::open(
                administration.source.clone(),
                administration.routing.clone(),
            )
            .await?,
        ),
        None => None,
    };
    let services = LocalControlServices {
        administration: options.administration,
        supervisor,
        gate: options.gate,
        cli_owned: options.cli_owned,
        handoff_instance_id: reloader
            .reload_state()
            .map(|state| state.server_instance_id)
            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
    };
    let mut listener = bound.listener;
    let result = accept_loop(
        &mut listener,
        &app,
        &listen,
        &reloader,
        &observe,
        &acp,
        &services,
    )
    .await;
    if let Some(supervisor) = &services.supervisor {
        supervisor.shutdown().await;
    }
    listener.cleanup().await;
    result
}

#[derive(Clone)]
struct LocalControlServices {
    administration: Option<Administration>,
    supervisor: Option<crate::supervisor::Supervisor>,
    gate: crate::daemon_handoff::HandoffGate,
    cli_owned: bool,
    handoff_instance_id: String,
}

async fn active_recording_count(acp: &AcpControlPlane) -> Result<Option<u64>> {
    use sea_orm::{ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter};

    let Some(inventory) = &acp.inventory else {
        return Ok(None);
    };
    let count = crate::acp_trajectory::entities::connections::Entity::find()
        .filter(crate::acp_trajectory::entities::connections::Column::State.eq("recording"))
        .count(&inventory.db)
        .await?;
    Ok(Some(count))
}

async fn accept_loop(
    listener: &mut transport::ControlListener,
    app: &Arc<App>,
    listen: &str,
    reloader: &Arc<dyn DaemonReloader>,
    observe: &Arc<dyn ObserveStatusProvider>,
    acp: &AcpControlPlane,
    services: &LocalControlServices,
) -> Result<()> {
    let (stop_tx, mut stop_rx) = tokio::sync::mpsc::unbounded_channel();
    loop {
        tokio::select! {
            accepted = listener.accept() => {
                let stream = accepted?;
                let app = app.clone();
                let listen = listen.to_string();
                let reloader = reloader.clone();
                let observe = observe.clone();
                let acp = acp.clone();
                let services = services.clone();
                let stop_tx = stop_tx.clone();
                tokio::spawn(async move {
                    match handle_connection(
                        stream,
                        &app,
                        &listen,
                        &reloader,
                        &observe,
                        &acp,
                        &services,
                    )
                    .await
                    {
                        Ok(true) => {
                            let _ = stop_tx.send(());
                        }
                        Ok(false) => {}
                        Err(error) => tracing::warn!(%error, "local control connection failed"),
                    }
                });
            }
            stopped = stop_rx.recv() => {
                if stopped.is_some() {
                    tracing::info!("stop command received — shutting down");
                }
                return Ok(());
            }
        }
    }
}

/// Handle one connection. Returns `Ok(true)` if it was a `Stop` command.
///
/// Generic over the concrete stream so the same logic drives a Unix
/// `UnixStream` and a Windows `NamedPipeServer` — both implement the tokio
/// async IO traits.
async fn handle_connection<S>(
    stream: S,
    app: &Arc<App>,
    listen: &str,
    reloader: &Arc<dyn DaemonReloader>,
    observe: &Arc<dyn ObserveStatusProvider>,
    acp: &AcpControlPlane,
    services: &LocalControlServices,
) -> Result<bool>
where
    S: AsyncRead + AsyncWrite + Unpin,
{
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    if reader.read_line(&mut line).await? == 0 {
        return Ok(false); // client hung up without sending anything
    }
    let command: DaemonCommand = match serde_json::from_str(line.trim()) {
        Ok(c) => c,
        Err(e) => {
            write_response(
                reader.get_mut(),
                &DaemonResponse::Error {
                    message: format!("invalid command: {e}"),
                },
            )
            .await?;
            return Ok(false);
        }
    };

    let is_stop = matches!(
        command,
        DaemonCommand::Stop | DaemonCommand::HandoffStop { .. }
    );
    let admission = if matches!(
        command,
        DaemonCommand::Stop
            | DaemonCommand::Status
            | DaemonCommand::HandoffHello
            | DaemonCommand::HandoffPrepare
            | DaemonCommand::HandoffAbort { .. }
            | DaemonCommand::HandoffStop { .. }
    ) {
        None
    } else {
        match services.gate.admit() {
            Some(admission) => Some(admission),
            None => {
                write_response(
                    reader.get_mut(),
                    &DaemonResponse::Error {
                        message: "daemon handoff in progress; retry".to_string(),
                    },
                )
                .await?;
                return Ok(false);
            }
        }
    };
    let response = dispatch(command, app, listen, reloader, observe, acp, services).await;
    write_response(reader.get_mut(), &response).await?;
    drop(admission);
    Ok(is_stop && matches!(response, DaemonResponse::Ok))
}

async fn dispatch(
    command: DaemonCommand,
    app: &Arc<App>,
    listen: &str,
    reloader: &Arc<dyn DaemonReloader>,
    observe: &Arc<dyn ObserveStatusProvider>,
    acp: &AcpControlPlane,
    services: &LocalControlServices,
) -> DaemonResponse {
    match command {
        DaemonCommand::Stop => DaemonResponse::Ok,
        DaemonCommand::HandoffHello => DaemonResponse::HandoffHello {
            version: crate::VERSION.to_string(),
            protocol: 1,
            instance_id: services.handoff_instance_id.clone(),
            build_id: crate::HANDOFF_BUILD_ID.to_string(),
            cli_owned: services.cli_owned,
        },
        DaemonCommand::HandoffPrepare => {
            if !services.cli_owned {
                return DaemonResponse::HandoffBusy {
                    reason: "this daemon is managed by an external process".to_string(),
                };
            }
            let token = match services.gate.prepare() {
                Ok(token) => token,
                Err(reason) => {
                    return DaemonResponse::HandoffBusy {
                        reason: reason.to_string(),
                    };
                }
            };
            let active_runs = match &services.supervisor {
                Some(supervisor) => supervisor.active_run_count().await,
                None => {
                    services.gate.abort(&token);
                    return DaemonResponse::HandoffBusy {
                        reason: "supervisor activity is unknown".to_string(),
                    };
                }
            };
            let leases = acp.runtime.active_lease_count();
            let active_recordings = match active_recording_count(acp).await {
                Ok(Some(count)) => count,
                Ok(None) => {
                    services.gate.abort(&token);
                    return DaemonResponse::HandoffBusy {
                        reason: "external ACP controller activity is unknown".to_string(),
                    };
                }
                Err(error) => {
                    services.gate.abort(&token);
                    return DaemonResponse::HandoffBusy {
                        reason: format!(
                            "external ACP controller activity could not be read: {error}"
                        ),
                    };
                }
            };
            if active_runs != 0 || leases != 0 || active_recordings != 0 {
                services.gate.abort(&token);
                DaemonResponse::HandoffBusy {
                    reason: format!(
                        "{active_runs} active supervised runs, {leases} ACP route leases, {active_recordings} active ACP recordings"
                    ),
                }
            } else {
                DaemonResponse::HandoffReady { token }
            }
        }
        DaemonCommand::HandoffAbort { token } => {
            if services.gate.abort(&token) {
                DaemonResponse::Ok
            } else {
                DaemonResponse::Error {
                    message: "handoff token is invalid".to_string(),
                }
            }
        }
        DaemonCommand::HandoffStop { token } => {
            if services.gate.valid(&token) {
                DaemonResponse::Ok
            } else {
                DaemonResponse::Error {
                    message: "handoff token is invalid or expired".to_string(),
                }
            }
        }
        DaemonCommand::Evolution { operation } => {
            let Some(runtime) = &acp.evolution else {
                return DaemonResponse::Error {
                    message: "checkpoint evolution is unavailable on this daemon".into(),
                };
            };
            match runtime.operate("local", operation).await {
                Ok(report) => DaemonResponse::Evolution {
                    report: Box::new(report),
                },
                Err(error) => DaemonResponse::Error {
                    message: error.to_string(),
                },
            }
        }
        DaemonCommand::AcpRecordingRegister {
            connection_id,
            api_principal,
            controller_instance_id,
        } => {
            let Some(inventory) = &acp.inventory else {
                return DaemonResponse::Error {
                    message: "gateway request inventory is unavailable on this daemon".into(),
                };
            };
            match inventory
                .register_capture(&connection_id, &api_principal, &controller_instance_id)
                .await
            {
                Ok(registration) => DaemonResponse::AcpRecordingRegistered { registration },
                Err(error) => DaemonResponse::Error {
                    message: error.to_string(),
                },
            }
        }
        DaemonCommand::Reload { env } => match reloader.reload_with_env(env).await {
            Ok(()) => {
                tracing::info!("reload succeeded");
                DaemonResponse::Ok
            }
            Err(e) => DaemonResponse::Error {
                message: format!("reload failed: {e}"),
            },
        },
        DaemonCommand::ReloadState => match reloader.reload_state() {
            Some(state) => DaemonResponse::ReloadState { state },
            None => DaemonResponse::Error {
                message: "reload state is unavailable on this daemon".to_string(),
            },
        },
        DaemonCommand::Status => {
            let routable = app
                .language_model()
                .map(|p| p.routing_table().list_models())
                .unwrap_or_default();
            // One pass over the table the model count already walks: the
            // distinct providers behind it, which is what an operator or an
            // agent actually wants to see next to "42 routable".
            let providers: Vec<String> = routable
                .iter()
                .flat_map(|m| m.providers.iter().cloned())
                .collect::<std::collections::BTreeSet<_>>()
                .into_iter()
                .collect();
            let router_state = router_state(&services.administration).await;
            let config_state = reloader.configuration_state().await;
            let (operations, drain_in_progress) = services.gate.observed_activity();
            let handoff_activity = HandoffActivity {
                operations,
                supervised_runs: match &services.supervisor {
                    Some(supervisor) => Some(supervisor.active_run_count().await),
                    None => None,
                },
                route_leases: acp.runtime.active_lease_count(),
                acp_recordings: active_recording_count(acp).await.ok().flatten(),
                drain_in_progress,
            };
            DaemonResponse::Status {
                pid: std::process::id(),
                daemon_version: Some(crate::VERSION.to_string()),
                handoff_protocol: Some(1),
                handoff_build_id: Some(crate::HANDOFF_BUILD_ID.to_string()),
                handoff_activity: Some(handoff_activity),
                cli_owned: Some(services.cli_owned),
                listen: listen.to_string(),
                models: routable.len(),
                providers,
                saved_routers: router_state.as_ref().and_then(|state| state.saved.clone()),
                running_routers: router_state
                    .as_ref()
                    .and_then(|state| state.running.clone()),
                router_restart_required: router_state.and_then(|state| state.restart_required),
                config_state: config_state.map(Box::new),
            }
        }
        DaemonCommand::Models => {
            let routers = router_state(&services.administration)
                .await
                .and_then(|state| state.running);
            DaemonResponse::Models {
                // The same read `Status` counts, returned whole. A daemon with no
                // language-model pipeline routes nothing, which is an empty
                // catalog rather than an error.
                models: app
                    .language_model()
                    .map(|p| p.routing_table().list_models())
                    .unwrap_or_default(),
                routers,
            }
        }
        DaemonCommand::AcpSessionSpend {
            api_principal,
            controller_instance_id,
            session_id,
        } => {
            if api_principal.trim().is_empty()
                || controller_instance_id.trim().is_empty()
                || session_id.trim().is_empty()
            {
                return DaemonResponse::Error {
                    message: "principal, controller and session must not be empty".to_string(),
                };
            }
            // Unbounded on purpose. This launcher mints a fresh controller id
            // per process, so in practice "since the epoch" is "this
            // controller's lifetime" — but that is a property of how we mint,
            // not of the identifier, which is a claim nothing verifies. What
            // the window must not do is reset a session's cumulative figure at
            // a month boundary the way the rolling ones would.
            let window = TimeWindow::Custom {
                start: DateTime::<Utc>::UNIX_EPOCH,
                end: Utc::now(),
            };
            match acp
                .metering
                .spend_summary_for_acp_session(
                    &api_principal,
                    &controller_instance_id,
                    &session_id,
                    window,
                )
                .await
            {
                Ok(summary) => DaemonResponse::AcpSessionSpend {
                    spend_micro_usd: summary.spend_micro_usd,
                    requests: summary.requests,
                    unpriced: summary.unpriced,
                },
                Err(error) => DaemonResponse::Error {
                    message: format!("session spend is unavailable: {error}"),
                },
            }
        }
        DaemonCommand::AcpControllerCleanup {
            api_principal,
            controller_instance_id,
        } => {
            acp.runtime
                .remove_controller(&api_principal, &controller_instance_id);
            DaemonResponse::Ok
        }
        DaemonCommand::AcpRouteList {
            api_principal,
            controller_instance_id,
            session_id,
        } => {
            if api_principal.trim().is_empty()
                || controller_instance_id.trim().is_empty()
                || session_id.trim().is_empty()
            {
                DaemonResponse::Error {
                    message: "API principal, controller id, and session id must not be empty"
                        .to_string(),
                }
            } else {
                let current = acp
                    .runtime
                    .current_route(&api_principal, &controller_instance_id, &session_id)
                    .map(|lease| lease.route().to_string());
                DaemonResponse::AcpRouteState {
                    available: route_suggestions(app),
                    scope: if current.is_some() {
                        "session"
                    } else {
                        "default"
                    }
                    .to_string(),
                    current,
                }
            }
        }
        DaemonCommand::AcpRouteSet {
            api_principal,
            controller_instance_id,
            session_id,
            route,
        } => {
            let Some(pipeline) = app.language_model() else {
                return DaemonResponse::Error {
                    message: "no language_model pipeline configured".to_string(),
                };
            };
            if api_principal.trim().is_empty()
                || controller_instance_id.trim().is_empty()
                || session_id.trim().is_empty()
                || route.trim().is_empty()
            {
                return DaemonResponse::Error {
                    message:
                        "API principal, controller id, session id, and route must not be empty"
                            .to_string(),
                };
            }
            if let Err(error) = pipeline
                .routing_table()
                .route_chain(&route, &RoutingPrefs::default(), &CallerContext::local())
                .await
            {
                return DaemonResponse::Error {
                    message: format!("route is not available: {error}"),
                };
            }
            match acp.runtime.set_route(
                &api_principal,
                &controller_instance_id,
                &session_id,
                &route,
            ) {
                Ok(lease) => DaemonResponse::AcpRouteState {
                    available: route_suggestions(app),
                    current: Some(lease.route().to_string()),
                    scope: "session".to_string(),
                },
                Err(message) => DaemonResponse::Error { message },
            }
        }
        DaemonCommand::AcpRouteReset {
            api_principal,
            controller_instance_id,
            session_id,
        } => {
            if api_principal.trim().is_empty()
                || controller_instance_id.trim().is_empty()
                || session_id.trim().is_empty()
            {
                DaemonResponse::Error {
                    message: "API principal, controller id, and session id must not be empty"
                        .to_string(),
                }
            } else {
                acp.runtime
                    .reset_route(&api_principal, &controller_instance_id, &session_id);
                DaemonResponse::AcpRouteState {
                    available: route_suggestions(app),
                    current: None,
                    scope: "default".to_string(),
                }
            }
        }
        DaemonCommand::Route { model } => {
            let Some(pipeline) = app.language_model() else {
                return DaemonResponse::Error {
                    message: "no language_model pipeline configured".to_string(),
                };
            };
            let table = pipeline.routing_table();
            let resolution = match table.resolve_model(&model).await {
                Ok(resolution) => resolution,
                Err(error) => {
                    return DaemonResponse::Error {
                        message: error.to_string(),
                    };
                }
            };
            let router_source = services.administration.as_ref().and_then(|administration| {
                resolution.router.as_ref().and_then(|identity| {
                    crate::actions::route::router_source(
                        &administration.routing.snapshot_config(),
                        &identity.router_id,
                    )
                })
            });
            let policy_report = services
                .administration
                .as_ref()
                .map(|administration| administration.policy.administration_snapshot());
            let candidate_models = crate::actions::route::policy_candidate_models(
                policy_report.as_ref(),
                resolution.policy.as_deref(),
                &resolution.clean_model,
            );
            let candidate_models = crate::actions::route::resolvable_policy_candidates(
                table.as_ref(),
                candidate_models,
                &resolution.prefs,
            )
            .await;
            let chain = if resolution.policy.is_some() {
                Ok(Vec::new())
            } else {
                table
                    .route_resolved(
                        &resolution.clean_model,
                        &resolution.prefs,
                        &CallerContext::local(),
                    )
                    .await
            };
            match chain {
                Ok(chain) => DaemonResponse::Route {
                    chain: chain
                        .into_iter()
                        .map(|t| RouteHop {
                            provider: t.provider_name,
                            service_id: t.service_id,
                            api_protocol: format!("{:?}", t.api_protocol).to_lowercase(),
                        })
                        .collect(),
                    resolved_model: Some(resolution.clean_model),
                    router: resolution.router,
                    router_source,
                    bound_policy: resolution.policy,
                    candidate_models,
                },
                Err(error) => DaemonResponse::Error {
                    message: error.to_string(),
                },
            }
        }
        DaemonCommand::ObserveStatus => DaemonResponse::ObserveStatus {
            payload: observe.status(),
        },
        DaemonCommand::Inspect { inspection } => {
            inspection_response(inspection, &services.administration).await
        }
        DaemonCommand::Sessions { request } => {
            let Some(supervisor) = &services.supervisor else {
                return DaemonResponse::Error {
                    message: "supervised sessions are unavailable on this daemon".to_string(),
                };
            };
            match supervisor.dispatch(request).await {
                Ok(response) => DaemonResponse::Sessions { response },
                Err(error) => DaemonResponse::Error {
                    message: format!("supervised session request failed: {error:#}"),
                },
            }
        }
    }
}

struct RouterState {
    saved: Option<Vec<crate::actions::models::RouterStatus>>,
    running: Option<Vec<crate::actions::models::RouterStatus>>,
    restart_required: Option<bool>,
}

fn router_configuration_restart_required(
    saved: &bitrouter_sdk::config::Config,
    running: &bitrouter_sdk::config::Config,
) -> bool {
    saved.routers != running.routers
}

async fn router_state(administration: &Option<Administration>) -> Option<RouterState> {
    let administration = administration.as_ref()?;
    let running = administration.routing.snapshot_config();
    let policy_report = administration.policy.administration_snapshot();
    let running_statuses = crate::actions::models::router_statuses(
        &running,
        crate::actions::models::PolicyReadiness::Known(&policy_report),
    )
    .await
    .ok();
    let saved = crate::reload::inspect_configuration_baseline(&administration.source)
        .await
        .ok();
    let saved_statuses = match &saved {
        Some(saved) => {
            crate::actions::models::disk_router_statuses_for_config(
                saved.config(),
                &administration.source,
            )
            .await
        }
        None => None,
    };
    let restart_required = saved
        .as_ref()
        .map(|saved| router_configuration_restart_required(saved.config(), &running));
    Some(RouterState {
        saved: saved_statuses,
        running: running_statuses,
        restart_required,
    })
}

async fn inspection_response(
    inspection: DaemonInspection,
    administration: &Option<Administration>,
) -> DaemonResponse {
    let Some(administration) = administration else {
        return DaemonResponse::Error {
            message: "live administration reads are unavailable on this daemon".to_string(),
        };
    };
    match inspection {
        DaemonInspection::Providers => DaemonResponse::Inspection {
            report: DaemonInspectionReport::Providers(administration.providers()),
        },
        DaemonInspection::Agents => DaemonResponse::Inspection {
            report: DaemonInspectionReport::Agents(administration.agents()),
        },
        DaemonInspection::Observe => DaemonResponse::Inspection {
            report: DaemonInspectionReport::Observe(administration.observe()),
        },
        DaemonInspection::Policy { input } => match administration.policy(input).await {
            Ok(report) => DaemonResponse::Inspection {
                report: DaemonInspectionReport::Policy(report),
            },
            Err(error) => DaemonResponse::Error {
                message: format!("policy inspection failed: {error}"),
            },
        },
    }
}

fn route_suggestions(app: &Arc<App>) -> Vec<String> {
    app.language_model()
        .map(|pipeline| {
            pipeline
                .routing_table()
                .list_models()
                .into_iter()
                .map(|model| model.id)
                .collect()
        })
        .unwrap_or_default()
}

async fn write_response<S>(stream: &mut S, response: &DaemonResponse) -> Result<()>
where
    S: AsyncWrite + Unpin,
{
    let mut json = serde_json::to_string(response).context("serialising daemon response")?;
    json.push('\n');
    stream.write_all(json.as_bytes()).await?;
    stream.flush().await?;
    Ok(())
}

// ===== client side =====

/// Connect to a running daemon's control endpoint and hand back the raw
/// stream. Used by [`send_command`] and by integration tests that need to
/// push bytes the normal client wouldn't (e.g. malformed input). Fails
/// clearly if no daemon is listening.
pub async fn connect_control(socket_path: &Path) -> Result<impl AsyncRead + AsyncWrite + Unpin> {
    transport::connect(socket_path).await
}

/// True when something is already bound to the control endpoint at `path`
/// (i.e. a daemon is — or was — running). On Unix this is "the socket file
/// exists"; on Windows it probes the named pipe. `restart` uses it to decide
/// whether to send a `Stop` and to wait for the old daemon to release the
/// endpoint before the replacement binds.
pub fn endpoint_in_use(path: &Path) -> bool {
    transport::endpoint_in_use(path)
}

/// Connect to a running daemon's control endpoint, send `command`, return its
/// response. Fails clearly if no daemon is listening.
pub async fn send_command(socket_path: &Path, command: &DaemonCommand) -> Result<DaemonResponse> {
    let stream = transport::connect(socket_path).await?;
    let mut reader = BufReader::new(stream);
    let mut json = serde_json::to_string(command).context("serialising command")?;
    json.push('\n');
    reader.get_mut().write_all(json.as_bytes()).await?;
    reader.get_mut().flush().await?;

    let mut line = String::new();
    reader
        .read_line(&mut line)
        .await
        .context("reading daemon response")?;
    if line.trim().is_empty() {
        anyhow::bail!("daemon closed the connection without responding");
    }
    serde_json::from_str(line.trim()).context("parsing daemon response")
}

// ===== platform transport =====

/// Unix transport: a `tokio::net::UnixListener` bound to the socket path,
/// tightened to mode `0600`. The socket file is removed on teardown.
#[cfg(unix)]
pub(crate) mod transport {
    use std::path::{Path, PathBuf};

    use anyhow::{Context, Result};
    use tokio::net::{UnixListener, UnixStream};

    /// Server-side listener wrapping a bound Unix socket.
    pub struct ControlListener {
        listener: UnixListener,
        path: PathBuf,
        identity: (u64, u64),
        _lock: std::fs::File,
    }

    /// Bind one owner-only endpoint, preserving active endpoints and unrelated files.
    pub async fn bind(path: &Path) -> Result<ControlListener> {
        use std::os::unix::fs::{FileTypeExt, MetadataExt, OpenOptionsExt, PermissionsExt};

        // Keep the lock file in place: unlinking it would let a new process lock
        // a different inode while another startup still owns the old one.
        let lock_path = path.with_extension("sock.lock");
        let lock = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .mode(0o600)
            .open(&lock_path)
            .with_context(|| format!("opening control lock {}", lock_path.display()))?;
        lock.try_lock()
            .with_context(|| format!("control socket {} is already owned", path.display()))?;
        match tokio::fs::symlink_metadata(path).await {
            Ok(metadata) => {
                anyhow::ensure!(
                    metadata.file_type().is_socket(),
                    "refusing to replace non-socket path {}",
                    path.display()
                );
                match tokio::time::timeout(
                    std::time::Duration::from_millis(250),
                    UnixStream::connect(path),
                )
                .await
                {
                    Ok(Err(error)) if error.kind() == std::io::ErrorKind::ConnectionRefused => {}
                    _ => anyhow::bail!("control socket {} is already in use", path.display()),
                }
                let current = tokio::fs::symlink_metadata(path).await?;
                anyhow::ensure!(
                    (current.dev(), current.ino()) == (metadata.dev(), metadata.ino()),
                    "control socket changed during startup: {}",
                    path.display()
                );
                tokio::fs::remove_file(path).await?;
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error).context("inspecting control endpoint"),
        }
        let listener = UnixListener::bind(path)
            .with_context(|| format!("binding control socket {}", path.display()))?;
        let metadata = std::fs::symlink_metadata(path)?;
        let bound = ControlListener {
            listener,
            path: path.to_path_buf(),
            identity: (metadata.dev(), metadata.ino()),
            _lock: lock,
        };
        tokio::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
            .await
            .with_context(|| format!("chmod 0600 {}", path.display()))?;
        tracing::info!(socket = %path.display(), "control socket listening (mode 0600)");
        Ok(bound)
    }

    impl Drop for ControlListener {
        fn drop(&mut self) {
            use std::os::unix::fs::MetadataExt;
            if let Ok(metadata) = std::fs::symlink_metadata(&self.path)
                && (metadata.dev(), metadata.ino()) == self.identity
            {
                let _ = std::fs::remove_file(&self.path);
            }
        }
    }

    impl ControlListener {
        /// Accept the next client connection.
        pub async fn accept(&mut self) -> Result<UnixStream> {
            let (stream, _addr) = self
                .listener
                .accept()
                .await
                .context("accepting control-socket connection")?;
            Ok(stream)
        }

        /// Remove the socket file on shutdown.
        pub async fn cleanup(self) {
            drop(self);
        }
    }

    /// Connect to the control socket.
    pub async fn connect(path: &Path) -> Result<UnixStream> {
        UnixStream::connect(path).await.with_context(|| {
            format!(
                "connecting to {} — is the daemon running? (`{cli} start`)",
                path.display(),
                cli = bitrouter_sdk::invocation::name()
            )
        })
    }

    /// On Unix the socket is a real file: its presence means a daemon bound it.
    pub fn endpoint_in_use(path: &Path) -> bool {
        path.exists()
    }
}

/// Windows transport: a named pipe whose name is derived from the control
/// path.
///
/// Security: the pipe is created with the **default** security descriptor.
/// That grants full control to the creating user, `LocalSystem` and
/// administrators, and only `GENERIC_READ` to `Everyone` — so a co-tenant
/// non-admin user cannot perform the read+write open a control client needs
/// (it is denied with `ERROR_ACCESS_DENIED`) and therefore cannot issue
/// `Stop` / `Reload`. This is the practical analog of the Unix socket's
/// `0600` mode. Tightening to a literally owner-only DACL would require a
/// custom `SECURITY_ATTRIBUTES` pointer via
/// `ServerOptions::create_with_security_attributes_raw`, which is `unsafe` —
/// and this crate is `#![forbid(unsafe_code)]`, so we rely on the default
/// descriptor instead.
#[cfg(windows)]
pub(crate) mod transport {
    use std::path::Path;
    use std::time::Duration;

    use anyhow::{Context, Result};
    use tokio::net::windows::named_pipe::{
        ClientOptions, NamedPipeClient, NamedPipeServer, ServerOptions,
    };

    /// `ERROR_PIPE_BUSY` — every pipe instance is currently connected. The
    /// server re-arms a fresh instance immediately after each accept, so this
    /// is a transient race the client retries through.
    const ERROR_PIPE_BUSY: i32 = 231;

    /// Map a filesystem-style control path to a named-pipe name. An explicit
    /// `\\.\pipe\…` path is honoured verbatim; any other path is hashed into a
    /// stable, valid pipe name so the daemon and every client agree on the
    /// same endpoint regardless of the launcher's working directory. Windows
    /// paths are case-insensitive, so the name is case-folded before hashing.
    pub fn pipe_name(path: &Path) -> String {
        let raw = path.to_string_lossy();
        let lower = raw.to_ascii_lowercase();
        if lower.starts_with(r"\\.\pipe\") || lower.starts_with(r"\\?\pipe\") {
            return raw.into_owned();
        }
        use sha2::{Digest, Sha256};
        let mut hasher = Sha256::new();
        hasher.update(lower.as_bytes());
        let digest = hasher.finalize();
        format!(r"\\.\pipe\bitrouter-{}", hex::encode(&digest[..16]))
    }

    /// Server-side listener wrapping the currently-waiting pipe instance.
    pub struct ControlListener {
        name: String,
        /// The next instance waiting for a client. `accept` hands this out and
        /// re-arms a replacement so there is no window with no listener.
        server: NamedPipeServer,
    }

    /// Create the first pipe instance. `first_pipe_instance(true)` makes this
    /// fail loudly if another process already owns the name (e.g. a second
    /// daemon racing the same control path) rather than silently sharing it.
    pub async fn bind(path: &Path) -> Result<ControlListener> {
        let name = pipe_name(path);
        let server = ServerOptions::new()
            .first_pipe_instance(true)
            .create(&name)
            .with_context(|| format!("creating control pipe {name}"))?;
        tracing::info!(pipe = %name, "control pipe listening");
        Ok(ControlListener { name, server })
    }

    impl ControlListener {
        /// Wait for the next client, then re-arm a fresh instance and return
        /// the connected one. Mirrors a `UnixListener::accept`.
        pub async fn accept(&mut self) -> Result<NamedPipeServer> {
            self.server
                .connect()
                .await
                .context("waiting for control-pipe client")?;
            let next = ServerOptions::new()
                .create(&self.name)
                .with_context(|| format!("re-arming control pipe {}", self.name))?;
            Ok(std::mem::replace(&mut self.server, next))
        }

        /// Nothing to unlink — a named pipe disappears when its last instance
        /// handle closes.
        pub async fn cleanup(self) {}
    }

    /// Connect to the control pipe, retrying briefly through the `ERROR_PIPE_BUSY`
    /// window the server opens while re-arming between clients.
    pub async fn connect(path: &Path) -> Result<NamedPipeClient> {
        let name = pipe_name(path);
        let mut last_busy: Option<std::io::Error> = None;
        for _ in 0..50 {
            match ClientOptions::new().open(&name) {
                Ok(client) => return Ok(client),
                Err(e) if e.raw_os_error() == Some(ERROR_PIPE_BUSY) => {
                    last_busy = Some(e);
                    tokio::time::sleep(Duration::from_millis(20)).await;
                }
                Err(e) => {
                    return Err(e).with_context(|| {
                        format!(
                            "connecting to {name} — is the daemon running? (`{cli} start`)",
                            cli = bitrouter_sdk::invocation::name()
                        )
                    });
                }
            }
        }
        let e = last_busy.unwrap_or_else(|| {
            std::io::Error::new(std::io::ErrorKind::TimedOut, "control pipe busy")
        });
        Err(e).with_context(|| format!("connecting to {name} — daemon stayed busy"))
    }

    /// Probe whether a daemon currently owns the pipe. A successful open (or a
    /// busy reply) means it does; "not found" means it has been released.
    pub fn endpoint_in_use(path: &Path) -> bool {
        let name = pipe_name(path);
        match ClientOptions::new().open(&name) {
            Ok(_client) => true,
            Err(e) => e.raw_os_error() == Some(ERROR_PIPE_BUSY),
        }
    }

    #[cfg(test)]
    mod tests {
        use super::pipe_name;
        use std::path::PathBuf;

        #[test]
        fn explicit_pipe_path_is_returned_verbatim() {
            // `\\.\pipe\…` is already a valid pipe name — pass it through so an
            // operator can override the derived name in `server.control_socket`.
            let p = PathBuf::from(r"\\.\pipe\bitrouter-custom");
            assert_eq!(pipe_name(&p), r"\\.\pipe\bitrouter-custom");
        }

        #[test]
        fn nt_style_pipe_path_is_returned_verbatim() {
            let p = PathBuf::from(r"\\?\pipe\bitrouter-custom");
            assert_eq!(pipe_name(&p), r"\\?\pipe\bitrouter-custom");
        }

        #[test]
        fn filesystem_path_maps_to_deterministic_pipe_name() {
            // The hash must be stable so the daemon (which binds) and the
            // client (which connects from a possibly-different CWD) agree.
            let a = pipe_name(&PathBuf::from(r"C:\Users\alice\.bitrouter\bitrouter.sock"));
            let b = pipe_name(&PathBuf::from(r"C:\Users\alice\.bitrouter\bitrouter.sock"));
            assert_eq!(a, b);
            assert!(
                a.starts_with(r"\\.\pipe\bitrouter-"),
                "unexpected pipe name {a}"
            );
        }

        #[test]
        fn case_differences_collapse_to_one_name() {
            // Windows paths are case-insensitive. If two invocations capitalise
            // the drive letter differently, they must still find the same pipe.
            let lower = pipe_name(&PathBuf::from(r"c:\users\alice\.bitrouter\bitrouter.sock"));
            let upper = pipe_name(&PathBuf::from(r"C:\Users\Alice\.BitRouter\bitrouter.sock"));
            assert_eq!(lower, upper);
        }

        #[test]
        fn different_paths_map_to_different_pipe_names() {
            let a = pipe_name(&PathBuf::from(r"C:\Users\alice\.bitrouter\bitrouter.sock"));
            let b = pipe_name(&PathBuf::from(r"C:\Users\bob\.bitrouter\bitrouter.sock"));
            assert_ne!(a, b);
        }
    }
}

/// Check one positive Unix PID without sending a signal.
///
/// Only a missing process (`ESRCH`) or an unrepresentable PID is considered
/// dead. Permission errors retain the PID as live, so startup does not reclaim
/// another user's PID file merely because this process cannot signal it.
#[cfg(unix)]
pub fn process_is_alive(pid: u32) -> bool {
    let Some(pid) = i32::try_from(pid)
        .ok()
        .and_then(rustix::process::Pid::from_raw)
    else {
        return false;
    };
    !matches!(
        rustix::process::test_kill_process(pid),
        Err(rustix::io::Errno::SRCH)
    )
}

/// Liveness check on Windows: `tasklist` filtered to the pid. `tasklist` ships
/// on every Windows install, so we shell out (rather than calling the Win32
/// API) to keep `apps/bitrouter` free of `unsafe`. When no process matches,
/// `tasklist` prints an informational line instead of a CSV row — so we look
/// for the quoted pid the CSV format emits (`"<pid>"`).
#[cfg(windows)]
pub fn process_is_alive(pid: u32) -> bool {
    let output = std::process::Command::new("tasklist")
        .args(["/FI", &format!("PID eq {pid}"), "/NH", "/FO", "CSV"])
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::null())
        .output();
    match output {
        Ok(out) => {
            let text = String::from_utf8_lossy(&out.stdout);
            text.contains(&format!("\"{pid}\""))
        }
        Err(_) => false,
    }
}

// ===== PID file =====

/// Write the current process id to `path`.
pub async fn write_pid_file(path: &Path) -> Result<()> {
    tokio::fs::write(path, std::process::id().to_string())
        .await
        .with_context(|| format!("writing pid file {}", path.display()))
}

/// Read a process id from a PID file, if it exists and is well-formed.
pub async fn read_pid_file(path: &Path) -> Option<u32> {
    let raw = tokio::fs::read_to_string(path).await.ok()?;
    raw.trim().parse().ok()
}

/// Remove a PID file, ignoring "not found".
pub async fn remove_pid_file(path: &Path) {
    let _ = tokio::fs::remove_file(path).await;
}

// ===== launch + readiness wait =====

/// How long [`start_and_wait`] polls for readiness before giving up. Sized for
/// a cold registry fetch + DB migrations on first run; the daemon keeps running
/// past this — the timeout only bounds how long the launcher blocks.
pub const DAEMON_READY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);

/// Poll cadence for the readiness loop.
const READY_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(100);

/// The result of launching a detached daemon and waiting for it to come up.
#[derive(Debug)]
pub enum DaemonStartOutcome {
    /// The daemon answered `Status` within the timeout. Carries its self-report.
    Ready(ReadyInfo),
    /// The process is alive but did not become ready before the deadline.
    NotReadyInTime {
        /// The launched child's process id.
        pid: u32,
    },
    /// The process exited during startup. Carries the exit-status string and
    /// the fresh tail of the daemon log (this run's output only).
    Exited {
        /// Rendered `ExitStatus`.
        status: String,
        /// Daemon log content written since launch.
        log_tail: String,
    },
}

/// Spawn `bro serve` as a detached background process writing to
/// `log_path` (append). Returns the child handle and the log's pre-spawn byte
/// length so the caller can quote only this run's output on early death.
///
/// Detach rationale: a new process group on Unix so the launcher shell's
/// SIGHUP (terminal close) does not propagate; DETACHED_PROCESS +
/// CREATE_NEW_PROCESS_GROUP on Windows so a console Ctrl-C is not delivered.
/// <https://doc.rust-lang.org/std/os/unix/process/trait.CommandExt.html#tymethod.process_group>
fn spawn_detached_serve(
    source: &crate::paths::ConfigSource,
    log_path: &Path,
) -> Result<(std::process::Child, u64)> {
    let exe = std::env::current_exe().context("locating current bitrouter binary")?;
    // Capture the log's pre-spawn size so we can quote *this run's* output back
    // to the user on early death instead of slurping stale content from prior
    // runs (the log is opened append-only).
    let log_size_before = std::fs::metadata(log_path).map(|m| m.len()).unwrap_or(0);
    let log = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(log_path)
        .with_context(|| format!("opening daemon log {}", log_path.display()))?;
    let log_err = log
        .try_clone()
        .context("duplicating log handle for stderr")?;

    let mut cmd = std::process::Command::new(&exe);
    cmd.arg("serve").arg("--managed-child");
    // For a `File` source pass `--config <abs path>` so the child loads the same
    // file even though it'll chdir to the home. For `Default` (zero-config) skip
    // the flag — the child re-runs `resolve_config` and lands on the same state.
    if let crate::paths::ConfigSource::File(path) = source {
        cmd.arg("--config").arg(path);
    }
    cmd.stdout(std::process::Stdio::from(log))
        .stderr(std::process::Stdio::from(log_err))
        .stdin(std::process::Stdio::null());
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        cmd.process_group(0);
    }
    #[cfg(windows)]
    {
        use std::os::windows::process::CommandExt;
        // Winbase.h constants — avoid pulling in a `windows`/`winapi` crate.
        const DETACHED_PROCESS: u32 = 0x0000_0008;
        const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200;
        cmd.creation_flags(DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP);
    }
    let child = cmd.spawn().with_context(|| {
        format!(
            "spawning detached `{} serve`",
            bitrouter_sdk::invocation::name()
        )
    })?;
    Ok((child, log_size_before))
}

/// Read the daemon log from `offset` to end. Returns "" on any read failure so
/// callers can fall back to a path-only hint. The pre-spawn `offset` ensures
/// only this run's content is quoted even though the log is append-only.
async fn read_log_tail(path: &Path, offset: u64) -> String {
    let bytes = match tokio::fs::read(path).await {
        Ok(b) => b,
        Err(_) => return String::new(),
    };
    let start = (offset as usize).min(bytes.len());
    String::from_utf8_lossy(&bytes[start..]).into_owned()
}

/// Print the daemon's tail-of-log to stderr as an indented, captioned block so
/// the user sees the actual failure inline instead of being pointed at a log
/// file they have to open separately. No-op when there is nothing useful to
/// show. Shared by `start` and `spawn`.
pub fn eprint_failure_log(log_path: &Path, content: &str) {
    let trimmed = content.trim_end();
    if trimmed.is_empty() {
        return;
    }
    let p = crate::style::Palette::for_stderr();
    eprintln!(
        "{dim}daemon log ({path}):{reset}",
        dim = p.dim,
        path = log_path.display(),
        reset = p.reset,
    );
    for line in trimmed.lines() {
        eprintln!("  {line}");
    }
    eprintln!();
}

/// Launch a detached `bro serve` and poll the control socket until it
/// answers `Status`, the process dies, or `timeout` elapses. The daemon keeps
/// running regardless of the outcome — this only reports what the launcher
/// observed. `socket` is the control-socket path to poll; pass `None` when it
/// could not be resolved (then only process-death is detectable).
///
/// Ensures the bitrouter home exists (the log lives inside it) but never chdirs
/// the calling process — only the child `serve` chdirs into the home.
pub async fn start_and_wait(
    source: &crate::paths::ConfigSource,
    log_path: &Path,
    socket: Option<&Path>,
    timeout: std::time::Duration,
) -> Result<DaemonStartOutcome> {
    crate::paths::ensure_home_directory(source.home())?;
    // Serialize local launchers across processes. Without this, two clients
    // can both observe no socket, start two daemons, and race the same SQLite
    // migrations before either control listener is available.
    let lock_path = source.home().join("bitrouter.start.lock");
    let startup_lock = std::fs::OpenOptions::new()
        .create(true)
        .write(true)
        .truncate(false)
        .open(&lock_path)
        .with_context(|| format!("opening startup lock {}", lock_path.display()))?;
    let _startup_lock = tokio::task::spawn_blocking(move || -> std::io::Result<std::fs::File> {
        startup_lock.lock()?;
        Ok(startup_lock)
    })
    .await??;
    if let Some(socket) = socket
        && let Some(info) = probe_status(socket).await?
    {
        return Ok(DaemonStartOutcome::Ready(info));
    }
    let (mut child, log_size_before) = spawn_detached_serve(source, log_path)?;

    let deadline = std::time::Instant::now() + timeout;
    loop {
        // Early-death: a bad config / port-in-use kills the child fast; surface
        // it with the log tail rather than waiting out the whole timeout. Any
        // exit before readiness means the daemon isn't serving, so the status
        // code is reported verbatim and treated as a startup failure by callers
        // (`serve` never exits 0 before it binds the control socket).
        if let Ok(Some(status)) = child.try_wait() {
            let log_tail = read_log_tail(log_path, log_size_before).await;
            return Ok(DaemonStartOutcome::Exited {
                status: status.to_string(),
                log_tail,
            });
        }
        if let Some(socket) = socket {
            match probe_status(socket).await {
                Ok(Some(info)) => return Ok(DaemonStartOutcome::Ready(info)),
                Ok(None) => {}
                Err(e) => tracing::debug!(error = %e, "readiness probe failed; retrying"),
            }
        }
        if std::time::Instant::now() >= deadline {
            return Ok(DaemonStartOutcome::NotReadyInTime { pid: child.id() });
        }
        tokio::time::sleep(READY_POLL_INTERVAL).await;
    }
}

#[cfg(test)]
mod tests {
    #[cfg(unix)]
    #[test]
    fn process_liveness_checks_one_representable_positive_pid() {
        assert!(super::process_is_alive(std::process::id()));
        assert!(!super::process_is_alive(0));
        assert!(!super::process_is_alive(u32::MAX));
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn bound_control_socket_preserves_other_owners_and_cleans_its_own_path()
    -> anyhow::Result<()> {
        let home = tempfile::tempdir_in("/tmp")?;
        let socket = home.path().join("control.sock");
        let first = super::bind_control_socket(&socket).await?;
        assert!(super::bind_control_socket(&socket).await.is_err());
        assert!(socket.exists());
        drop(first);
        assert!(!socket.exists());

        // A foreign listener that does not use our lock must also survive.
        let foreign = tokio::net::UnixListener::bind(&socket)?;
        assert!(super::bind_control_socket(&socket).await.is_err());
        assert!(socket.exists());
        drop(foreign);
        // The now-stale socket can be reclaimed by the next host startup.
        let recovered = super::bind_control_socket(&socket).await?;
        drop(recovered);
        assert!(!socket.exists());
        Ok(())
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn bound_control_socket_does_not_remove_replacements_or_regular_files()
    -> anyhow::Result<()> {
        let home = tempfile::tempdir_in("/tmp")?;
        let socket = home.path().join("control.sock");
        let bound = super::bind_control_socket(&socket).await?;
        std::fs::remove_file(&socket)?;
        let replacement = tokio::net::UnixListener::bind(&socket)?;
        drop(bound);
        assert!(
            socket.exists(),
            "cleanup must not unlink another listener's inode"
        );
        drop(replacement);
        std::fs::remove_file(&socket)?;
        std::fs::write(&socket, "keep this file")?;
        assert!(super::bind_control_socket(&socket).await.is_err());
        assert_eq!(std::fs::read_to_string(&socket)?, "keep this file");
        Ok(())
    }

    use super::*;

    #[test]
    fn commands_round_trip_as_json() {
        for cmd in [
            DaemonCommand::Stop,
            DaemonCommand::Reload { env: Vec::new() },
            DaemonCommand::ReloadState,
            DaemonCommand::Status,
            DaemonCommand::Route {
                model: "gpt-5".to_string(),
            },
            DaemonCommand::ObserveStatus,
            DaemonCommand::Inspect {
                inspection: DaemonInspection::Policy {
                    input: PolicyInput::default(),
                },
            },
        ] {
            let json = serde_json::to_string(&cmd).unwrap();
            let back: DaemonCommand = serde_json::from_str(&json).unwrap();
            // tag-based round trip
            assert_eq!(std::mem::discriminant(&cmd), std::mem::discriminant(&back));
        }
    }

    #[test]
    fn inspection_response_round_trips_as_json() -> anyhow::Result<()> {
        let response = DaemonResponse::Inspection {
            report: DaemonInspectionReport::Providers(ProvidersReport {
                resolved_via: "live".to_string(),
                providers: Vec::new(),
            }),
        };
        let json = serde_json::to_string(&response)?;
        let decoded: DaemonResponse = serde_json::from_str(&json)?;
        match decoded {
            DaemonResponse::Inspection {
                report: DaemonInspectionReport::Providers(report),
            } => assert_eq!(report.resolved_via, "live"),
            other => anyhow::bail!("expected provider inspection, got {other:?}"),
        }
        Ok(())
    }

    #[test]
    fn observe_status_payload_round_trips() {
        let payload = ObserveStatusPayload {
            compiled_in: true,
            exporter_wired: true,
            endpoint: Some("http://collector:4318".to_string()),
            header_count: 2,
            service_name: Some("bitrouter".to_string()),
            resource_attribute_count: 1,
            sampler: Some("parentbased_always_on".to_string()),
            sampler_arg: None,
            metrics_enabled: true,
            api_key_count: 42,
            api_key_cap: 1024,
            user_id_count: 7,
            user_id_cap: 256,
            active_spans: 3,
        };
        let json = serde_json::to_string(&DaemonResponse::ObserveStatus {
            payload: payload.clone(),
        })
        .unwrap();
        let back: DaemonResponse = serde_json::from_str(&json).unwrap();
        match back {
            DaemonResponse::ObserveStatus { payload: p } => {
                assert_eq!(p.endpoint, payload.endpoint);
                assert_eq!(p.api_key_count, 42);
                assert_eq!(p.active_spans, 3);
            }
            other => panic!("expected ObserveStatus, got {other:?}"),
        }
    }

    #[test]
    fn noop_observe_status_reports_unwired() {
        let p = NoopObserveStatus { compiled_in: true }.status();
        assert!(p.compiled_in);
        assert!(!p.exporter_wired);
        assert!(p.endpoint.is_none());
        assert_eq!(p.api_key_count, 0);
    }

    #[test]
    fn legacy_unit_reload_command_still_deserialises() {
        // Pre-env wire format: `{"cmd":"reload"}` with no env field.
        // The `#[serde(default)]` on `env` keeps this working so an
        // older client (or a script speaking the v1 wire format) can
        // still issue a no-env reload.
        let back: DaemonCommand = serde_json::from_str(r#"{"cmd":"reload"}"#).unwrap();
        match back {
            DaemonCommand::Reload { env } => assert!(env.is_empty()),
            other => panic!("expected Reload, got {other:?}"),
        }
    }

    #[test]
    fn reload_command_carries_env_overrides() {
        let cmd = DaemonCommand::Reload {
            env: vec![("OPENAI_API_KEY".to_string(), "sk-test".to_string())],
        };
        let json = serde_json::to_string(&cmd).unwrap();
        let back: DaemonCommand = serde_json::from_str(&json).unwrap();
        match back {
            DaemonCommand::Reload { env } => {
                assert_eq!(env.len(), 1);
                assert_eq!(
                    env[0],
                    ("OPENAI_API_KEY".to_string(), "sk-test".to_string())
                );
            }
            other => panic!("expected Reload, got {other:?}"),
        }
    }

    #[test]
    fn responses_round_trip_as_json() {
        let resp = DaemonResponse::Status {
            pid: 42,
            daemon_version: Some(crate::VERSION.to_string()),
            handoff_protocol: Some(1),
            handoff_build_id: Some(crate::HANDOFF_BUILD_ID.to_string()),
            handoff_activity: None,
            cli_owned: Some(true),
            listen: "0.0.0.0:4356".to_string(),
            models: 3,
            providers: vec!["openai".to_string()],
            saved_routers: None,
            running_routers: None,
            router_restart_required: None,
            config_state: None,
        };
        let json = serde_json::to_string(&resp).unwrap();
        let back: DaemonResponse = serde_json::from_str(&json).unwrap();
        match back {
            DaemonResponse::Status {
                pid,
                models,
                providers,
                ..
            } => {
                assert_eq!(pid, 42);
                assert_eq!(models, 3);
                assert_eq!(providers, ["openai"]);
            }
            other => panic!("expected Status, got {other:?}"),
        }
    }

    #[test]
    fn legacy_status_response_keeps_configuration_evidence_unknown() -> anyhow::Result<()> {
        let response: DaemonResponse = serde_json::from_str(
            r#"{
                "resp":"status",
                "pid":42,
                "listen":"127.0.0.1:4356",
                "models":3,
                "providers":["openai"]
            }"#,
        )?;
        let DaemonResponse::Status {
            saved_routers,
            running_routers,
            router_restart_required,
            config_state,
            ..
        } = response
        else {
            anyhow::bail!("legacy status payload decoded as the wrong response variant");
        };
        assert!(saved_routers.is_none());
        assert!(running_routers.is_none());
        assert!(router_restart_required.is_none());
        assert!(config_state.is_none());
        Ok(())
    }

    #[test]
    fn router_restart_detection_uses_secret_defaults_without_exposing_them() -> anyhow::Result<()> {
        let saved = bitrouter_sdk::config::parse(
            "routers:\n  coding:\n    selection:\n      kind: model\n      model: demo/model\n    defaults:\n      system_prompt: saved secret\n      params:\n        temperature: 0.2\n",
        )?;
        let running = bitrouter_sdk::config::parse(
            "routers:\n  coding:\n    selection:\n      kind: model\n      model: demo/model\n    defaults:\n      system_prompt: prior secret\n      params:\n        temperature: 0.8\n",
        )?;

        let saved_inventory = saved.router_inventory()?;
        let saved_digest = saved_inventory
            .first()
            .ok_or_else(|| anyhow::anyhow!("saved router inventory is empty"))?
            .binding_digest
            .clone();
        let running_inventory = running.router_inventory()?;
        let running_digest = running_inventory
            .first()
            .ok_or_else(|| anyhow::anyhow!("running router inventory is empty"))?
            .binding_digest
            .clone();
        assert_eq!(
            saved_digest, running_digest,
            "redacted binding identity deliberately excludes default values"
        );
        assert!(router_configuration_restart_required(&saved, &running));
        Ok(())
    }

    #[test]
    fn resolve_socket_joins_relative_paths_onto_config_directory() {
        let resolved = resolve_socket_path(
            Path::new("/Users/x/.bitrouter/bitrouter.yaml"),
            "./bitrouter.sock",
        );
        // Leading `./` is stripped so the rendered path doesn't read
        // `…/.bitrouter/./bitrouter.sock`.
        assert_eq!(
            resolved,
            PathBuf::from("/Users/x/.bitrouter/bitrouter.sock")
        );
    }

    #[test]
    fn resolve_socket_handles_relative_without_dot_prefix() {
        let resolved = resolve_socket_path(
            Path::new("/Users/x/.bitrouter/bitrouter.yaml"),
            "bitrouter.sock",
        );
        assert_eq!(
            resolved,
            PathBuf::from("/Users/x/.bitrouter/bitrouter.sock")
        );
    }

    #[test]
    fn resolve_socket_joins_nested_relative_path() {
        let resolved = resolve_socket_path(
            Path::new("/Users/x/.bitrouter/bitrouter.yaml"),
            "sub/bitrouter.sock",
        );
        assert_eq!(
            resolved,
            PathBuf::from("/Users/x/.bitrouter/sub/bitrouter.sock")
        );
    }

    #[test]
    fn resolve_socket_passes_absolute_paths_through() {
        let resolved = resolve_socket_path(
            Path::new("/Users/x/.bitrouter/bitrouter.yaml"),
            "/var/run/bitrouter.sock",
        );
        assert_eq!(resolved, PathBuf::from("/var/run/bitrouter.sock"));
    }

    #[test]
    fn resolve_socket_handles_config_paths_without_parent() {
        // `bitrouter.yaml` with no directory component → fall back to the
        // raw socket value (CWD-relative). The leading `./` is stripped.
        let resolved = resolve_socket_path(Path::new("bitrouter.yaml"), "./bitrouter.sock");
        assert_eq!(resolved, PathBuf::from("bitrouter.sock"));
    }

    #[test]
    fn is_not_reachable_detects_socket_not_found() {
        let io_err = std::io::Error::from(std::io::ErrorKind::NotFound);
        let err: anyhow::Error =
            anyhow::Error::new(io_err).context("connecting to /tmp/bitrouter.sock");
        assert!(is_not_reachable(&err));
    }

    #[test]
    fn is_not_reachable_detects_connection_refused() {
        let io_err = std::io::Error::from(std::io::ErrorKind::ConnectionRefused);
        let err: anyhow::Error =
            anyhow::Error::new(io_err).context("connecting to /tmp/bitrouter.sock");
        assert!(is_not_reachable(&err));
    }

    #[test]
    fn is_not_reachable_ignores_unrelated_io_errors() {
        let io_err = std::io::Error::from(std::io::ErrorKind::PermissionDenied);
        let err: anyhow::Error =
            anyhow::Error::new(io_err).context("connecting to /tmp/bitrouter.sock");
        assert!(!is_not_reachable(&err));
    }

    #[test]
    fn is_not_reachable_false_for_plain_messages() {
        let err = anyhow::anyhow!("daemon closed the connection without responding");
        assert!(!is_not_reachable(&err));
    }
}