openlatch-client 0.1.18

/// 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";
}

// ---------------------------------------------------------------------------
// 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>,
}

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 if `config.cloud.enabled` is true.
///   When `None`, cloud forwarding is disabled regardless of config.
///
/// # 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
    };

    // 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 only when cloud is enabled AND a credential store is available.
    #[allow(clippy::type_complexity)]
    let (cloud_tx, cloud_state_opt, outbox_opt, cloud_worker_task): (
        _,
        _,
        _,
        Option<tokio::task::JoinHandle<()>>,
    ) = if config.cloud.enabled {
        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();

            // 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();
            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_state = worker_state.clone();
                        let openlatch_dir = openlatch_dir.clone();
                        let outbox = worker_outbox.clone();
                        let shutdown_rx = shutdown_rx.clone();
                        async move {
                            let mut rx = rx.lock_owned().await;
                            crate::cloud::worker::run_cloud_worker_on(
                                &mut rx,
                                provider,
                                cloud_config,
                                worker_state,
                                openlatch_dir,
                                outbox,
                                Some(shutdown_rx),
                            )
                            .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_client = match reqwest::Client::builder()
                .timeout(std::time::Duration::from_millis(
                    config.cloud.timeout_total_ms,
                ))
                .build()
            {
                Ok(c) => Some(c),
                Err(e) => {
                    tracing::warn!(error = %e, "alerts long-poll http client init failed");
                    None
                }
            };
            // 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(),
                            )
                        },
                    );
                }
                (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() {
                match reqwest::Client::builder()
                    .timeout(std::time::Duration::from_millis(
                        config.cloud.timeout_total_ms,
                    ))
                    .build()
                {
                    Ok(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();
                        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(),
                                )
                            },
                        );
                        tracing::info!(
                            target: "policy",
                            poll_interval_secs = config.policy.poll_interval_secs,
                            "policy bundle poller started"
                        );
                    }
                    Err(e) => {
                        // Disk-bundle-only: the resident bundle keeps
                        // enforcing, it just never refreshes.
                        tracing::warn!(
                            target: "policy",
                            code = crate::error::ERR_BUNDLE_FETCH_FAILED,
                            error = %e,
                            "policy poller http client init failed; the resident bundle keeps enforcing but will not refresh"
                        );
                    }
                }
            }

            (Some(tx), Some(cloud_state), outbox, Some(cloud_worker))
        } else {
            tracing::info!("cloud forwarding enabled in config but no credential store provided — cloud forwarding disabled");
            (None, None, None, None)
        }
    } else {
        (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;
        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();
        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.
        let boundary_upstream = config.boundary.upstream_url();
        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(
                        boundary_upstream.clone(),
                        boundary_port,
                        boundary::DEFAULT_INFLIGHT,
                        &boundary_patterns,
                    )
                    .with_measurement(boundary_registry.clone(), boundary_cloud_tx.clone())
                    .with_policy(boundary_policy.clone())
                    .with_wiring(boundary_wiring.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 && config.boundary.owns_agent_wiring() {
            unwire_boundary_config();
        }
        (None, None)
    };

    // Captured before `config` is folded into `AppState`, so the teardown at the
    // bottom of this function can ask the same question without reaching back
    // through the `Arc`.
    #[cfg(feature = "boundary")]
    let owns_agent_wiring = config.boundary.owns_agent_wiring();

    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(),
    });

    // 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: 2s timeout, non-blocking)
    if config.update.check {
        let current = env!("CARGO_PKG_VERSION").to_string();
        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 state_for_update = state_for_update.clone();
                async move {
                    if let Some(latest) = update::check_for_update(&current).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_guard;
    let _poll_handle;
    if let Ok(agent) = crate::hooks::detect_agent() {
        let settings_path = match &agent {
            crate::hooks::DetectedAgent::ClaudeCode { settings_path, .. } => settings_path.clone(),
        };
        let openlatch_dir = crate::config::openlatch_dir();
        let token_file = openlatch_dir.join("daemon.token");

        reconciler::run_startup_reconcile(&settings_path, &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,
                settings_path.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
                }
            },
        );

        _watcher_guard = match watcher::spawn_watcher(&settings_path, reconcile_tx.clone()) {
            Ok(w) => {
                tracing::info!("filesystem watcher active");
                Some(w)
            }
            Err(e) => {
                tracing::warn!(error = %e, "filesystem watcher failed — falling back to poll-only");
                None
            }
        };

        _poll_handle = Some(watcher::spawn_poll_fallback(reconcile_tx));
        tracing::info!("reconciler started (reactive watcher + 30s poll)");
    } else {
        _watcher_guard = None;
        _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.
        if owns_agent_wiring {
            unwire_boundary_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) = reqwest::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).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,
    };

    // 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
// ---------------------------------------------------------------------------

/// The detected agent's `settings.json`, or `None` when no supported agent is
/// installed on this machine.
///
/// Absence is not an error here: a daemon on a host with no agent has nothing to
/// wire and nothing to reconcile. `detect_agent` is a pure lookup, so calling it
/// from the daemon adds no dependency the process did not already have.
#[cfg(feature = "boundary")]
fn agent_settings_path() -> Option<std::path::PathBuf> {
    match crate::hooks::detect_agent() {
        Ok(crate::hooks::DetectedAgent::ClaudeCode { settings_path, .. }) => Some(settings_path),
        Err(e) => {
            tracing::debug!(code = %e.code, "no agent detected — nothing to wire to the boundary");
            None
        }
    }
}

/// 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::preflight::{self, Verdict};
    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 wired = wiring.is_wired();
        let failures = upstream_failures();
        let forwarding_broke = failures > last_failures;
        last_failures = failures;

        // Shut gate → probe until it opens. Open gate → only when live traffic
        // has started failing.
        if !wired || forwarding_broke {
            let upstream = config.boundary.upstream.clone();
            match preflight::probe(port, &upstream, preflight::PREFLIGHT_TIMEOUT).await {
                Ok(()) => {
                    consecutive_failures = 0;
                    wiring.set_verdict(Verdict::Ok);
                    if !wired {
                        wire_boundary_config(&config, port, &wiring);
                    }
                }
                Err(reason) => {
                    consecutive_failures = consecutive_failures.saturating_add(1);
                    if wired {
                        tracing::error!(
                            port,
                            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,
                            reason = %reason,
                            attempt = consecutive_failures,
                            "boundary preflight failed — the agent stays unwired and model calls \
                             go straight to the provider (nothing is captured)"
                        );
                    }
                    wiring.set_verdict(Verdict::Failed(reason));
                    // Unconditional, not just when `wired`: a daemon killed with
                    // SIGKILL leaves its `ANTHROPIC_BASE_URL` behind, and this
                    // process starts with `wired == false` while the file still
                    // points at us. Reconciling here is what makes a failed gate
                    // self-healing instead of merely non-committal.
                    if config.boundary.owns_agent_wiring() {
                        unwire_boundary_config();
                        wiring.set_wired(false);
                    }
                }
            }
        }

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

/// Point the detected agent at the boundary listener, once a preflight probe has
/// proven it can actually forward.
///
/// Call sites: exactly one, [`run_wiring_supervisor`], and only on a 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.
#[cfg(feature = "boundary")]
fn wire_boundary_config(
    config: &Config,
    port: u16,
    wiring: &crate::boundary::preflight::WiringState,
) {
    // An isolated instance (non-default port) binds but does not touch the
    // 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
    // — the two-owner divergence this whole ownership move removed, reappearing
    // as two daemons instead of two commands. Route sessions here explicitly.
    if !config.boundary.owns_agent_wiring() {
        tracing::info!(
            port,
            "isolated boundary instance — {} left untouched; run sessions through this \
             listener with ANTHROPIC_BASE_URL=http://127.0.0.1:{port}",
            agent_settings_path()
                .map(|p| p.display().to_string())
                .unwrap_or_else(|| "the agent config".into()),
        );
        return;
    }
    let Some(settings_path) = agent_settings_path() else {
        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(&settings_path, port, &install_id) {
        Ok(()) => {
            wiring.set_wired(true);
            tracing::info!(
                port,
                path = %settings_path.display(),
                "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,
                path = %settings_path.display(),
                "failed to wire agent to the model boundary — agents will talk to the provider directly"
            );
        }
    }
}

/// Remove the agent's boundary wiring.
///
/// Idempotent and additive-safe: `remove_boundary_config` reclaims
/// `ANTHROPIC_BASE_URL` only when it still points at OUR loopback, and strips
/// only OUR install-id line from `ANTHROPIC_CUSTOM_HEADERS`, so a customer's
/// corporate gateway and headers survive untouched.
#[cfg(feature = "boundary")]
fn unwire_boundary_config() {
    let Some(settings_path) = agent_settings_path() else {
        return;
    };
    match crate::hooks::remove_boundary_config(&settings_path) {
        Ok(()) => tracing::info!(
            path = %settings_path.display(),
            "agent boundary wiring removed — agents connect to the provider directly"
        ),
        Err(e) => tracing::warn!(
            code = %e.code,
            error = %e.message,
            path = %settings_path.display(),
            "failed to remove agent boundary wiring — ANTHROPIC_BASE_URL may point at a dead port"
        ),
    }
}

/// 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 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));
        }
    }
}