openlatch-client 0.3.3

/// Daemon HTTP server — composes all leaf modules into the running service.
///
/// This module provides:
/// - [`AppState`]: shared state cloned into every axum handler via `Arc`
/// - [`start_server`]: entry point that builds the router, binds TCP, and runs to completion
/// - Signal handling: graceful shutdown on SIGTERM (Unix) or Ctrl+C (all platforms)
/// - Route layout: authenticated POST routes + unauthenticated GET routes
pub mod admin;
pub mod auth;
pub mod config_monitor;
pub mod dedup;
pub mod fallback_replay;
pub mod handlers;
pub mod identity;
pub mod policy_poller;
pub mod reconciler;
pub mod watcher;

use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use axum::{extract::DefaultBodyLimit, middleware, routing::get, routing::post, Router};
use secrecy::SecretString;
use tokio::net::TcpListener;

use crate::cloud::{CloudState, CredentialProvider};
use crate::config::Config;
use crate::core::logging::tamper_log::{TamperLogger, TamperLoggerHandle};
use crate::core::supervision::task::{spawn_supervised, HealthRegistry, RestartPolicy, TaskSpec};
use crate::logging::{EventLogger, EventLoggerHandle};
use crate::privacy::PrivacyFilter;
use crate::update;

/// Names of the supervised subsystems, as they appear in `daemon.log`,
/// `GET /health` and `openlatch status`. Centralised so the strings the
/// operator greps for cannot drift from the strings the code registers.
mod subsystem {
    pub const BOUNDARY: &str = "boundary";
    pub const BOUNDARY_WIRING: &str = "boundary-wiring";
    pub const CLOUD_WORKER: &str = "cloud-worker";
    pub const ALERTS_LONG_POLL: &str = "alerts-long-poll";
    pub const POLICY_POLLER: &str = "policy-poller";
    pub const RECONCILER: &str = "reconciler";
    pub const DEDUP_EVICTOR: &str = "dedup-evictor";
    pub const LOG_CLEANUP: &str = "log-cleanup";
    pub const EVENT_LOG_WRITER: &str = "event-log-writer";
    pub const TAMPER_LOG_WRITER: &str = "tamper-log-writer";
    pub const FALLBACK_REPLAY: &str = "fallback-replay";
    pub const UPDATE_CHECK: &str = "update-check";
    pub const AUTO_UPDATE_WORKER: &str = "auto-update-worker";
    pub const UPDATE_SENTINEL: &str = "update-sentinel";
    pub const EGRESS_MONITOR: &str = "egress-monitor";
}

// ---------------------------------------------------------------------------
// CredentialStore adapter for cloud worker CredentialProvider
// ---------------------------------------------------------------------------

/// Adapts a `crate::auth::CredentialStore` to the `CredentialProvider` trait
/// required by the cloud worker.
///
/// The adapter ignores `OlError` from `retrieve()` and converts it to `None`,
/// which causes the worker to skip POSTs (fail-open) until a valid key exists.
struct CredentialStoreAdapter {
    store: Arc<dyn crate::auth::CredentialStore>,
}

impl CredentialProvider for CredentialStoreAdapter {
    fn retrieve(&self) -> Option<SecretString> {
        match self.store.retrieve() {
            Ok(key) => Some(key),
            Err(e) => {
                tracing::debug!(
                    code = e.code,
                    "credential provider: retrieve failed — returning None to worker"
                );
                None
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Policy plane
// ---------------------------------------------------------------------------

/// The daemon's policy plane: the resident rule set plus the poll state that
/// changes *without* the bundle changing.
///
/// Written only by [`policy_poller`]; read by the verdict path in
/// [`handlers::ingest_cloudevent`] and by `/metrics`. The rule set sits behind
/// an [`arc_swap::ArcSwap`] so a read is lock-free and wait-free and an
/// in-flight evaluation always sees one consistent bundle — the poller
/// publishes a whole new bundle rather than mutating the resident one.
///
/// `AppState::policy` is `None` when `[policy] enabled = false`, which is the
/// complete off switch: nothing is fetched, nothing is evaluated, no `olpolicy*`
/// attribute is stamped, and the daemon answers exactly as it did before policy
/// existed. Any bundle on disk is left untouched so re-enabling does not need a
/// re-download.
pub struct PolicyRuntime {
    /// The atomic handle the verdict path reads. `None` inside means no bundle
    /// has ever been activated — the fail-open case.
    pub handle: crate::core::policy::PolicyHandle,
    /// `false` after a failed poll attempt → `olpolicyoffline`.
    pub last_fetch_ok: Arc<AtomicBool>,
    /// Unix seconds of the last successful poll (any `2xx` **or** `304`); `0`
    /// means never. `/metrics` reads this rather than re-reading
    /// `bundle.meta.json` from disk on every scrape.
    pub last_poll_ok_at: Arc<AtomicI64>,
}

impl PolicyRuntime {
    /// Load and digest-verify the cached bundle **synchronously**.
    ///
    /// This runs during daemon startup, before `axum::serve` accepts its first
    /// request. A restarted daemon must be enforcing on its first served hook:
    /// starting empty and waiting for the poller's boot fetch leaves an
    /// unprotected window on EVERY restart, and with the network down at boot
    /// the host would run with no policy at all despite a perfectly good bundle
    /// sitting on disk. Closing that gap is the whole point of the
    /// local-authoritative design, so this must never be moved into the spawned
    /// poller task.
    ///
    /// It is a few KB of disk read plus one SHA-256. A rejected or absent cache
    /// is not fatal — the daemon comes up with no policy (allow everything,
    /// marked as having no bundle) and the poller's boot fetch repairs it.
    pub fn load_from_disk(base_dir: &Path) -> Self {
        use crate::core::policy::{store, ResidentBundle};

        let cached = match store::load(base_dir) {
            Ok(cached) => cached,
            Err(e) => {
                // Covers the tampered-bundle case: the digest is re-verified on
                // every load, not only after a fetch, so a local edit that
                // deletes the rule blocking someone is rejected here.
                tracing::warn!(
                    target: "policy",
                    code = e.code(),
                    error = %e,
                    "cached policy bundle rejected at startup; running with no policy until the next successful fetch"
                );
                None
            }
        };

        let last_fetch_ok = Arc::new(AtomicBool::new(true));
        let last_poll_ok_at = Arc::new(AtomicI64::new(0));
        let resident = cached.map(|c| {
            // Seed the poll state from the meta file so a restart does not
            // reset the staleness clock and so a daemon running WITHOUT a
            // poller (no credential provider) still reports the truth. The
            // poller re-seeds these identically on its own startup.
            last_fetch_ok.store(c.meta.last_fetch_ok, Ordering::Relaxed);
            if let Some(secs) = c
                .meta
                .last_poll_ok_at
                .as_deref()
                .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
                .map(|dt| dt.timestamp())
            {
                last_poll_ok_at.store(secs, Ordering::Relaxed);
            }
            ResidentBundle::from_bundle(&c.bundle)
        });

        Self {
            handle: crate::core::policy::new_handle(resident),
            last_fetch_ok,
            last_poll_ok_at,
        }
    }

    /// One INFO line describing what the daemon is enforcing at startup.
    ///
    /// Without it, `enabled = false` and a silently-broken poller look
    /// identical during the canary.
    fn log_startup_state(&self) {
        match self.handle.load().as_ref() {
            Some(b) => tracing::info!(
                target: "policy",
                revision = b.revision,
                rules = b.command_rules.len(),
                request_rules = b.request_rules.len(),
                enforcement_enabled = b.enforcement_enabled,
                organization_id = %b.organization_id,
                "policy engine enabled; enforcing the cached bundle"
            ),
            None => tracing::info!(
                target: "policy",
                "policy engine enabled; no bundle — allowing everything until the first successful fetch"
            ),
        }
    }
}

/// Shared state injected into every axum handler via `Arc<AppState>`.
///
/// Fields are either inherently thread-safe (`AtomicU64`, `DashMap`, `mpsc::Sender`)
/// or wrapped in appropriate synchronization primitives.
pub struct AppState {
    /// Resolved daemon configuration (port, log dir, retention, etc.)
    pub config: Arc<Config>,
    /// Bearer token for authenticating POST requests.
    /// SECURITY: Never log this value.
    pub token: String,
    /// In-memory dedup store with 100ms TTL.
    pub dedup: dedup::DedupStore,
    /// Async event logger (sends to background writer task via mpsc).
    pub event_logger: EventLogger,
    /// Pre-compiled privacy filter for credential masking.
    pub privacy_filter: PrivacyFilter,
    /// Total events processed (not counting deduped duplicates).
    pub event_counter: AtomicU64,
    /// Oneshot sender for triggering graceful shutdown via POST /shutdown.
    /// Wrapped in Mutex so the handler can take ownership without `&mut self`.
    pub shutdown_tx: tokio::sync::Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
    /// Wall-clock time when the daemon started (for uptime reporting).
    pub started_at: std::time::Instant,
    /// Latest available version string, populated by the async update check on startup.
    /// `None` means either the check has not completed yet, or the current version is latest.
    pub available_update: Mutex<Option<String>>,
    /// Cloud forwarding channel sender. `None` if cloud forwarding is disabled or unconfigured.
    ///
    /// Handlers call `try_send(CloudEvent)` — non-blocking, off the verdict critical path.
    /// CLOUD-01/CLOUD-09: fire-and-forget pattern.
    pub cloud_tx: Option<tokio::sync::mpsc::Sender<crate::cloud::CloudEvent>>,
    /// Shared cloud state: auth_error flag visible to the status command (CLOUD-08).
    /// `None` if cloud forwarding is disabled.
    pub cloud_state: Option<CloudState>,
    /// Machine's local (LAN) IPv4 address, resolved once at startup.
    pub local_ipv4: Option<std::net::Ipv4Addr>,
    /// Machine's local (LAN) IPv6 address, resolved once at startup.
    pub local_ipv6: Option<std::net::Ipv6Addr>,
    /// Machine's public (internet-facing) IPv4 address, resolved once at startup.
    pub public_ipv4: Option<std::net::Ipv4Addr>,
    /// Machine's public (internet-facing) IPv6 address, resolved once at startup.
    pub public_ipv6: Option<std::net::Ipv6Addr>,
    /// Async writer for `~/.openlatch/tamper.jsonl`. The reconciler sends
    /// detected/healed `TamperEvent`s through this on drift. `None` only in
    /// test builds that construct `AppState` directly without the daemon
    /// startup path.
    pub tamper_logger: Option<TamperLogger>,
    /// Shared reference to the durable outbox. `None` when cloud forwarding
    /// is disabled or `cloud.outbox_enabled = false`. Held on AppState so
    /// the `/metrics` handler can surface pending byte/entry counts, and so
    /// the future fallback-replay routine can append-after-parse without
    /// reopening the file.
    pub outbox: Option<Arc<crate::cloud::outbox::Outbox>>,
    /// Single-update lock. `POST /admin/update` swaps this from `false`
    /// to `true` atomically — a second concurrent caller gets 503.
    /// Successful applies do not reset it: the daemon is about to
    /// restart, and the new daemon comes up with the lock fresh-`false`
    /// because the field is `AtomicBool::new(false)` again.
    pub update_in_progress: Arc<AtomicBool>,
    /// Long-poll status surface for `GET /admin/update/status`. The
    /// apply task in `daemon::admin` updates this as it advances
    /// through stages so the CLI can render progress.
    pub update_status: Arc<Mutex<update::UpdateStatusSnapshot>>,
    /// Cooperative shutdown request signal — apply pipeline notifies
    /// this when the swap is committed and axum should drain. The main
    /// `axum::serve(...).with_graceful_shutdown(...)` future races this
    /// against the existing oneshot + OS signal handlers.
    pub admin_shutdown_request: Arc<tokio::sync::Notify>,
    /// Unix-seconds timestamp of the last live hook ingest. Updated on
    /// the hot path with `Ordering::Relaxed`. The auto-update worker
    /// reads it to decide whether the agent is currently active.
    pub last_hook_at_unix_secs: Arc<AtomicU64>,
    /// Count of live hook handlers currently in flight. Incremented at
    /// the top of the ingest handler and decremented from the
    /// `HookActivityGuard`'s `Drop` impl, so a handler that early-returns
    /// or panics still leaves the counter consistent. The auto-update
    /// worker refuses to apply non-critical updates while this is
    /// non-zero, blocking long-running hooks that would otherwise look
    /// idle to entry-only timestamping.
    pub hooks_in_flight: Arc<AtomicU32>,
    /// In-memory cache of `(path → content_hash)` observations driving
    /// FS-watcher dedup against native Claude Code hooks. Populated by
    /// the config monitor; surfaced via `/admin/inventory/status`.
    pub content_hash_cache: Arc<config_monitor::ContentHashCache>,
    /// Sender for re-driving the config monitor (manual rescan, native
    /// hook routing, project-scope register). `None` when the monitor is
    /// disabled by config or failed to start.
    pub config_monitor_request_tx:
        Option<tokio::sync::mpsc::Sender<config_monitor::ConfigChangeRequest>>,
    /// Pending alerts queued by the cloud's deep-analysis worker, fetched
    /// via the `/api/v1/alerts/pending` long-poll. Surfaced to the user
    /// at the next outbound hook response (translator injects them as
    /// `permissionDecisionReason` for PreToolUse / `additionalContext`
    /// for SessionStart). Always present (empty when no alerts have
    /// been received yet).
    pub pending_alerts: Arc<config_monitor::PendingAlerts>,
    /// Local policy evaluation state (D47). `None` when
    /// `[policy] enabled = false` — the complete off switch. See
    /// [`PolicyRuntime`].
    pub policy: Option<PolicyRuntime>,
    /// Shared active-session registry (model-boundary, D-09). The hook side
    /// (`handlers::process_envelope`) stamps it on `SessionStart` / tool-call
    /// hooks; the boundary listener (spawned in this same function when
    /// `spawn_boundary`) reads the **same** `Arc` to resolve attribution +
    /// assurance in-process (B-2). Always present; empty until the first hook.
    pub registry: Arc<crate::boundary::session::SessionRegistry>,
    /// Live state of every supervised in-process subsystem. Read by `/health`
    /// (which reports `degraded` when a `RestartPolicy::Always` task is not
    /// running), `/metrics`, and `openlatch status`. Written only by the
    /// supervisors in `core::supervision::task`.
    pub health: Arc<HealthRegistry>,
    /// Runtime egress health: the resolved route and whether it currently
    /// works. Written by every factory consumer at its own send site, read by
    /// `/admin/egress/status`, `/metrics` and `/health`.
    ///
    /// **Always present**, including on a host that has never authenticated:
    /// egress is not a cloud-credential capability, and a fresh install whose
    /// proxy is wrong needs the answer more than anyone.
    pub egress: crate::egress::EgressState,
}

impl AppState {
    /// Store a newly discovered available version.
    pub fn set_available_update(&self, version: String) {
        if let Ok(mut guard) = self.available_update.lock() {
            *guard = Some(version);
        }
    }

    /// Return the latest available version, if one has been discovered.
    pub fn get_available_update(&self) -> Option<String> {
        self.available_update.lock().ok().and_then(|g| g.clone())
    }
}

/// Start the daemon HTTP server and run until a shutdown signal is received.
///
/// Binds to `127.0.0.1:{config.port}`. The server shuts down gracefully on:
/// - SIGTERM (Unix) or Ctrl+C (all platforms)
/// - HTTP POST /shutdown (authenticated)
///
/// After shutdown, prints a summary to stderr and waits for the event logger to drain.
///
/// # Parameters
///
/// - `credential_store`: optional credential store for cloud forwarding.
///   When `Some`, the cloud worker is spawned. When `None`, there is nothing to
///   authenticate with, so forwarding stays idle — the only way it does not run.
///
/// # Errors
///
/// Returns an error if the TCP listener cannot be bound (e.g., port in use).
pub async fn start_server(
    config: Config,
    token: String,
    credential_store: Option<Arc<dyn crate::auth::CredentialStore>>,
    spawn_boundary: bool,
) -> anyhow::Result<(u64, u64)> {
    // SECURITY: Bind to 127.0.0.1 by default.
    // Only bind 0.0.0.0 inside Docker/container environments where the network
    // boundary provides isolation instead of the loopback interface.
    // OPENLATCH_BIND_ALL must be set to "true" or "1" (not merely present) to
    // avoid `OPENLATCH_BIND_ALL=false` silently enabling wide binding.
    let bind_host = match std::env::var("OPENLATCH_BIND_ALL").as_deref() {
        Ok("true") | Ok("1") => "0.0.0.0",
        _ => "127.0.0.1",
    };
    let bind_addr = format!("{}:{}", bind_host, config.port);
    let listener = TcpListener::bind(&bind_addr).await?;

    tracing::info!(
        port = config.port,
        addr = %bind_addr,
        "daemon listening"
    );

    // Write daemon.port file so the hook binary can discover the port
    if let Err(e) = crate::config::write_port_file(config.port) {
        tracing::warn!(error = %e, "failed to write daemon.port file");
    }

    serve_with_listener(
        listener,
        config,
        token,
        credential_store,
        spawn_boundary,
        /* reconcile_wiring = */ true,
    )
    .await
}

/// Start the daemon with a pre-bound TCP listener.
///
/// Accepts an already-bound listener — useful for integration tests where port 0
/// is bound by the OS for a random free port, avoiding test conflicts.
///
/// # Errors
///
/// Returns an error if the axum server fails during operation.
pub async fn start_server_with_listener(
    listener: TcpListener,
    config: Config,
    token: String,
    credential_store: Option<Arc<dyn crate::auth::CredentialStore>>,
    spawn_boundary: bool,
) -> anyhow::Result<(u64, u64)> {
    // `reconcile_wiring = false`: this entry point serves an embedded or test
    // daemon that shares the machine with a real one. `spawn_boundary = false`
    // here means "not my job", not "nothing holds the port" — reconciling from
    // it would strip the wiring out from under a live daemon (and, in a test,
    // out of the developer's own ~/.claude/settings.json).
    serve_with_listener(
        listener,
        config,
        token,
        credential_store,
        spawn_boundary,
        /* reconcile_wiring = */ false,
    )
    .await
}

/// Internal implementation: serve HTTP on the given listener.
///
/// `spawn_boundary` decides whether the in-process model-boundary listener is
/// co-launched here. It is `true` only on the full daemon path (`openlatch start`
/// / supervised / `init --foreground` with the boundary on) and `false` for tests
/// and for an explicit opt-out — the intentional asymmetry that keeps a
/// non-daemon process from binding the pinned boundary port.
///
/// `reconcile_wiring` says whether a `spawn_boundary = false` start is
/// authoritative about the pinned port being unheld. Only a real daemon process
/// is (see [`start_server_with_listener`] for the case that is not).
async fn serve_with_listener(
    listener: TcpListener,
    config: Config,
    token: String,
    credential_store: Option<Arc<dyn crate::auth::CredentialStore>>,
    spawn_boundary: bool,
    reconcile_wiring: bool,
) -> anyhow::Result<(u64, u64)> {
    // Every long-lived task in this process runs under the in-process
    // supervisor: a panic restarts the subsystem instead of silently killing it
    // while `/health` keeps answering `ok`. `health` is the observable side of
    // that (read by `/health`, `/metrics`, `openlatch status`);
    // `tasks_shutdown_tx` is the single teardown broadcast every supervisor and
    // every supervised loop honours, so `openlatch stop` drains them all at once
    // rather than one bespoke channel at a time.
    let health = Arc::new(HealthRegistry::new());
    let (tasks_shutdown_tx, tasks_shutdown_rx) = tokio::sync::watch::channel(false);

    // The two log writers get a SEPARATE signal that is never sent — this
    // sender is simply held for the whole function. Their real terminator is
    // their channel closing, which happens only after `Arc::try_unwrap(state)`
    // drops the last sender at the very end of shutdown. Handing them
    // `tasks_shutdown_tx` instead would abort them mid-drain and throw away the
    // final batch of audit lines, which is precisely what the drain below
    // exists to preserve.
    let (_logs_shutdown_tx, logs_shutdown_rx) = tokio::sync::watch::channel(false);

    let log_dir = config.log_dir.clone();
    // The receiver lives in the `Arc`, not in the future, so a panicked writer
    // restarts onto the SAME channel with its queued audit lines intact.
    let (event_logger, event_log_rx) = EventLogger::channel();
    let event_log_rx = Arc::new(tokio::sync::Mutex::new(event_log_rx));
    let logger_handle = {
        let log_dir = log_dir.clone();
        EventLoggerHandle::from_task(spawn_supervised(
            &health,
            // OnFailure, not Always: the writer's normal exit is its channel
            // closing at shutdown, and restarting on that would spin. A panic
            // still gets it back.
            TaskSpec::new(subsystem::EVENT_LOG_WRITER, RestartPolicy::OnFailure),
            logs_shutdown_rx.clone(),
            move || {
                let rx = event_log_rx.clone();
                let log_dir = log_dir.clone();
                async move {
                    let mut rx = rx.lock_owned().await;
                    crate::logging::run_event_writer(log_dir, &mut rx).await
                }
            },
        ))
    };

    let privacy_filter = PrivacyFilter::new(&config.extra_patterns);

    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();

    // Pending-alerts ring buffer is daemon-scoped — populated by the
    // long-poll task spawned inside the cloud-forwarding block below,
    // drained by the hooks handler at outbound translation time.
    let pending_alerts = Arc::new(config_monitor::PendingAlerts::new());

    // Policy plane (D47). The cached bundle is loaded and digest-verified HERE,
    // synchronously, before `axum::serve` starts accepting requests further
    // down — not in the spawned poller, which would reintroduce the
    // unprotected-restart window this exists to close. The poller (spawned
    // inside the cloud block below) then does its own immediate boot fetch.
    let policy_runtime = if config.policy.enabled {
        let runtime = PolicyRuntime::load_from_disk(&crate::config::openlatch_dir());
        runtime.log_startup_state();
        Some(runtime)
    } else {
        tracing::info!(
            target: "policy",
            "policy engine disabled by config ([policy] enabled = false); no bundle is fetched and no resident bundle is consulted"
        );
        None
    };

    // Runtime egress health, and the task that watches it. Built and spawned
    // OUTSIDE the credential-gated block below, on purpose: a freshly installed
    // host that has not run `openlatch auth login` yet has no cloud worker at
    // all, and it is exactly the host most likely to be sitting behind a proxy
    // nobody has configured. It still gets live `/admin/egress/status`, live
    // `/metrics` keys and a supervised observer.
    let egress_state = crate::egress::EgressState::new(&config.egress);
    for warning in egress_state.warnings() {
        tracing::warn!(target: "egress", "{warning}");
    }

    // The handle set every long-lived consumer reads and the self-heal pass writes.
    //
    // This exists because the daemon builds each outbound client exactly ONCE, right here,
    // and hands it to a task that then owns it for the process lifetime. Persisting a
    // newly discovered proxy without replacing those clients would leave every live pool
    // talking to the dead one until the next restart — which is what D-18 means by
    // *applied, not just persisted*. See [`crate::egress::EgressClients`].
    let cloud_timeouts = crate::egress::Timeouts {
        connect: Some(std::time::Duration::from_millis(
            config.cloud.timeout_connect_ms,
        )),
        total: Some(std::time::Duration::from_millis(
            config.cloud.timeout_total_ms,
        )),
    };
    let poll_timeouts = crate::egress::Timeouts::total(std::time::Duration::from_millis(
        config.cloud.timeout_total_ms,
    ));
    let egress_clients = Arc::new(crate::egress::EgressClients::new(
        cloud_timeouts,
        poll_timeouts,
    ));
    if let Err(e) = egress_clients.apply(&config.egress) {
        // Every consumer's client comes from one configuration, so a failure is a property
        // of the route rather than of a consumer — one line, once. The handles stay empty,
        // which is exactly the posture `[proxy] allow_direct = false` asks for: there is no
        // client for anything on this host to quietly go direct with (D-21). Recording it
        // twice reaches `FAILURE_THRESHOLD`, so `/health`, `/metrics` and `doctor` report a
        // failed egress instead of a healthy daemon whose every request dies.
        tracing::error!(
            target: "egress",
            code = %e.code,
            error = %e.message,
            "no egress client could be built; outbound traffic is refused until the route is fixed"
        );
        egress_state.record_failure(e.code, e.message.clone());
        egress_state.record_failure(e.code, e.message);
    }

    // Daemon-start `proxy_configured` (G-07). Emitted only when the resolved shape differs
    // from the on-disk memo: the auto-update path restarts the daemon, and an in-memory
    // memo would re-emit fleet-wide on every release. Shape only, never the address (D-15).
    crate::egress::emit_proxy_shape_if_changed(
        &crate::config::openlatch_dir(),
        &egress_state.snapshot(),
        0,
    );

    {
        let monitor_state = egress_state.clone();
        let monitor_shutdown = tasks_shutdown_rx.clone();
        let heal = crate::egress::SelfHeal {
            // **I-2 gated:** the discovery ladder is I-2's. Until it lands the self-heal
            // pass has nothing to try and says so; the one-line change when it does is
            // `resolver: Some(Arc::new(<I-2's service-context resolver>))`.
            resolver: None,
            clients: egress_clients.clone(),
            base: Arc::new(config.egress.clone()),
            api_url: config.cloud.api_url.clone(),
            openlatch_dir: crate::config::openlatch_dir(),
        };
        spawn_supervised(
            &health,
            // A forever-loop, so `Always`: it must be running for the whole
            // daemon lifetime, and `/health` says so when it is not.
            TaskSpec::new(subsystem::EGRESS_MONITOR, RestartPolicy::Always),
            tasks_shutdown_rx.clone(),
            // Fresh future per call — the factory re-clones its handles rather
            // than moving them, so a restart observes the same state.
            move || {
                crate::egress::run_egress_monitor(
                    monitor_state.clone(),
                    monitor_shutdown.clone(),
                    Some(heal.clone()),
                )
            },
        );
    }

    // Shared model-boundary session registry (D-09). Created ONCE here so the
    // hook side (AppState → process_envelope) and the boundary listener (spawned
    // below when `spawn_boundary`) hold the SAME Arc — a hook upsert is visible
    // to the very next boundary request. Empty until the first hook fires.
    let registry = Arc::new(crate::boundary::session::SessionRegistry::default());

    // Cloud forwarding setup (CLOUD-01, D-03): spawn the worker whenever a
    // credential store is available. There is no "cloud disabled" branch —
    // forwarding to the platform is not a preference (see `CloudConfig::enabled`).
    #[allow(clippy::type_complexity)]
    let (cloud_tx, cloud_state_opt, outbox_opt, cloud_worker_task): (
        _,
        _,
        _,
        Option<tokio::task::JoinHandle<()>>,
    ) = {
        // No `if config.cloud.enabled` gate any more: forwarding to the
        // platform is not optional (see `CloudConfig::enabled`). The worker
        // still needs a credential store — without one there is nothing to
        // authenticate with — but that is a capability question, not a
        // preference.
        if let Some(store) = credential_store {
            let (tx, rx) = tokio::sync::mpsc::channel(config.cloud.channel_size);
            let cloud_state = CloudState::new();

            // Build cloud worker config from resolved daemon config
            let cloud_config = crate::cloud::CloudConfig {
                api_url: config.cloud.api_url.clone(),
                timeout_connect_ms: config.cloud.timeout_connect_ms,
                timeout_total_ms: config.cloud.timeout_total_ms,
                retry_delay_ms: config.cloud.retry_delay_ms,
                channel_size: config.cloud.channel_size,
                rate_limit_default_secs: 30,
                credential_poll_interval_ms: config.cloud.credential_poll_interval_ms,
                fallback_max_bytes: config.cloud.fallback_max_bytes,
                batch_max_events: config.cloud.batch_max_events,
                batch_max_wait_ms: config.cloud.batch_max_wait_ms,
            };

            let openlatch_dir = crate::config::openlatch_dir();
            let provider: Arc<dyn CredentialProvider> = Arc::new(CredentialStoreAdapter { store });
            let worker_state = cloud_state.clone();
            let alerts_provider = provider.clone();
            let alerts_state = cloud_state.clone();
            // The policy poller needs a `CredentialProvider` and a `CloudState`,
            // and both only exist inside this block — `provider` is moved into
            // the cloud worker below and `openlatch_dir` with it. Clone the
            // three now, exactly as the alerts long-poll already does.
            let policy_provider = provider.clone();
            let policy_state = cloud_state.clone();
            let policy_dir = openlatch_dir.clone();
            let worker_egress =
                crate::egress::EgressReporter::recording(&config.egress, egress_state.clone());

            // Durable outbox: failed POSTs spool here; the worker's drain task
            // replays them after every successful health probe so events
            // captured while offline eventually reach the cloud. Disabled via
            // `cloud.outbox_enabled = false` or `cloud.outbox_max_bytes = 0`.
            let outbox = if config.cloud.outbox_enabled {
                Some(Arc::new(crate::cloud::outbox::Outbox::new(
                    &openlatch_dir,
                    config.cloud.outbox_max_bytes,
                )))
            } else {
                None
            };

            // The daemon-wide `tasks_shutdown_tx` doubles as the cloud worker's
            // explicit flush signal. Channel closure alone is not a reliable
            // trigger: `cloud_tx` lives on the `Arc<AppState>` and is cloned
            // into the config monitor and the tamper reconciler's sinks, so the
            // mpsc may not close promptly on `openlatch stop` — the
            // `Arc::try_unwrap` below already warns about exactly that case.
            //
            // Receiver behind an `Arc<Mutex<_>>` for the same reason as the
            // event-log writer: a panicked run must be restartable onto the
            // same channel, buffered events included.
            let cloud_rx = Arc::new(tokio::sync::Mutex::new(rx));
            let worker_outbox = outbox.clone();
            // `source` -> wire format, built ONCE here and handed to the worker.
            // The worker flushes batches, so a `detect_agents()` call per
            // envelope would stat the filesystem on every event; and
            // `core::cloud` must not call `crate::hooks::`, which would invert
            // the core -> hooks direction. A host that installs a new agent
            // mid-run picks it up on the next daemon restart.
            let source_formats: crate::cloud::worker::SourceFormats = crate::hooks::detect_agents()
                .iter()
                .filter_map(|a| {
                    a.binding
                        .boundary_wiring()
                        .map(|w| (a.binding.agent_type(), w.wire_format))
                })
                .collect();
            // The same `Arc` the hook path upserts into and the boundary reads
            // (created above). The worker resolves a session for the producers
            // that cannot name one themselves — the config monitor and the
            // tamper reconciler — so their events land in a session the
            // platform can display instead of in none at all.
            let worker_sessions = registry.clone();
            // The swap handle, not a client: the worker re-reads it per flush, per drain
            // pass and per health probe, so a self-heal pass reaches the highest-volume
            // egress consumer on the host without restarting it.
            let worker_client = egress_clients.cloud.clone();
            let cloud_worker = spawn_supervised(
                &health,
                TaskSpec::new(subsystem::CLOUD_WORKER, RestartPolicy::Always),
                tasks_shutdown_rx.clone(),
                {
                    let shutdown_rx = tasks_shutdown_rx.clone();
                    move || {
                        let rx = cloud_rx.clone();
                        let provider = provider.clone();
                        let cloud_config = cloud_config.clone();
                        let worker_egress = worker_egress.clone();
                        let worker_state = worker_state.clone();
                        let openlatch_dir = openlatch_dir.clone();
                        let outbox = worker_outbox.clone();
                        let shutdown_rx = shutdown_rx.clone();
                        let worker_client = worker_client.clone();
                        let source_formats = source_formats.clone();
                        let sessions = worker_sessions.clone();
                        async move {
                            let mut rx = rx.lock_owned().await;
                            crate::cloud::worker::run_cloud_worker_on(
                                &mut rx,
                                provider,
                                cloud_config,
                                worker_egress,
                                worker_state,
                                openlatch_dir,
                                outbox,
                                Some(shutdown_rx),
                                worker_client,
                                source_formats,
                                sessions,
                            )
                            .await
                        }
                    }
                },
            );

            tracing::info!(
                api_url = %config.cloud.api_url,
                channel_size = config.cloud.channel_size,
                "cloud forwarding worker started"
            );

            // Pending-alerts long-poll. Reuses the same credential
            // provider + cloud state as the worker so token rotation +
            // adaptive backoff (5–60 s) stay coordinated. Failures are
            // logged at debug! and the loop continues fail-open.
            let alerts_url = config.cloud.api_url.clone();
            let alerts_target = pending_alerts.clone();
            let alerts_egress =
                crate::egress::EgressReporter::recording(&config.egress, egress_state.clone());
            // The shared handle, re-read per long-poll iteration. `current()` here answers
            // the same question the per-consumer build used to: is there a usable route at
            // all? A `None` means the one `apply` above already reported why.
            let alerts_client = egress_clients.alerts.if_routed();
            // Platform's GET /api/v1/alerts/pending requires X-OpenLatch-Machine-Id
            // (D-2.07). Without an agent_id we can't satisfy the contract — skip
            // spawning rather than spamming 400/missing_machine_id every poll.
            match (alerts_client, config.agent_id.clone()) {
                (Some(client), Some(machine_id)) => {
                    spawn_supervised(
                        &health,
                        TaskSpec::new(subsystem::ALERTS_LONG_POLL, RestartPolicy::Always),
                        tasks_shutdown_rx.clone(),
                        move || {
                            config_monitor::run_long_poll(
                                alerts_target.clone(),
                                alerts_state.clone(),
                                alerts_provider.clone(),
                                alerts_url.clone(),
                                machine_id.clone(),
                                client.clone(),
                                alerts_egress.clone(),
                            )
                        },
                    );
                }
                (Some(_), None) => {
                    tracing::warn!(
                        "alerts long-poll skipped: agent_id missing from config.toml; run `openlatch init` to provision"
                    );
                }
                _ => {}
            }

            // Policy bundle poller. Shares the credential provider and cloud
            // state with the worker so token rotation and the auth-error pause
            // stay coordinated — the poller READS `is_auth_error` and never
            // writes it (D46). The resident bundle it publishes was already
            // loaded from disk above; the poller's job is only to keep it
            // fresh.
            if let Some(policy) = policy_runtime.as_ref() {
                // The shared handle, re-read per poll. `current()` answers the same
                // question the per-consumer build used to; a `None` means the one `apply`
                // above already reported why.
                match egress_clients.poller.if_routed() {
                    Some(client) => {
                        let policy_handle = policy.handle.clone();
                        let policy_last_fetch_ok = policy.last_fetch_ok.clone();
                        let policy_last_poll_ok_at = policy.last_poll_ok_at.clone();
                        let policy_api_url = config.cloud.api_url.clone();
                        let policy_cfg = config.policy.clone();
                        // Sent as `X-OpenLatch-Agent-Id` so the platform can
                        // compose this agent's `client_config.agent_context`;
                        // `None` before `openlatch init` provisions one, and
                        // the poller then omits the header.
                        let policy_agent_id = config.agent_id.clone();
                        let policy_egress = crate::egress::EgressReporter::recording(
                            &config.egress,
                            egress_state.clone(),
                        );
                        spawn_supervised(
                            &health,
                            TaskSpec::new(subsystem::POLICY_POLLER, RestartPolicy::Always),
                            tasks_shutdown_rx.clone(),
                            move || {
                                policy_poller::run_policy_poller(
                                    policy_handle.clone(),
                                    policy_last_fetch_ok.clone(),
                                    policy_last_poll_ok_at.clone(),
                                    policy_state.clone(),
                                    policy_provider.clone(),
                                    policy_api_url.clone(),
                                    policy_cfg.clone(),
                                    policy_dir.clone(),
                                    client.clone(),
                                    policy_agent_id.clone(),
                                    policy_egress.clone(),
                                )
                            },
                        );
                        tracing::info!(
                            target: "policy",
                            poll_interval_secs = config.policy.poll_interval_secs,
                            "policy bundle poller started"
                        );
                    }
                    None => {
                        // Disk-bundle-only: the resident bundle keeps
                        // enforcing, it just never refreshes.
                        tracing::warn!(
                            target: "policy",
                            code = crate::error::ERR_BUNDLE_FETCH_FAILED,
                            "no egress route is permitted, so the policy poller is not started; the resident bundle keeps enforcing but will not refresh"
                        );
                    }
                }
            }

            (Some(tx), Some(cloud_state), outbox, Some(cloud_worker))
        } else {
            tracing::info!(
                "no credential store provided — cloud forwarding cannot authenticate and is idle"
            );
            (None, None, None, None)
        }
    };

    // `policy.enabled = true` with cloud forwarding off (or no credential
    // store) is a reachable configuration, and `cloud_state_opt` is `Some`
    // exactly when the block above ran. Define the case rather than leaving it
    // to be invented: run DISK-BUNDLE-ONLY. The cached bundle still loads and
    // still enforces — enforcement is local-authoritative — it just never
    // refreshes. Never silently disable enforcement, never panic.
    if policy_runtime.is_some() && cloud_state_opt.is_none() {
        tracing::warn!(
            target: "policy",
            "policy is enabled but no credential provider is available; running disk-bundle-only — the resident bundle keeps enforcing and will not refresh"
        );
    }

    let startup_started = std::time::Instant::now();

    // Detect host IPs once at startup (bounded to ~3s worst case per resolver).
    // See `src/core/net/mod.rs`: failures collapse to None, the daemon never blocks on this.
    let host_ips = crate::net::HostIps::detect().await;
    tracing::info!(
        local_ipv4 = host_ips
            .local_ipv4
            .map(|a| a.to_string())
            .as_deref()
            .unwrap_or("none"),
        local_ipv6 = host_ips
            .local_ipv6
            .map(|a| a.to_string())
            .as_deref()
            .unwrap_or("none"),
        public_ipv4 = host_ips
            .public_ipv4
            .map(|a| a.to_string())
            .as_deref()
            .unwrap_or("none"),
        public_ipv6 = host_ips
            .public_ipv6
            .map(|a| a.to_string())
            .as_deref()
            .unwrap_or("none"),
        "host ips detected"
    );

    // Warm the process-global OS-user memo off the event path (I-1). On an
    // AD-joined or LDAP host `getpwuid_r` can reach NSS and take real
    // milliseconds; the first captured event must not be the one that pays for
    // it. Nothing awaits this — if it has not finished by the first hook, that
    // hook simply resolves it inline, exactly as it would have anyway.
    tokio::task::spawn_blocking(crate::daemon::identity::os_user);

    // Build the tamper-evidence logger alongside the event logger. The
    // handle is bound in function scope so the background writer task
    // stays alive for the daemon's lifetime — dropping the handle would
    // close the channel and the task would exit.
    let openlatch_dir_for_tamper = crate::config::openlatch_dir();
    let (tamper_logger, tamper_rx) = TamperLogger::channel();
    let tamper_rx = Arc::new(tokio::sync::Mutex::new(tamper_rx));
    let _tamper_logger_handle: TamperLoggerHandle =
        TamperLoggerHandle::from_task(spawn_supervised(
            &health,
            TaskSpec::new(subsystem::TAMPER_LOG_WRITER, RestartPolicy::OnFailure),
            logs_shutdown_rx.clone(),
            move || {
                let rx = tamper_rx.clone();
                let dir = openlatch_dir_for_tamper.clone();
                async move {
                    let mut rx = rx.lock_owned().await;
                    crate::core::logging::tamper_log::run_tamper_writer(dir, &mut rx).await
                }
            },
        ));

    // Configuration plane monitor — observes manifest-declared config files,
    // hashes them, forwards `ai.openlatch.config.*` CloudEvents through the
    // existing `cloud_tx` rail. Held on AppState (`content_hash_cache` for
    // dedup against native hooks; `config_monitor_request_tx` for admin /
    // CLI rescans). The handle stays alive for the daemon's lifetime.
    let cache_max = config.inventory_monitor.cache_max_entries.max(64);
    let content_hash_cache = Arc::new(config_monitor::ContentHashCache::new(cache_max));
    let mut config_monitor_handle: Option<config_monitor::ConfigMonitorHandle> = None;
    let mut config_monitor_request_tx: Option<
        tokio::sync::mpsc::Sender<config_monitor::ConfigChangeRequest>,
    > = None;
    if config.inventory_monitor.enabled {
        match config_monitor::manifest::load_embedded() {
            Ok(manifest) => {
                let monitor = config_monitor::ConfigMonitor::new(
                    Arc::new(manifest),
                    content_hash_cache.clone(),
                    privacy_filter.clone(),
                    cloud_tx.clone(),
                    event_logger.clone(),
                    Arc::new(config.clone()),
                );
                match monitor.spawn().await {
                    Ok(handle) => {
                        config_monitor_request_tx = Some(handle.request_tx.clone());
                        config_monitor_handle = Some(handle);
                        tracing::info!("config monitor active");
                    }
                    Err(e) => {
                        tracing::error!(
                            code = crate::error::ERR_INVENTORY_INIT_FAILED,
                            error = %e,
                            "config monitor failed to start"
                        );
                    }
                }
            }
            Err(e) => {
                tracing::error!(
                    code = crate::error::ERR_INVENTORY_MANIFEST_PARSE,
                    error = %e,
                    "failed to load inventory manifest; config monitoring disabled"
                );
            }
        }
    } else {
        tracing::info!("config monitor disabled by config");
    }

    // Model-boundary listener (plan 01 forward + plan 02 measurement). A SECOND
    // listener in THIS process, co-launched only on the full-daemon path
    // (`spawn_boundary`) — NOT on init's background setup daemon or tests, so
    // neither binds the pinned boundary port. It shares this daemon's session
    // registry (attribution) and cloud rail (economics emission).
    //
    // THIS BLOCK OWNS THE AGENT WIRING. The invariant it exists to hold:
    // `ANTHROPIC_BASE_URL` is present in the agent's settings.json IF AND ONLY
    // IF a listener holds the pinned port. It used to be split across two
    // owners — `init` wrote the base URL before anything bound, `openlatch
    // boundary disable` removed it — and the two diverged the moment `init
    // --foreground` ran: the config pointed every agent on the machine at 7600
    // and nothing ever bound it, so every Claude Code session died on
    // ECONNREFUSED. Whoever holds the port writes the config; nobody else does.
    //
    // The FIRST bind therefore happens HERE, outside the supervised task, and
    // its `Err` propagates out of `serve_with_listener` — the daemon exits and
    // settings.json is left untouched. A pre-occupied 7600 is a startup failure,
    // not a degraded mode, because a degraded mode is indistinguishable from a
    // healthy one from the agent's side.
    //
    // The task stays SUPERVISED with `RestartPolicy::Always` for everything
    // AFTER that first bind: a mid-life `axum::serve` error retries forever,
    // capped at 60 s, rebinding the SAME pinned port (never re-probing another,
    // D-25) — which is also what waits out the Windows TIME_WAIT rebind hazard
    // (mio sets SO_REUSEADDR on Unix only). The pre-bound listener is handed to
    // the first attempt through a `Mutex<Option<_>>`; restarts find it empty and
    // rebind. Retrying the same port is what keeps the config honest once
    // written; hard-failing the first bind is what keeps it from being written
    // dishonestly.
    //
    // The teardown broadcast is the daemon-wide `tasks_shutdown_tx` (OL-1300):
    // the boundary binds a SEPARATE pinned port (7600) that `/shutdown` never
    // reaches, so without an explicit stop it keeps that port bound after the
    // hook server drains and `openlatch stop` fails "process still running".
    //
    // The wiring itself is NOT written here. Binding proves the port is held; it
    // proves nothing about the leg that actually breaks in the field — reaching
    // `api.anthropic.com` through our own forward path. A boundary that binds
    // and cannot forward looks healthy from here and kills every session on the
    // machine, so the write is gated on a round trip instead of on a bind and
    // belongs to `boundary_wiring` below, which owns it for the daemon's whole
    // life (see `run_wiring_supervisor`).
    #[cfg(feature = "boundary")]
    #[allow(clippy::type_complexity)]
    let (boundary_task, wiring_task): (
        Option<tokio::task::JoinHandle<()>>,
        Option<tokio::task::JoinHandle<()>>,
    ) = if spawn_boundary {
        use crate::boundary;
        use crate::boundary::wire_format::WireFormat;
        let boundary_port = config.boundary.port;

        // Hard-fail: no listener, no wiring, no daemon.
        let first_listener = boundary::bind_pinned(boundary_port).await?;
        tracing::info!(
            port = boundary_port,
            "boundary listener bound (loopback only)"
        );

        // Outlives every serve attempt on purpose: a boundary restart rebuilds
        // `BoundaryState`, and a gate that reset to `pending` on each restart
        // would tell `init` and `doctor` "no verdict yet" about a listener the
        // supervisor has already judged.
        let wiring = Arc::new(boundary::preflight::WiringState::default());

        let boundary_patterns = config.extra_patterns.clone();
        // D-28: whether an acting `prefix_reorder` (L-0) rule may rewrite the
        // forwarded request. Ships `false`; L-1/L-2 are unaffected by it.
        let boundary_transforms_act = config.boundary.transforms_act;
        let boundary_registry = registry.clone();
        let boundary_cloud_tx = cloud_tx.clone();
        let boundary_shutdown_rx = tasks_shutdown_rx.clone();
        // Handed to the first attempt, empty for every restart after it.
        let pre_bound = Arc::new(tokio::sync::Mutex::new(Some(first_listener)));
        // Share the resident bundle so authored `request` rules — and the
        // `select` narrowing on them — reach the transform engine. Without this
        // the engine sees only its hardcoded baseline and every `select` key is
        // inert. `None` when the policy engine is off, which degrades to exactly
        // the previous behaviour.
        //
        // Cloned out here rather than read inside the closure: the closure is
        // `move`, and capturing `policy_runtime` itself would take it from the
        // `AppState` construction below.
        let boundary_policy = policy_runtime.as_ref().map(|p| p.handle.clone());
        let boundary_wiring = wiring.clone();
        // Resolved once, out here: a per-attempt parse would let a restart
        // silently change where every model call on this host is going.
        // `upstream_for` returns a STRING (it owns the precedence, not the
        // parse), so the parse is explicit here and FALLS BACK rather than
        // failing: this is the one path where a startup error takes every
        // session on the machine down.
        let boundary_upstream =
            reqwest::Url::parse(&config.boundary.upstream_for(WireFormat::AnthropicMessages))
                .unwrap_or_else(|_| boundary::default_upstream());
        // Resolved ONCE, walking every variant, so the state's lookup can never
        // miss. Without this the per-format map is unreachable and D-04/D-05 are
        // inert no matter how correct the config parsing is.
        let boundary_upstream_map: std::collections::BTreeMap<String, String> = WireFormat::ALL
            .iter()
            .map(|f| (f.as_str().to_string(), config.boundary.upstream_for(*f)))
            .collect();
        // Cloned into the supervised factory below. The forward CLIENT no longer comes
        // from here — it rides the shared handle just below, so a restart cannot revert
        // the boundary to a route a self-heal pass has already replaced.
        let boundary_egress = config.egress.clone();
        let boundary_client = egress_clients.boundary.clone();
        let boundary_reporter =
            crate::egress::EgressReporter::recording(&config.egress, egress_state.clone());
        let serve_task = spawn_supervised(
            &health,
            TaskSpec::new(subsystem::BOUNDARY, RestartPolicy::Always),
            tasks_shutdown_rx.clone(),
            move || {
                // Fresh state per attempt: a restart must not inherit the
                // semaphore permits or connection pool of the run that died.
                let bstate = Arc::new(
                    boundary::BoundaryState::new_with_egress(
                        boundary_upstream.clone(),
                        boundary_port,
                        boundary::DEFAULT_INFLIGHT,
                        &boundary_patterns,
                        &boundary_egress,
                    )
                    .with_upstream_map(boundary_upstream_map.clone())
                    .with_measurement(boundary_registry.clone(), boundary_cloud_tx.clone())
                    .with_transforms_act(boundary_transforms_act)
                    .with_policy(boundary_policy.clone())
                    .with_wiring(boundary_wiring.clone())
                    .with_client_handle(boundary_client.clone())
                    .with_egress_reporter(boundary_reporter.clone()),
                );
                boundary::serve_attempt(pre_bound.clone(), bstate, boundary_shutdown_rx.clone())
            },
        );

        // The gate. Its first tick runs immediately and IS the install-time
        // check — there is no separate one, so `init` and a 3 a.m. supervisor
        // restart are held to the same bar, and the wiring has exactly one
        // owner in both cases.
        let wiring_config = Arc::new(config.clone());
        let wiring_state = wiring.clone();
        let wiring_shutdown = tasks_shutdown_rx.clone();
        let wiring_task = spawn_supervised(
            &health,
            TaskSpec::new(subsystem::BOUNDARY_WIRING, RestartPolicy::Always),
            tasks_shutdown_rx.clone(),
            move || {
                run_wiring_supervisor(
                    wiring_config.clone(),
                    boundary_port,
                    wiring_state.clone(),
                    wiring_shutdown.clone(),
                )
            },
        );

        (Some(serve_task), Some(wiring_task))
    } else {
        // Reconciliation. The boundary is off, so nothing in this process will
        // ever hold 7600 — any `ANTHROPIC_BASE_URL` still on disk is a leftover
        // from a SIGKILLed daemon or a since-flipped config, and it points every
        // agent at a dead port. Clearing it at startup is what makes the
        // invariant self-healing rather than dependent on a clean shutdown.
        if reconcile_wiring {
            unwire_every_agent(&config);
        }
        (None, None)
    };

    let state = Arc::new(AppState {
        config: Arc::new(config.clone()),
        token,
        dedup: dedup::DedupStore::new(),
        event_logger,
        privacy_filter,
        event_counter: AtomicU64::new(0),
        shutdown_tx: tokio::sync::Mutex::new(Some(shutdown_tx)),
        started_at: std::time::Instant::now(),
        available_update: Mutex::new(None),
        cloud_tx,
        cloud_state: cloud_state_opt,
        local_ipv4: host_ips.local_ipv4,
        local_ipv6: host_ips.local_ipv6,
        public_ipv4: host_ips.public_ipv4,
        public_ipv6: host_ips.public_ipv6,
        tamper_logger: Some(tamper_logger),
        outbox: outbox_opt,
        update_in_progress: Arc::new(AtomicBool::new(false)),
        update_status: Arc::new(Mutex::new(update::UpdateStatusSnapshot::idle())),
        admin_shutdown_request: Arc::new(tokio::sync::Notify::new()),
        last_hook_at_unix_secs: Arc::new(AtomicU64::new(0)),
        hooks_in_flight: Arc::new(AtomicU32::new(0)),
        content_hash_cache,
        config_monitor_request_tx,
        pending_alerts,
        policy: policy_runtime,
        registry,
        health: health.clone(),
        egress: egress_state,
    });

    // Hold the config-monitor handle alive for the daemon's lifetime so the
    // watchers stay active; drop happens when this function returns.
    let _config_monitor_handle = config_monitor_handle;

    // Telemetry: daemon_started — port + measured startup duration + cloud
    // forwarding state. Captured here once we're committed to serving (the
    // listener is bound, state is built); the actual `axum::serve` call
    // happens immediately below.
    crate::telemetry::capture_global(crate::telemetry::Event::daemon_started(
        state.config.port,
        startup_started
            .elapsed()
            .as_millis()
            .min(u128::from(u64::MAX)) as u64,
        state.config.cloud.enabled,
    ));

    // Fallback-log replay: catch up on events the hook binary wrote while
    // the daemon was unreachable (offline reboot, daemon crashed between
    // hooks, etc). The task runs once at startup and then on every
    // `drain_notify` signal fired by the cloud worker.
    //
    // `OnFailure`: the loop never returns on its own, so a completion can only
    // mean "cloud forwarding is disabled, nothing to replay" — restarting that
    // would spin. A panic mid-replay still gets the task back.
    {
        let state_for_replay = state.clone();
        spawn_supervised(
            &health,
            TaskSpec::new(subsystem::FALLBACK_REPLAY, RestartPolicy::OnFailure),
            tasks_shutdown_rx.clone(),
            move || fallback_replay::run(state_for_replay.clone()),
        );
    }

    // UPDT-01: Spawn async update check at startup (T-02-14: bounded, non-blocking)
    if config.update.check {
        let current = env!("CARGO_PKG_VERSION").to_string();
        let update_egress = config.egress.clone();
        let state_for_update = state.clone();
        spawn_supervised(
            &health,
            TaskSpec::new(subsystem::UPDATE_CHECK, RestartPolicy::OnFailure),
            tasks_shutdown_rx.clone(),
            move || {
                let current = current.clone();
                let update_egress = update_egress.clone();
                let state_for_update = state_for_update.clone();
                async move {
                    if let Some(latest) = update::check_for_update(&current, &update_egress).await {
                        tracing::warn!(code = crate::error::ERR_VERSION_OUTDATED, latest_version = %latest, "Update available: run `npx openlatch@latest`");
                        state_for_update.set_available_update(latest);
                    }
                }
            },
        );
    }

    // Background auto-update worker. CI / cargo-install / disabled-by-config
    // are all short-circuited inside `run_auto_update_worker` so the daemon
    // startup path stays single-shape regardless of environment.
    //
    // `OnFailure`, not `Always`: those short-circuits are clean early returns,
    // and `Always` would restart them on a 60 s loop forever on every CI runner.
    if config.update.auto_update {
        let state_for_worker = state.clone();
        spawn_supervised(
            &health,
            TaskSpec::new(subsystem::AUTO_UPDATE_WORKER, RestartPolicy::OnFailure),
            tasks_shutdown_rx.clone(),
            move || run_auto_update_worker(state_for_worker.clone()),
        );
    } else {
        tracing::info!(target: "update", "auto-update worker disabled by config");
    }

    // Post-restart sentinel pickup. If the previous daemon swapped
    // itself just before we booted, an `update-sentinel.json` file is
    // sitting in `~/.openlatch/`. Wait for axum to bind, probe our own
    // `/health` endpoint, and on success clean up the `.bak` sibling +
    // sentinel + write install-state.json with the new version. On
    // probe failure, leave the artefacts in place — the
    // restart-loop rollback consumes them on the next start.
    if let Some(sentinel) = update::read_sentinel() {
        let port = config.port;
        spawn_supervised(
            &health,
            TaskSpec::new(subsystem::UPDATE_SENTINEL, RestartPolicy::OnFailure),
            tasks_shutdown_rx.clone(),
            move || {
                let sentinel = sentinel.clone();
                async move {
                    // Give axum ~5s to settle: bind, accept the first
                    // connection, finish wiring routes. The brainstorm doc
                    // calls this exact delay out (§ 4 "post-restart healthz
                    // probe ~5 s after binding").
                    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
                    if !probe_self_health(port).await {
                        tracing::warn!(target: "update", from = %sentinel.from, to = %sentinel.to, "post-restart /health probe failed; leaving sentinel + .bak in place for restart-loop rollback");
                        return;
                    }
                    tracing::info!(target: "update", from = %sentinel.from, to = %sentinel.to, "post-restart healthz probe succeeded; cleaning up");
                    if let Err(e) = update::cleanup_bak_files() {
                        tracing::warn!(target: "update", error = %e, "cleanup of .bak siblings failed (non-fatal)");
                    }
                    if let Err(e) = update::delete_sentinel() {
                        tracing::warn!(target: "update", error = %e, "delete of update sentinel failed (non-fatal)");
                    }
                    crate::install_state::InstallState::stamp_for_running_binary(env!(
                        "CARGO_PKG_VERSION"
                    ));
                }
            },
        );
    }

    // Spawn periodic dedup eviction to prevent unbounded memory growth
    let state_for_evict = state.clone();
    // Turbofished: the body is an infinite `loop`, so the future's output type
    // would otherwise fall back to `!` (a hard error from edition 2024 on).
    spawn_supervised::<_, _, ()>(
        &health,
        TaskSpec::new(subsystem::DEDUP_EVICTOR, RestartPolicy::Always),
        tasks_shutdown_rx.clone(),
        move || {
            let state = state_for_evict.clone();
            async move {
                let mut interval = tokio::time::interval(std::time::Duration::from_secs(30));
                loop {
                    interval.tick().await;
                    state.dedup.evict_expired();
                }
            }
        },
    );

    // Periodic log cleanup — supervised daemons never re-enter
    // init/lifecycle, so without this task audit JSONLs + rotated
    // `daemon.log.YYYY-MM-DD` files grow unbounded. Tick immediately
    // (catch up after long offline windows) then daily.
    let state_for_cleanup = state.clone();
    spawn_supervised::<_, _, ()>(
        &health,
        TaskSpec::new(subsystem::LOG_CLEANUP, RestartPolicy::Always),
        tasks_shutdown_rx.clone(),
        move || {
            let state = state_for_cleanup.clone();
            async move {
                let mut interval = tokio::time::interval(std::time::Duration::from_secs(86_400));
                loop {
                    interval.tick().await;
                    let log_dir = state.config.log_dir.clone();
                    let retention = state.config.retention_days;
                    match tokio::task::spawn_blocking(move || {
                        crate::logging::cleanup_old_logs(&log_dir, retention)
                    })
                    .await
                    {
                        Ok(Ok(deleted)) if deleted > 0 => {
                            tracing::info!(
                                deleted,
                                retention_days = retention,
                                "cleaned up old log files"
                            );
                        }
                        Ok(Ok(_)) => {}
                        Ok(Err(e)) => {
                            tracing::warn!(error = %e, "periodic log cleanup failed");
                        }
                        Err(e) => {
                            tracing::warn!(error = %e, "periodic log cleanup task join error");
                        }
                    }
                }
            }
        },
    );

    // Tamper-evidence: reconciler + reactive watcher + poll safety net.
    // Startup reconciliation runs BEFORE axum binds to catch Scenario-2 drift.
    let _watcher_guards;
    let _poll_handle;
    // PLURAL. This was `detect_agent()` — the FIRST agent — the last such call
    // site left in the daemon. Every other agent's hooks were then registered,
    // enforced, and watched by nothing: no tamper detection, no heal, and a
    // `doctor` that still called them installed because the file it read was
    // the one nobody had touched.
    let detected = crate::hooks::detect_agents();
    if !detected.is_empty() {
        // One target per agent, each pairing the file with the binding that
        // heals it, so a heal can never reach a different agent from the one
        // the drift was observed in.
        let targets: Vec<reconciler::AgentTarget> = detected
            .iter()
            .map(reconciler::AgentTarget::for_agent)
            .collect();
        let openlatch_dir = crate::config::openlatch_dir();
        let token_file = openlatch_dir.join("daemon.token");

        reconciler::run_startup_reconcile(targets.clone(), &openlatch_dir, config.port);

        let (reconcile_tx, reconcile_rx) = tokio::sync::mpsc::channel(100);

        let sinks = reconciler::TamperSinks {
            logger: state.tamper_logger.clone(),
            cloud_tx: state.cloud_tx.clone(),
            agent_id: state.config.agent_id.clone().unwrap_or_default(),
            // Rides the same `clientversion` wire attribute as a hook event, so
            // it uses the same source — a tamper event reporting a different
            // version from the hook events beside it is the drift
            // OPENLATCH_VERSION exists to remove.
            client_version: env!("OPENLATCH_VERSION").to_string(),
        };

        // Behind an `Arc<Mutex<_>>` so a panic mid-reconcile restarts onto the
        // same request channel rather than leaving tamper detection dead for
        // the rest of the daemon's life while `/health` reports `ok`.
        let r = Arc::new(tokio::sync::Mutex::new(
            reconciler::Reconciler::new_with_sinks(
                reconcile_rx,
                targets.clone(),
                openlatch_dir,
                config.port,
                token_file,
                sinks,
            ),
        ));
        spawn_supervised(
            &health,
            TaskSpec::new(subsystem::RECONCILER, RestartPolicy::Always),
            tasks_shutdown_rx.clone(),
            move || {
                let r = r.clone();
                async move {
                    let mut guard = r.lock_owned().await;
                    guard.run().await
                }
            },
        );

        // One watcher per agent config file, all feeding the one reconcile
        // channel: any file changing triggers a pass over every target, which
        // costs a re-read of the others and buys the guarantee that a request
        // can never arrive for a file nobody is watching. The guards are held
        // for the daemon's lifetime — dropping one stops its watcher.
        _watcher_guards = targets
            .iter()
            .filter_map(|target| {
                match watcher::spawn_watcher(&target.settings_path, reconcile_tx.clone()) {
                    Ok(w) => {
                        tracing::info!(
                            agent = target.binding.agent_type(),
                            "filesystem watcher active"
                        );
                        Some(w)
                    }
                    Err(e) => {
                        tracing::warn!(
                            error = %e,
                            agent = target.binding.agent_type(),
                            "filesystem watcher failed — falling back to poll-only"
                        );
                        None
                    }
                }
            })
            .collect();

        _poll_handle = Some(watcher::spawn_poll_fallback(reconcile_tx));
        tracing::info!(
            agents = targets.len(),
            "reconciler started (reactive watcher + 30s poll)"
        );
    } else {
        _watcher_guards = Vec::new();
        _poll_handle = None;
    }

    // Hook route — single generic CloudEvents v1.0.2 ingest endpoint. The
    // handler validates Content-Type inline (accepts
    // application/cloudevents+json, application/cloudevents-batch+json, and
    // application/json during the transition) and parses the envelope as
    // either a single object or a JSON array.
    let hook_routes = Router::new()
        .route("/hooks", post(handlers::ingest_cloudevent))
        .route_layer(middleware::from_fn_with_state(
            state.clone(),
            auth::bearer_auth,
        ));

    // Shutdown route — requires Bearer token but no JSON body
    let shutdown_route = Router::new()
        .route("/shutdown", post(handlers::shutdown_handler))
        .route_layer(middleware::from_fn_with_state(
            state.clone(),
            auth::bearer_auth,
        ));

    // Public routes — no authentication required
    let public_routes = Router::new()
        .route("/health", get(handlers::health))
        .route("/metrics", get(handlers::metrics));

    // Admin routes — Bearer-auth gated. Hosts the manual-update RPC
    // (`POST /admin/update`) and the long-poll status endpoint
    // (`GET /admin/update/status`). Mounted under the `/admin/*` prefix
    // so future privileged endpoints share the same auth posture.
    let admin_routes = admin::router(state.clone());

    let app = Router::new()
        .merge(hook_routes)
        .merge(shutdown_route)
        .merge(public_routes)
        .merge(admin_routes)
        // SECURITY: 1MB body limit — reject oversized payloads with 413 before parsing
        .layer(DefaultBodyLimit::max(1_048_576))
        .with_state(state.clone());

    let admin_shutdown_request = state.admin_shutdown_request.clone();
    axum::serve(listener, app)
        .with_graceful_shutdown(async move {
            tokio::select! {
                _ = signal_handler() => {
                    tracing::info!("received OS shutdown signal");
                }
                _ = shutdown_rx => {
                    tracing::info!("received shutdown via /shutdown endpoint");
                }
                _ = admin_shutdown_request.notified() => {
                    tracing::info!(target: "update", "received shutdown for in-flight auto-update");
                }
            }
        })
        .await?;

    // One broadcast stops every supervisor AND every supervised loop: no
    // supervisor respawns after this point, and the loops that listen to the
    // same channel (cloud worker's flush, boundary's graceful drain) start
    // draining immediately.
    let _ = tasks_shutdown_tx.send(true);

    // Give the cloud worker a bounded window to finish its flush. Without the
    // wait the buffered events would die with the runtime; with an unbounded
    // wait a wedged cloud could block `openlatch stop` indefinitely. Anything
    // it cannot POST in the window it has already spooled to the outbox, so
    // nothing is silently lost.
    if let Some(cloud_worker) = cloud_worker_task {
        if tokio::time::timeout(std::time::Duration::from_secs(5), cloud_worker)
            .await
            .is_err()
        {
            tracing::warn!(
                "cloud worker did not finish its shutdown flush within 5s — abandoning it"
            );
        }
    }

    // OL-1300: join the boundary supervisor so the pinned port (7600) is
    // released before the runtime drops. Without this the boundary task
    // outlives `/shutdown`, keeps the process alive, and `openlatch
    // stop`/`restart` fails "process still running". Bounded like the cloud
    // worker so a wedged listener can't hang stop.
    #[cfg(feature = "boundary")]
    if let Some(boundary_task) = boundary_task {
        // The gate first: it is the only thing that writes the agent config, and
        // a probe still in flight could re-wire on its way out, right after the
        // teardown below has cleared it.
        if let Some(wiring_task) = wiring_task {
            if tokio::time::timeout(std::time::Duration::from_secs(5), wiring_task)
                .await
                .is_err()
            {
                tracing::warn!("boundary wiring supervisor did not stop within 5s — abandoning it");
            }
        }
        if tokio::time::timeout(std::time::Duration::from_secs(5), boundary_task)
            .await
            .is_err()
        {
            tracing::warn!("boundary listener did not shut down within 5s — abandoning it");
        }
        // The other half of the invariant. We are no longer holding the pinned
        // port, so the agent config must stop claiming we are — otherwise every
        // Claude Code session started after this stop dies on ECONNREFUSED,
        // which is precisely the failure this ownership move exists to end.
        //
        // Reached from every graceful teardown: `signal_handler` (ctrl_c /
        // SIGTERM / SIGHUP), `POST /shutdown`, and the auto-update restart —
        // they all funnel through the single `axum::serve` graceful-shutdown
        // above. SIGKILL escapes it by construction; `openlatch stop` and the
        // next daemon start reconcile that case.
        //
        // Isolated instances skip it: they never wrote the file, so removing
        // from it would revoke the canonical daemon's wiring. Asked per agent,
        // inside the loop — a bool captured before the bindings exist cannot
        // answer a per-binding question.
        unwire_every_agent(&state.config);
    }

    // Capture final stats before releasing state
    let uptime_secs = state.started_at.elapsed().as_secs();
    let events = state
        .event_counter
        .load(std::sync::atomic::Ordering::Relaxed);

    crate::logging::daemon_log::log_shutdown(uptime_secs, events);
    crate::telemetry::capture_global(crate::telemetry::Event::daemon_stopped(uptime_secs, events));
    // Capture only enqueues. The global handle's sender never closes (it lives
    // in a `OnceLock`), so nothing triggers the batch task's final drain and
    // the batch timer loses the race with process exit — `daemon_stopped` was
    // being captured and then dropped on the floor every single shutdown.
    // Bounded by `telemetry::FLUSH_BUDGET`; an unreachable endpoint delays the
    // exit by that much and no more.
    if !crate::telemetry::flush_global().await {
        tracing::debug!("telemetry: final flush did not complete within budget");
    }

    // Release the Arc so `EventLogger`'s sender is dropped, then drain the writer
    // — but ONLY when that drop actually made us the last sender.
    //
    // `EventLoggerHandle::shutdown` joins the writer task, and the writer only
    // exits once its channel CLOSES, i.e. once the last `EventLogger` sender is
    // gone. Its own doc states the precondition: "the caller must drop the
    // sender BEFORE calling this method, otherwise the writer task will block
    // waiting for more events." This call site used to detect that the
    // precondition was violated, warn about it, and then await anyway.
    //
    // `try_unwrap` fails whenever a detached background task (reconciler,
    // filesystem watcher, dedup-eviction loop) still holds an `Arc<AppState>`
    // clone — the NORMAL case, not a rare one; those tasks are only reaped when
    // the runtime drops, and the runtime cannot drop while we are still awaiting
    // here. So the await deadlocked the daemon on EVERY shutdown: it drained both
    // listeners, released both ports, logged "daemon stopped", and then hung
    // forever holding no port. `openlatch stop` saw a live pid, fell through
    // graceful `/shutdown` and SIGTERM, and dead-ended at OL-1300 "process still
    // running" 100% of the time.
    //
    // Skipping the join in that branch costs nothing observable: both listeners
    // are already closed, so no new events can be produced, and the writer
    // flushes after every drained batch — at worst a final in-flight batch is
    // abandoned, exactly what the pre-existing warning already advertises. The
    // sole-owner join stays bounded so a wedged disk cannot hang `stop` either.
    match Arc::try_unwrap(state) {
        Ok(_state) => {
            // Sole owner — the sender is gone, so the writer will observe the
            // channel close and exit. Bounded anyway: a wedged disk write must
            // never be able to hang `openlatch stop`.
            if tokio::time::timeout(std::time::Duration::from_secs(5), logger_handle.shutdown())
                .await
                .is_err()
            {
                tracing::warn!("event-log drain did not finish within 5s — abandoning it");
            }
        }
        Err(arc) => {
            tracing::warn!(
                strong_refs = Arc::strong_count(&arc),
                "AppState still has references at shutdown — final event-log batch may be dropped"
            );
            drop(arc);
            // Deliberately NOT joining the writer: our sender is not the last
            // one, so the channel never closes and the join could never return.
        }
    }

    Ok((uptime_secs, events))
}

/// Format a duration in seconds as a human-readable uptime string.
///
/// Examples: `"45s"`, `"3m12s"`, `"2h14m"`
pub fn format_uptime(secs: u64) -> String {
    let hours = secs / 3600;
    let minutes = (secs % 3600) / 60;
    let seconds = secs % 60;
    if hours > 0 {
        format!("{}h{}m", hours, minutes)
    } else if minutes > 0 {
        format!("{}m{}s", minutes, seconds)
    } else {
        format!("{}s", seconds)
    }
}

/// Probe our own `/health` endpoint to confirm a freshly-restarted
/// daemon is healthy enough to discard its `.bak` siblings + sentinel.
/// 2-second timeout matches the existing startup-update-check budget.
async fn probe_self_health(port: u16) -> bool {
    let Ok(client) = crate::egress::client_builder()
        .timeout(std::time::Duration::from_secs(2))
        .build()
    else {
        return false;
    };
    let url = format!("http://127.0.0.1:{port}/health");
    match client.get(&url).send().await {
        Ok(r) => r.status().is_success(),
        Err(_) => false,
    }
}

// ---------------------------------------------------------------------------
// Background auto-update worker
// ---------------------------------------------------------------------------

/// Long-running task that polls the npm registry and applies updates
/// during quiet windows. Lives for the daemon's lifetime; the tokio
/// runtime drops it on shutdown along with every other detached task
/// (same pattern as the dedup-eviction loop above).
///
/// On `auto_update = true` daemons the worker starts ~10 s after
/// startup to give the rest of the daemon (telemetry, reconciler,
/// fallback replay) time to settle. The first poll fires immediately
/// — the loop is `sleep`-after-poll, not `interval.tick()`-before-poll,
/// so the tail of a 6 h cadence never delays the first check.
///
/// In CI (auto-detected via `telemetry::is_ci_environment()`) the
/// worker is a no-op — every CI job is a fresh install, applying mid-
/// run only churns telemetry. `cargo install`-managed binaries are
/// likewise skipped: the auto-update path refuses them and the user
/// must `cargo install --force`.
async fn run_auto_update_worker(state: Arc<AppState>) {
    use crate::install_state::{detect_install_method, InstallMethod};

    if crate::telemetry::is_ci_environment() {
        tracing::debug!(target: "update", "CI environment detected; auto-update worker disabled");
        return;
    }

    if matches!(detect_install_method(), InstallMethod::CargoInstall) {
        tracing::info!(target: "update", "cargo-install path detected; auto-update worker disabled — use `cargo install --force --locked openlatch-client`");
        return;
    }

    // ~10 s settle delay so we don't compete with reconciler startup.
    tokio::time::sleep(std::time::Duration::from_secs(10)).await;

    let normal_interval =
        std::time::Duration::from_secs(state.config.update.check_interval_secs.max(1));
    let critical_interval = std::time::Duration::from_secs(3600);
    let defer_interval = std::time::Duration::from_secs(300);

    let mut pending_since: Option<std::time::Instant> = None;
    let current_version = env!("CARGO_PKG_VERSION").to_string();

    loop {
        let next_sleep = match worker_iteration(&state, &current_version, pending_since).await {
            WorkerOutcome::Idle => {
                pending_since = None;
                normal_interval
            }
            WorkerOutcome::Deferred {
                severity: update::Severity::Critical,
            } => {
                if pending_since.is_none() {
                    pending_since = Some(std::time::Instant::now());
                }
                critical_interval
            }
            WorkerOutcome::Deferred { .. } => {
                if pending_since.is_none() {
                    pending_since = Some(std::time::Instant::now());
                }
                defer_interval
            }
            WorkerOutcome::Failed {
                severity: update::Severity::Critical,
            } => {
                pending_since = None;
                critical_interval
            }
            WorkerOutcome::Failed { .. } => {
                pending_since = None;
                normal_interval
            }
        };

        // No explicit shutdown listener: `Notify::notify_waiters` drops
        // notifications fired while the worker is mid-iteration, which
        // would strand the loop in a multi-hour sleep until runtime
        // drop. Matching the dedup-eviction loop above, the tokio
        // runtime's drop on daemon shutdown cancels this task.
        tokio::time::sleep(next_sleep).await;
    }
}

#[derive(Debug, Clone, Copy)]
enum WorkerOutcome {
    Idle,
    Deferred { severity: update::Severity },
    Failed { severity: update::Severity },
}

/// One iteration of the worker loop: probe, decide, optionally apply.
async fn worker_iteration(
    state: &Arc<AppState>,
    current_version: &str,
    pending_since: Option<std::time::Instant>,
) -> WorkerOutcome {
    let registry_origin = state.config.update.registry_origin.clone();
    let download_timeout =
        std::time::Duration::from_secs(state.config.update.download_timeout_secs.max(1));

    let result = update::check(current_version, &registry_origin, &state.config.egress).await;
    let (latest, severity, min_supported) = match result {
        update::CheckResult::Available {
            latest,
            severity,
            min_supported,
            ..
        } => (latest, severity, min_supported),
        update::CheckResult::UpToDate { .. } | update::CheckResult::Failed { .. } => {
            return WorkerOutcome::Idle;
        }
    };

    // Mirror the admin RPC's min_supported_client gate. Without this
    // the worker would take the apply lock + emit `update_started` only
    // for `prepare_swap_artefacts` to refuse a few seconds later.
    if let Some(ref min) = min_supported {
        if !update::version_at_least(current_version, min) {
            crate::telemetry::capture_global(
                crate::telemetry::Event::update_blocked_by_min_supported(
                    current_version,
                    &latest,
                    min,
                ),
            );
            tracing::info!(
                target: "update",
                latest = %latest,
                min_supported = %min,
                "auto-update blocked: client older than min_supported_client"
            );
            return WorkerOutcome::Idle;
        }
    }

    let pending_age = pending_since
        .map(|t| std::time::Instant::now().saturating_duration_since(t))
        .unwrap_or_default();

    if !update::should_apply_now(
        severity,
        &state.last_hook_at_unix_secs,
        &state.hooks_in_flight,
        pending_age,
        state.config.update.quiet_window_secs,
        state.config.update.max_defer_secs,
    ) {
        tracing::debug!(target: "update", latest = %latest, severity = %severity.as_str(), "deferring update — agent active or quiet window not met");
        return WorkerOutcome::Deferred { severity };
    }

    if state
        .update_in_progress
        .compare_exchange(
            false,
            true,
            std::sync::atomic::Ordering::AcqRel,
            std::sync::atomic::Ordering::Acquire,
        )
        .is_err()
    {
        tracing::info!(target: "update", "auto-update worker yielding to in-flight manual apply");
        return WorkerOutcome::Deferred { severity };
    }

    // Stamp the long-poll status snapshot so a concurrent
    // `GET /admin/update/status` sees the worker's progress instead of
    // the prior idle/completed snapshot.
    {
        let mut snap = state.update_status.lock().expect("status mutex poisoned");
        *snap = update::UpdateStatusSnapshot {
            status: update::UpdateStatusKind::InProgress,
            stage: Some(update::ApplyStage::Check),
            from: Some(current_version.to_string()),
            to: Some(latest.clone()),
            started_at: Some(crate::install_state::now_rfc3339()),
            ended_at: None,
            error: None,
        };
    }

    let opts = update::ApplyOptions {
        current_version: current_version.to_string(),
        registry_origin,
        download_timeout,
        force_cargo_install: false,
        mode: update::ApplyMode::Rpc,
        egress: state.config.egress.clone(),
    };

    // Hand off to the same apply path the manual RPC uses. On success
    // it never returns (process restarts); on failure it releases the
    // lock itself.
    admin::run_apply_in_daemon(state.clone(), opts, severity).await;

    // If we get here the apply failed. The lock has been released by
    // `run_apply_in_daemon`; report failed for backoff purposes.
    WorkerOutcome::Failed { severity }
}

// ---------------------------------------------------------------------------
// Agent wiring — owned by whoever holds the pinned boundary port
// ---------------------------------------------------------------------------

/// May THIS daemon write THIS agent's boundary wiring?
///
/// [`crate::config::BoundaryConfig::owns_agent_wiring`] answers the same
/// question process-wide, but it resolves an explicit opt-in through
/// `claude_code::config_is_machine_global()` — one agent's shape, asked of
/// every agent. On a sandbox with `own_agent_wiring = true`, a non-default port,
/// a relocated `CLAUDE_CONFIG_DIR` and no `CODEX_HOME`, that predicate answers
/// *yes* (Claude's config is not machine-global) and the loop then writes the
/// developer's **real** `~/.codex/config.toml`. So the daemon composes the
/// question per binding instead, from the two facts it already has.
///
/// `core::config` is deliberately untouched: threading a binding into it is a
/// scope expansion the binding contract does not sanction, and the process-wide
/// predicate keeps its CLI callers, which have no binding in hand.
#[cfg(feature = "boundary")]
fn owns_wiring_for(config: &Config, binding: &dyn crate::hooks::binding::AgentBinding) -> bool {
    let on_default_port = config.boundary.port == crate::boundary::default_boundary_port();
    match config.boundary.own_agent_wiring {
        Some(false) => false,
        None => on_default_port,
        // Belt and braces, per agent: an explicit opt-in still may not reach
        // THIS agent's machine-global config from a non-default port. Opting in
        // is a decision about your own sandbox, not a way to seize a shared
        // file — and one relocated agent beside one machine-global agent on the
        // same daemon must get different answers.
        Some(true) => on_default_port || !binding.config_is_machine_global(),
    }
}

/// How often the wiring supervisor comes back around.
#[cfg(feature = "boundary")]
const WIRING_TICK: std::time::Duration = std::time::Duration::from_secs(60);

/// Ceiling on the backoff between probes while the gate is shut.
///
/// The failure modes that keep it shut — no network, captive portal, a VPN that
/// has not come up — resolve on human timescales, so a five-minute ceiling
/// re-wires promptly enough while a laptop that spends a day offline pays a
/// handful of probes rather than a thousand.
#[cfg(feature = "boundary")]
const WIRING_BACKOFF_MAX: std::time::Duration = std::time::Duration::from_secs(300);

/// Delay before the next probe, given the loop's base tick and how many probes
/// have failed back-to-back.
///
/// Zero failures means the gate is open and the loop is idling at its tick.
/// Otherwise: exponential from the tick, capped. Pulled out as a pure function —
/// `base` passed in rather than read from a global — because a backoff that
/// silently stops backing off is the kind of bug that only surfaces as a support
/// ticket about provider rate limits.
#[cfg(feature = "boundary")]
fn wiring_delay(base: std::time::Duration, consecutive_failures: u32) -> std::time::Duration {
    if consecutive_failures == 0 {
        return base;
    }
    let shift = (consecutive_failures - 1).min(8);
    base.saturating_mul(1u32 << shift).min(WIRING_BACKOFF_MAX)
}

/// The wiring loop's base tick.
///
/// [`WIRING_TICK`] in production. The env override exists for one reason: the
/// self-healing behaviour — unwire when the provider goes away, re-wire when it
/// comes back — is only observable across ticks, and a test that waited a real
/// minute per transition would not be run. Floored so it can never become a busy
/// loop, and never set outside a test harness: a tick short enough to be
/// testable would probe the provider often enough to look like abuse.
#[cfg(feature = "boundary")]
fn wiring_tick() -> std::time::Duration {
    match std::env::var("OPENLATCH_BOUNDARY_WIRING_TICK_MS") {
        Ok(v) => match v.parse::<u64>() {
            Ok(ms) => std::time::Duration::from_millis(ms.max(50)),
            Err(_) => WIRING_TICK,
        },
        Err(_) => WIRING_TICK,
    }
}

/// Own the agent wiring for the daemon's whole life: prove the boundary can
/// actually forward, then write `ANTHROPIC_BASE_URL` — and take it back the
/// moment that stops being true.
///
/// The first tick runs immediately and is the install-time gate; every tick
/// after it is the watchdog. They are the same code on purpose. An install-time
/// check alone only ever proves the forwarder worked once, at a moment nobody
/// was using it, and the failures that hurt — a provider change, a VPN coming
/// up, a regression in the forward path — all arrive later, with the agent
/// already pointed at us.
///
/// **The probe is not run on every tick.** While the gate is open, it fires only
/// when [`crate::boundary::proxy::upstream_failures`] has grown since the last
/// look — real traffic failing is the signal, and a boundary quietly serving a
/// working session needs no synthetic request to prove it. While the gate is
/// shut there is no traffic to learn from, so it probes on a backoff until
/// upstream comes back.
///
/// It watches `upstream_failures`, NOT `pass_through_failures`: the latter
/// counts fallible OpenLatch steps that degraded to forwarding unmodified, which
/// the agent never notices. Only the synthetic 502 means the agent got nothing,
/// and only that is evidence the wiring should be reconsidered.
///
/// Every failure path here leaves agents talking straight to the provider. That
/// is degraded — nothing is captured — but it is honest and it works, which is
/// the one thing a dangling `ANTHROPIC_BASE_URL` is not.
#[cfg(feature = "boundary")]
async fn run_wiring_supervisor(
    config: Arc<Config>,
    port: u16,
    wiring: Arc<crate::boundary::preflight::WiringState>,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
) {
    use crate::boundary::proxy::upstream_failures;

    let mut last_failures = upstream_failures();
    let mut consecutive_failures: u32 = 0;
    let tick = wiring_tick();

    loop {
        if *shutdown.borrow() {
            return;
        }

        let failures = upstream_failures();
        let forwarding_broke = failures > last_failures;
        last_failures = failures;

        let agents = crate::hooks::detect_agents();
        // Seed before the first probe. Without it an agent that has a request
        // plane but no verdict yet has no key at all in the status JSON, and
        // every consumer — `classify_boundary`, `init`'s wait rule, `doctor` —
        // falls through to its "some other reason" arm and calls a host that is
        // merely still checking `unwired`. None of those failures is a compile
        // error.
        for a in &agents {
            if a.binding.boundary_wiring().is_some() {
                wiring.seed(a.agent_type());
            }
        }

        let any_failed = wire_agents(agents, port, forwarding_broke, &config, &wiring).await;
        // Backoff stays PROCESS-WIDE and counts a tick in which any agent
        // failed. Not hammering a dead network is a host-level concern; a timer
        // per agent would be three timers for one cause.
        if any_failed {
            consecutive_failures = consecutive_failures.saturating_add(1);
        } else {
            consecutive_failures = 0;
        }

        let delay = wiring_delay(tick, consecutive_failures);
        tokio::select! {
            _ = tokio::time::sleep(delay) => {}
            _ = shutdown.changed() => return,
        }
    }
}

/// One tick of the wiring loop: probe each agent **in its own format** and wire
/// the ones that come back green.
///
/// Returns `true` when any agent's probe failed, which is the only thing the
/// supervisor's process-wide backoff needs to learn from a per-agent pass.
///
/// Extracted from [`run_wiring_supervisor`] so it can be tested at all: the
/// supervisor is an infinite loop with a shutdown watch, and `detect_agents()`
/// has no injection seam — the tests hand this function two fake-binding agents
/// directly and drive the real probe over a real ephemeral listener.
///
/// **Continue, never abort.** A Codex failure must not leave Claude Code's
/// request plane unwired on a host that had both: that is losing coverage we
/// already had. The per-agent unwire in the `Err` arm replaces a blanket one
/// that dropped every agent on one failure.
#[cfg(feature = "boundary")]
async fn wire_agents(
    agents: Vec<crate::hooks::DetectedAgent>,
    port: u16,
    // Computed once by the supervisor from `upstream_failures()`; never
    // recomputed here, so every agent in one tick sees the same answer.
    forwarding_broke: bool,
    config: &Config,
    wiring: &crate::boundary::preflight::WiringState,
) -> bool {
    use crate::boundary::preflight::{self, Verdict};

    let mut any_failed = false;
    for agent in agents {
        // No request plane at all — nothing to probe and nothing to write.
        // Not an error: it is a question that does not apply to this agent.
        let Some(w) = agent.binding.boundary_wiring() else {
            continue;
        };
        let a = agent.agent_type();
        // Shut gate → probe until it opens. Open gate → only when live traffic
        // has started failing. Per agent, because one agent's open gate says
        // nothing about another's.
        if wiring.is_wired(a) && !forwarding_broke {
            continue;
        }

        // The probe speaks THIS agent's format against THIS format's upstream.
        // Probing as Anthropic and then wiring Codex would prove a Claude round
        // trip and open the gate for a plane nobody checked — a disabled
        // subsystem rendering healthy, which is the failure the gate exists to
        // stop.
        let upstream = config.boundary.upstream_for(w.wire_format);
        match preflight::probe(port, w.wire_format, &upstream, preflight::PREFLIGHT_TIMEOUT).await {
            Ok(()) => {
                wiring.set_verdict(a, Verdict::Ok);
                // ONLY when this agent is not already wired. The loop re-probes
                // a wired agent on every `forwarding_broke` tick on purpose;
                // writing on every green re-probe would rewrite the customer's
                // config file and re-emit the wiring log lines on every
                // upstream hiccup.
                if !wiring.is_wired(a) {
                    // THE DAEMON WRAPPER, never `hooks::write_boundary_config`:
                    // the wrapper carries the isolated-instance guard and the
                    // install-id resolution, and skipping it is how an `olbox`
                    // daemon starts writing the developer's real agent config.
                    wire_boundary_config(config, &*agent.binding, port, wiring);
                }
            }
            Err(reason) => {
                any_failed = true;
                // These two lines are the ONLY operator-visible signal that a
                // request plane was torn off disk. `agent` is on both, because
                // on a two-agent host "which one" is the whole question.
                if wiring.is_wired(a) {
                    tracing::error!(
                        port,
                        agent = a,
                        reason = %reason,
                        "boundary preflight failed on a wired listener — removing the agent \
                         wiring so sessions fall back to a direct provider connection"
                    );
                } else {
                    tracing::warn!(
                        port,
                        agent = a,
                        reason = %reason,
                        "boundary preflight failed — the agent stays unwired and model calls \
                         go straight to the provider (nothing is captured)"
                    );
                }
                // The old `attempt = consecutive_failures` field is gone: the
                // counter is the supervisor's, and this tick may carry more
                // than one agent's failure.
                wiring.set_verdict(a, Verdict::Failed(reason));
                // Per-agent unwire, replacing a blanket one. The self-healing
                // reason still holds — a SIGKILLed daemon leaves its wiring
                // behind and this process starts with `is_wired` false while
                // the file still points at us — it is just scoped to the agent
                // whose probe failed.
                //
                // The guard is at the CALL SITE: `unwire_boundary_config` has
                // none of its own, and an unguarded unwire in an `olbox` daemon
                // strips the boundary wiring out of the developer's real config.
                if owns_wiring_for(config, &*agent.binding) {
                    unwire_boundary_config(config, &*agent.binding);
                    // Both statements, always. Drop this one and a failed probe
                    // unwires the file while leaving the in-memory gate open, so
                    // the loop's own `if wiring.is_wired(a)` line skips the
                    // agent on every later tick: permanently unwired on disk,
                    // permanently "wired" in state, never re-probed.
                    wiring.set_wired(a, false);
                }
            }
        }
    }
    any_failed
}

/// Point ONE agent at the boundary listener, once a preflight probe has proven
/// it can actually forward.
///
/// Call sites: exactly one, [`wire_agents`], and only on that agent's own green
/// probe. A failure to write is logged, never fatal — "listener up, config not
/// written" leaves agents talking straight to the provider, which is degraded
/// but honest. The reverse ordering is the one that is not survivable.
///
/// **Not a thin pass-through over [`crate::hooks::write_boundary_config`].** It
/// carries the isolated-instance guard and the install-id resolution, and a
/// loop that called the `hooks` function directly would make an `olbox` or
/// non-default-port daemon write the developer's real agent config — the
/// two-owner divergence the ownership move removed, reappearing as two daemons
/// instead of two commands.
#[cfg(feature = "boundary")]
fn wire_boundary_config(
    config: &Config,
    binding: &dyn crate::hooks::binding::AgentBinding,
    port: u16,
    wiring: &crate::boundary::preflight::WiringState,
) {
    // The file this agent's wiring goes in — `settings.json` for Claude Code,
    // `config.toml` for Codex. Never `hook_config_path()` for both: naming a
    // file the writer did not touch is how a log line stops being evidence.
    let path = crate::hooks::boundary_config_path(binding);
    let path_label = path
        .as_ref()
        .map(|p| p.display().to_string())
        .unwrap_or_else(|| "the agent config".into());

    // An isolated instance (non-default port) binds but does not touch a
    // machine-global agent config: that file has one owner, the daemon on the
    // default port, and a second writer would take the wiring out from under
    // it. Asked PER BINDING — one relocated agent beside one machine-global
    // agent on the same daemon get different answers.
    if !owns_wiring_for(config, binding) {
        tracing::info!(
            port,
            agent = binding.agent_type(),
            "isolated boundary instance — {path_label} left untouched; route sessions through \
             this listener explicitly ({})",
            isolated_wiring_hint(binding, port),
        );
        return;
    }
    // PII-free per-install id (F-22) — reuses the existing `agent_id`,
    // provisioning one if a pre-`init` config lacks it.
    let install_id = match config.agent_id.clone() {
        Some(id) => id,
        None => crate::config::ensure_agent_id(&crate::config::openlatch_dir().join("config.toml"))
            .unwrap_or_default(),
    };
    match crate::hooks::write_boundary_config(binding, port, &install_id) {
        Ok(()) => {
            wiring.set_wired(binding.agent_type(), true);
            // The protocol we just configured this agent to speak. Recorded
            // here, at the write, so it can never claim a format the file does
            // not actually name — it is what lets an unattributable request be
            // traced back to the sole agent on this install speaking its format
            // instead of being reported as `unknown`.
            if let Some(w) = binding.boundary_wiring() {
                wiring.set_wired_format(binding.agent_type(), w.wire_format);
            }
            tracing::info!(
                port,
                agent = binding.agent_type(),
                path = %path_label,
                "agent wired to the model boundary — model calls route via http://127.0.0.1:{port}"
            );
            // P7: surfaced where the write actually happens, so the note can
            // never outlive the thing it describes.
            tracing::info!(
                "Claude Code disables Remote Control while ANTHROPIC_BASE_URL is set — \
                 stop the daemon (`openlatch stop`) to restore a direct connection"
            );
        }
        Err(e) => {
            tracing::error!(
                code = %e.code,
                error = %e.message,
                agent = binding.agent_type(),
                path = %path_label,
                "failed to wire agent to the model boundary — agents will talk to the provider directly"
            );
        }
    }
}

/// How an operator points ONE session at an isolated listener by hand, in that
/// agent's own vocabulary.
///
/// Naming `ANTHROPIC_BASE_URL` at a Codex agent is the same defect as pointing
/// a Codex user at `api.anthropic.com`: an instruction they cannot act on,
/// printed by the subsystem that is supposed to explain itself.
#[cfg(feature = "boundary")]
fn isolated_wiring_hint(binding: &dyn crate::hooks::binding::AgentBinding, port: u16) -> String {
    use crate::hooks::binding::EndpointConvention;
    match binding.boundary_wiring().map(|w| w.endpoint) {
        Some(EndpointConvention::EnvVars { base_url, .. }) => {
            format!("{base_url}=http://127.0.0.1:{port}")
        }
        Some(EndpointConvention::TomlProvider { provider_name, .. }) => format!(
            "a [model_providers.{provider_name}] table with base_url = \"http://127.0.0.1:{port}/v1\""
        ),
        None => "this agent has no request plane".to_string(),
    }
}

/// Remove ONE agent's boundary wiring, and put back whatever it named before us.
///
/// **It carries no guard of its own, and never has.** `owns_wiring_for` is
/// asked at each of its three call sites — the boundary-off startup
/// reconciliation, the graceful teardown, and a failed probe — because those
/// are the sites that know which agents are in play. An unguarded call from an
/// `olbox` daemon strips the boundary wiring out of the developer's real config.
///
/// Idempotent and additive-safe: [`crate::hooks::remove_boundary_config`]
/// reclaims the endpoint only while it still names OUR loopback listener, and
/// strips only OUR install-id header, so a customer's corporate gateway and
/// headers survive untouched.
#[cfg(feature = "boundary")]
fn unwire_boundary_config(config: &Config, binding: &dyn crate::hooks::binding::AgentBinding) {
    let path_label = crate::hooks::boundary_config_path(binding)
        .map(|p| p.display().to_string())
        .unwrap_or_else(|| "the agent config".into());
    match crate::hooks::remove_boundary_config(binding) {
        Ok(()) => tracing::info!(
            agent = binding.agent_type(),
            path = %path_label,
            "agent boundary wiring removed — agents connect to the provider directly"
        ),
        Err(e) => tracing::warn!(
            code = %e.code,
            error = %e.message,
            agent = binding.agent_type(),
            path = %path_label,
            "failed to remove agent boundary wiring — the agent may still point at \
             127.0.0.1:{}, which nothing is listening on",
            config.boundary.port,
        ),
    }
}

/// Unwire every agent this daemon owns the wiring for.
///
/// The startup reconciliation and the graceful teardown both need it, and both
/// used to pass a single agent — which left a second agent's config pointing at
/// a port nothing is listening on, the dangling-endpoint failure the wiring
/// tests exist to prevent. The ownership guard is inside the loop, per agent,
/// so one relocated agent and one machine-global agent get the right answer each.
#[cfg(feature = "boundary")]
fn unwire_every_agent(config: &Config) {
    for agent in crate::hooks::detect_agents() {
        if agent.binding.boundary_wiring().is_none() {
            continue;
        }
        if owns_wiring_for(config, &*agent.binding) {
            unwire_boundary_config(config, &*agent.binding);
        }
    }
}

/// Wait for an OS shutdown signal (SIGTERM/SIGHUP on Unix, Ctrl+C everywhere).
///
/// **SIGHUP is handled deliberately.** Its default disposition is an immediate
/// terminate: closing the terminal that ran `openlatch start --foreground`
/// killed the daemon dead — no drain, no PID-file cleanup, and the next `start`
/// reporting `Cleared stale PID file`. There is no config-reload path in this
/// codebase, so a graceful shutdown is the honest minimal behaviour, and it
/// turns an instant kill into a clean drain that any OS supervisor then
/// restarts. (`spawn_daemon_background` additionally `setsid()`s, so the
/// background daemon has no controlling terminal to be hung up on at all.)
async fn signal_handler() {
    #[cfg(unix)]
    {
        use tokio::signal::unix::{signal, SignalKind};
        let mut sigterm =
            signal(SignalKind::terminate()).expect("failed to register SIGTERM handler");
        let mut sighup = signal(SignalKind::hangup()).expect("failed to register SIGHUP handler");
        tokio::select! {
            _ = tokio::signal::ctrl_c() => {}
            _ = sigterm.recv() => {}
            _ = sighup.recv() => {
                tracing::info!("received SIGHUP — draining (no config-reload path exists)");
            }
        }
    }
    #[cfg(not(unix))]
    {
        tokio::signal::ctrl_c()
            .await
            .expect("failed to register ctrl_c handler");
    }
}

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

    /// A green gate idles at the tick; a shut one backs off and, crucially,
    /// STOPS backing off at the ceiling. A backoff that keeps doubling past its
    /// cap is the kind of bug that only surfaces as a laptop that never re-wires
    /// after a day offline.
    #[cfg(feature = "boundary")]
    #[test]
    fn wiring_delay_backs_off_and_caps() {
        let t = WIRING_TICK;
        assert_eq!(wiring_delay(t, 0), t, "a green gate idles at the tick");
        assert_eq!(
            wiring_delay(t, 1),
            t,
            "the first failure retries at the tick"
        );
        assert_eq!(wiring_delay(t, 2), t * 2);
        assert_eq!(wiring_delay(t, 3), t * 4);
        // 60s * 8 = 480s, past the 300s ceiling.
        assert_eq!(wiring_delay(t, 4), WIRING_BACKOFF_MAX);
        // Still capped — and still finite — far past any plausible streak, so a
        // long offline stretch keeps probing rather than drifting to never.
        assert_eq!(wiring_delay(t, 50), WIRING_BACKOFF_MAX);
        assert_eq!(wiring_delay(t, u32::MAX), WIRING_BACKOFF_MAX);
    }

    // -- the per-agent wiring loop (§4) --------------------------------------

    /// Everything a `wire_agents` test needs, standing up the ONE seam the
    /// production path has: a real boundary listener whose two wire formats
    /// point at two different upstreams.
    ///
    /// `preflight::probe` is called directly by the loop and has no injection
    /// point, so "a stubbed probe result" is not a thing that can be written
    /// here. `anthropic-messages` forwards to a live mock (the probe succeeds)
    /// and `openai-responses` to a closed port (the boundary answers its own
    /// synthetic 502 and the probe fails) — one Ok arm and one Err arm through
    /// the real code path, which is what every test below has to observe.
    #[cfg(feature = "boundary")]
    struct WiringFixture {
        /// The ephemeral boundary port the probes go through.
        port: u16,
        /// Where `OPENLATCH_DIR` points, so `boundary_endpoints`' record lands
        /// in a tempdir and never in the developer's real `~/.openlatch`.
        _openlatch_dir: tempfile::TempDir,
        /// The agents' config directories.
        agent_root: tempfile::TempDir,
        /// Restored on drop, under the one lock for that variable.
        _dir_lock: std::sync::MutexGuard<'static, ()>,
        previous_dir: Option<std::ffi::OsString>,
    }

    #[cfg(feature = "boundary")]
    impl Drop for WiringFixture {
        fn drop(&mut self) {
            match self.previous_dir.take() {
                Some(v) => std::env::set_var("OPENLATCH_DIR", v),
                None => std::env::remove_var("OPENLATCH_DIR"),
            }
        }
    }

    #[cfg(feature = "boundary")]
    impl WiringFixture {
        async fn new() -> Self {
            use crate::boundary::wire_format::WireFormat;
            use crate::boundary::{mock, serve_ephemeral, BoundaryState};

            // `spawn_always_200`, not the one-shot capture: the D-16 test probes
            // TWICE in one tick, and a mock that accepts one connection would
            // fail the second probe on the mock rather than on the code.
            let upstream = mock::spawn_always_200().await;
            let dead = mock::closed_port().await;
            let map: std::collections::BTreeMap<String, String> = [
                (
                    WireFormat::AnthropicMessages.as_str().to_string(),
                    format!("http://127.0.0.1:{upstream}"),
                ),
                (
                    WireFormat::OpenAiResponses.as_str().to_string(),
                    format!("http://127.0.0.1:{dead}"),
                ),
            ]
            .into_iter()
            .collect();
            let state = Arc::new(
                BoundaryState::new(
                    reqwest::Url::parse(&format!("http://127.0.0.1:{upstream}")).unwrap(),
                    0,
                    8,
                    &[],
                )
                .with_upstream_map(map),
            );
            let port = serve_ephemeral(state).await;

            // The env redirect goes LAST, after every `.await` in this
            // function: the lock is a `std::sync::Mutex` and holding one across
            // an await point is a deadlock waiting to happen on a
            // single-threaded runtime — `clippy::await_holding_lock` is a hard
            // error here and it is right.
            //
            // `config::OPENLATCH_DIR_ENV_LOCK` is THE lock for `OPENLATCH_DIR`;
            // the global order for this test binary is
            // `config::OPENLATCH_DIR_ENV_LOCK` -> `claude_code::CONFIG_DIR_ENV_LOCK`
            // -> `staging::HOOK_BIN_ENV_LOCK` -> `codex_cli::CONFIG_DIR_ENV_LOCK`.
            let _dir_lock = crate::config::OPENLATCH_DIR_ENV_LOCK
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let _openlatch_dir = tempfile::tempdir().expect("tempdir");
            let previous_dir = std::env::var_os("OPENLATCH_DIR");
            std::env::set_var("OPENLATCH_DIR", _openlatch_dir.path());

            Self {
                port,
                _openlatch_dir,
                agent_root: tempfile::tempdir().expect("tempdir"),
                _dir_lock,
                previous_dir,
            }
        }

        /// A `Config` that can never reach the developer's own files: the agent
        /// id is pinned (so `ensure_agent_id` never writes a real
        /// `config.toml`) and the ownership guard is opted in on the default
        /// port, which is what a sandbox looks like.
        fn config(&self) -> Config {
            let mut cfg = Config::defaults();
            cfg.agent_id = Some("agt_test".to_string());
            cfg.boundary.own_agent_wiring = Some(true);
            cfg.boundary.port = crate::boundary::default_boundary_port();
            cfg
        }

        /// One agent, its config directory under the fixture's tempdir.
        fn agent(
            &self,
            agent_type: &'static str,
            fmt: crate::boundary::wire_format::WireFormat,
            machine_global: bool,
        ) -> crate::hooks::DetectedAgent {
            use crate::hooks::binding::{BoundaryWiring, EndpointConvention};
            let dir = self.agent_root.path().join(agent_type);
            std::fs::create_dir_all(&dir).expect("agent dir");
            crate::hooks::DetectedAgent {
                kind: crate::hooks::AgentKind::ClaudeCode,
                binding: Arc::new(crate::hooks::binding::test_support::FakeBinding {
                    agent_type,
                    display_name: agent_type,
                    config_dir: dir,
                    boundary_wiring: Some(BoundaryWiring {
                        wire_format: fmt,
                        endpoint: EndpointConvention::EnvVars {
                            base_url: "ANTHROPIC_BASE_URL",
                            headers: "ANTHROPIC_CUSTOM_HEADERS",
                        },
                        install_id_header: "x-openlatch-install-id",
                    }),
                    config_is_machine_global: machine_global,
                    ..Default::default()
                }),
            }
        }

        /// Whether this agent's config file names our listener.
        fn is_written(&self, agent_type: &str) -> bool {
            let path = self
                .agent_root
                .path()
                .join(agent_type)
                .join("settings.json");
            std::fs::read_to_string(path)
                .map(|raw| raw.contains("ANTHROPIC_BASE_URL"))
                .unwrap_or(false)
        }
    }

    /// A Codex probe that fails must not cost Claude Code the request plane it
    /// already had. A blanket unwire on the first failure loses coverage we
    /// have today, which is the regression the per-agent loop exists to
    /// prevent.
    #[cfg(feature = "boundary")]
    #[tokio::test(flavor = "multi_thread")]
    async fn wiring_loop_continues_past_a_failed_agent() {
        use crate::boundary::preflight::{Verdict, WiringState};
        use crate::boundary::wire_format::WireFormat;

        let fx = WiringFixture::new().await;
        let cfg = fx.config();
        let wiring = WiringState::default();
        // Codex FIRST, so the failing agent is the one the loop meets before
        // the healthy one: an implementation that aborts on the first failure
        // passes with the order reversed.
        let agents = vec![
            fx.agent("codex-cli", WireFormat::OpenAiResponses, false),
            fx.agent("claude-code", WireFormat::AnthropicMessages, false),
        ];

        let any_failed = wire_agents(agents, fx.port, false, &cfg, &wiring).await;

        assert!(
            any_failed,
            "the Codex probe failed and the tick must say so"
        );
        assert!(
            wiring.is_wired("claude-code"),
            "the healthy agent is still wired after the failing one"
        );
        assert_eq!(wiring.verdict("claude-code"), Verdict::Ok);
        assert!(fx.is_written("claude-code"));
        assert!(
            !wiring.is_wired("codex-cli"),
            "the failing agent is not wired"
        );
        assert!(matches!(wiring.verdict("codex-cli"), Verdict::Failed(_)));
    }

    /// The write is gated on THAT agent's own round trip. Wiring Codex on the
    /// strength of a Claude probe is a disabled subsystem rendering healthy —
    /// and a boundary that binds a port it cannot forward through takes every
    /// session on the machine down.
    #[cfg(feature = "boundary")]
    #[tokio::test(flavor = "multi_thread")]
    async fn probe_failure_blocks_only_that_agents_write() {
        use crate::boundary::preflight::WiringState;
        use crate::boundary::wire_format::WireFormat;

        let fx = WiringFixture::new().await;
        let cfg = fx.config();
        let wiring = WiringState::default();
        let agents = vec![
            fx.agent("claude-code", WireFormat::AnthropicMessages, false),
            fx.agent("codex-cli", WireFormat::OpenAiResponses, false),
        ];

        wire_agents(agents, fx.port, false, &cfg, &wiring).await;

        assert!(
            fx.is_written("claude-code"),
            "the agent whose probe passed is written"
        );
        assert!(
            !fx.is_written("codex-cli"),
            "the agent whose probe failed must be left untouched — the endpoint is \
             written only after a proven round trip IN THAT AGENT'S FORMAT"
        );
    }

    /// `set_wired(a, false)` in the Err arm is not optional. Drop it and the
    /// loop's own `if wiring.is_wired(a)` line skips that agent on every later
    /// tick: permanently unwired on disk, permanently "wired" in state, never
    /// re-probed — and `boundary_status` still reporting `wired: true`.
    #[cfg(feature = "boundary")]
    #[tokio::test(flavor = "multi_thread")]
    async fn failed_probe_clears_wired_so_the_next_tick_reprobes() {
        use crate::boundary::preflight::{Verdict, WiringState};
        use crate::boundary::wire_format::WireFormat;

        let fx = WiringFixture::new().await;
        let cfg = fx.config();
        let wiring = WiringState::default();

        // Pretend a previous tick had wired it, exactly as a green probe
        // followed by a provider outage leaves it.
        wiring.set_wired("codex-cli", true);
        wiring.set_verdict("codex-cli", Verdict::Ok);

        // Tick one, with `forwarding_broke = true` — the one thing that makes
        // the loop re-probe an agent it believes is wired, and exactly what a
        // provider going away looks like in production.
        let first = wire_agents(
            vec![fx.agent("codex-cli", WireFormat::OpenAiResponses, false)],
            fx.port,
            true,
            &cfg,
            &wiring,
        )
        .await;
        assert!(first, "the probe failed");
        assert!(
            !wiring.is_wired("codex-cli"),
            "a failed probe must clear the in-memory gate as well as the file"
        );

        // Tick two, with `forwarding_broke = false`: the agent must be probed
        // AGAIN rather than skipped. A stale `wired = true` would make the loop
        // `continue` past it, leaving the verdict untouched.
        wiring.set_verdict("codex-cli", Verdict::Pending);
        let second = wire_agents(
            vec![fx.agent("codex-cli", WireFormat::OpenAiResponses, false)],
            fx.port,
            false,
            &cfg,
            &wiring,
        )
        .await;
        assert!(second, "the next tick re-probes it");
        assert!(
            matches!(wiring.verdict("codex-cli"), Verdict::Failed(_)),
            "a re-probed agent gets a fresh verdict; a skipped one keeps Pending"
        );
    }

    /// THE D-16 GATE. `own_agent_wiring = true` on a NON-default port is a
    /// sandbox saying "these agent configs are mine". It is not permission to
    /// write a MACHINE-GLOBAL agent's config — and the process-wide
    /// `owns_agent_wiring()` resolves that question through Claude Code's
    /// resolver only, so a sandbox with a relocated `CLAUDE_CONFIG_DIR` and no
    /// `CODEX_HOME` passes it and the loop writes the developer's real
    /// `~/.codex/config.toml`. Asked per binding, the two agents get different
    /// answers.
    #[cfg(feature = "boundary")]
    #[tokio::test(flavor = "multi_thread")]
    async fn isolated_daemon_never_writes_a_machine_global_agent() {
        use crate::boundary::preflight::{Verdict, WiringState};
        use crate::boundary::wire_format::WireFormat;

        let fx = WiringFixture::new().await;
        let mut cfg = fx.config();
        // The sandbox shape: opted in, but NOT on the default port.
        cfg.boundary.port = crate::boundary::default_boundary_port() + 99;
        let wiring = WiringState::default();

        // Both probe in the format that succeeds, so the only thing separating
        // them is the ownership answer.
        let agents = vec![
            fx.agent("relocated", WireFormat::AnthropicMessages, false),
            fx.agent("machine-global", WireFormat::AnthropicMessages, true),
        ];

        let any_failed = wire_agents(agents, fx.port, false, &cfg, &wiring).await;

        assert!(!any_failed, "both probes pass; only the guard differs");
        assert_eq!(wiring.verdict("relocated"), Verdict::Ok);
        assert_eq!(wiring.verdict("machine-global"), Verdict::Ok);

        assert!(
            fx.is_written("relocated"),
            "an agent whose config this sandbox owns is wired"
        );
        assert!(
            wiring.is_wired("relocated"),
            "and its wiring flag follows the write"
        );
        assert!(
            !fx.is_written("machine-global"),
            "a machine-global agent's config is NOT this instance's to write"
        );
        assert!(
            !wiring.is_wired("machine-global"),
            "and it must not be reported as wired either"
        );
    }

    /// The guard, in isolation and in both directions, so the three-way
    /// precedence cannot be reduced to the port check by accident.
    #[cfg(feature = "boundary")]
    #[test]
    fn owns_wiring_for_asks_the_binding_on_a_non_default_port() {
        use crate::hooks::binding::test_support::FakeBinding;

        let relocated = FakeBinding {
            config_is_machine_global: false,
            ..Default::default()
        };
        let global = FakeBinding {
            config_is_machine_global: true,
            ..Default::default()
        };

        let mut cfg = Config::defaults();
        cfg.boundary.port = crate::boundary::default_boundary_port();

        // Default port: the canonical daemon owns every agent's wiring.
        cfg.boundary.own_agent_wiring = None;
        assert!(owns_wiring_for(&cfg, &relocated));
        assert!(owns_wiring_for(&cfg, &global));

        // An explicit opt-OUT is absolute.
        cfg.boundary.own_agent_wiring = Some(false);
        assert!(!owns_wiring_for(&cfg, &relocated));
        assert!(!owns_wiring_for(&cfg, &global));

        // Non-default port, no opt-in: nobody's wiring is ours.
        cfg.boundary.port = crate::boundary::default_boundary_port() + 99;
        cfg.boundary.own_agent_wiring = None;
        assert!(!owns_wiring_for(&cfg, &relocated));
        assert!(!owns_wiring_for(&cfg, &global));

        // Non-default port WITH the opt-in: per agent, and only per agent.
        cfg.boundary.own_agent_wiring = Some(true);
        assert!(owns_wiring_for(&cfg, &relocated));
        assert!(
            !owns_wiring_for(&cfg, &global),
            "opting in is a decision about your own sandbox, not a way to seize a shared file"
        );
    }

    /// The tick seam must stay a seam: unset means production cadence, and a
    /// value small enough to spin is floored rather than honoured.
    #[cfg(feature = "boundary")]
    #[test]
    fn wiring_tick_defaults_to_production_and_never_busy_loops() {
        assert_eq!(wiring_tick(), WIRING_TICK, "unset must mean the real tick");
        assert_eq!(
            wiring_delay(std::time::Duration::from_millis(50), 0),
            std::time::Duration::from_millis(50)
        );
    }

    #[test]
    fn test_openlatch_marker_detected_in_settings() {
        let with_hooks = r#"{"hooks": {"_openlatch": true, "preToolUse": []}}"#;
        assert!(with_hooks.contains("\"_openlatch\""));

        let without_hooks = r#"{"hooks": {"preToolUse": []}}"#;
        assert!(!without_hooks.contains("\"_openlatch\""));
    }

    #[test]
    fn test_format_uptime_seconds_only() {
        assert_eq!(format_uptime(0), "0s");
        assert_eq!(format_uptime(45), "45s");
        assert_eq!(format_uptime(59), "59s");
    }

    #[test]
    fn test_format_uptime_minutes_and_seconds() {
        assert_eq!(format_uptime(60), "1m0s");
        assert_eq!(format_uptime(192), "3m12s");
        assert_eq!(format_uptime(3599), "59m59s");
    }

    #[test]
    fn test_format_uptime_hours_and_minutes() {
        assert_eq!(format_uptime(3600), "1h0m");
        assert_eq!(format_uptime(8094), "2h14m");
        assert_eq!(format_uptime(7200), "2h0m");
    }

    // -- policy startup (D47) ------------------------------------------------

    mod policy_startup {
        use super::*;
        use crate::core::policy::store::{self, BundleMeta};
        use crate::generated::types::PolicyBundle;

        const BODY: &str = r#"{"schema_version":1,"revision":42,"organization_id":"0192f8a1-4c3b-7e2a-9f10-5d8c3b1a7e42","built_at":"2026-07-21T09:00:00Z","enforcement_enabled":true,"signature":null,"rules":[{"rule_id":"OL-CMD-ENF","kind":"command","match_pattern":"*olcanary-enforce*","action":"deny","mode":"enforce","severity":"high","reason":"Canary enforce"}]}"#;

        /// Write a valid, digest-consistent cache into `base`.
        fn seed(base: &std::path::Path, last_poll_ok_at: Option<&str>) {
            let body = BODY.as_bytes();
            let bundle: PolicyBundle = serde_json::from_slice(body).expect("fixture parses");
            let mut meta = BundleMeta::activated(
                &bundle,
                store::digest_of(body),
                Some("\"sha256:deadbeef\"".to_string()),
            );
            meta.last_poll_ok_at = last_poll_ok_at.map(str::to_string);
            store::store(base, body, &meta).expect("cache writes");
        }

        /// The load is a plain synchronous call that returns an already-populated
        /// handle — no task, no await, nothing the HTTP listener could outrun.
        /// That is what makes "enforcing on the first served request" true on
        /// every restart, including one with the network down.
        #[test]
        fn cached_bundle_is_resident_before_anything_can_serve() {
            let tmp = tempfile::tempdir().expect("tempdir");
            seed(tmp.path(), None);

            let runtime = PolicyRuntime::load_from_disk(tmp.path());

            let guard = runtime.handle.load();
            let bundle = guard.as_ref().as_ref().expect("bundle resident at startup");
            assert_eq!(bundle.revision, 42);
            assert_eq!(bundle.command_rules.len(), 1);
            assert_eq!(bundle.command_rules[0].rule_id, "OL-CMD-ENF");
            assert!(bundle.enforcement_enabled);
        }

        /// A locally edited `bundle.json` no longer hashes to the digest the
        /// meta file records. It must be discarded, not loaded — otherwise a
        /// user deletes the rule that blocks them and the daemon happily runs
        /// the edited version.
        #[test]
        fn tampered_bundle_is_rejected_and_the_daemon_starts_with_no_policy() {
            let tmp = tempfile::tempdir().expect("tempdir");
            seed(tmp.path(), None);
            std::fs::write(
                store::bundle_path(tmp.path()),
                BODY.replace(r#""rules":[{"#, r#""rules":[{"x":1,"#),
            )
            .expect("tamper writes");

            let runtime = PolicyRuntime::load_from_disk(tmp.path());

            assert!(
                runtime.handle.load().is_none(),
                "a tampered bundle must never activate"
            );
        }

        #[test]
        fn no_cache_starts_with_no_policy() {
            let tmp = tempfile::tempdir().expect("tempdir");
            let runtime = PolicyRuntime::load_from_disk(tmp.path());
            assert!(runtime.handle.load().is_none());
            assert_eq!(runtime.last_poll_ok_at.load(Ordering::Relaxed), 0);
        }

        /// A restart must not reset the staleness clock, and a daemon running
        /// disk-bundle-only (no credential provider, so no poller) still has to
        /// report the truth on `/metrics`.
        #[test]
        fn poll_clock_is_seeded_from_the_meta_file() {
            let tmp = tempfile::tempdir().expect("tempdir");
            seed(tmp.path(), Some("2026-07-21T09:00:00Z"));

            let runtime = PolicyRuntime::load_from_disk(tmp.path());

            assert_eq!(
                runtime.last_poll_ok_at.load(Ordering::Relaxed),
                1_784_624_400
            );
            assert!(runtime.last_fetch_ok.load(Ordering::Relaxed));
        }
    }
}