openlatch-client 0.6.1

/// 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;
#[cfg(feature = "model-relay")]
pub(crate) mod cline_hub;
/// CLI `stop` after a dead or killed daemon takes capture off Cline.app's hub (06 §5).
#[cfg(feature = "model-relay")]
pub(crate) use cline_hub::release_hub as release_cline_hub_after_death;
pub mod cline_prompt_backfill;
pub mod config_monitor;
pub mod dedup;
#[cfg(feature = "model-relay")]
pub(crate) mod endpoint_wiring;
pub mod fallback_replay;
pub mod handlers;
pub mod hold_queue;
pub mod identity;
pub mod optimize;
pub mod policy_poller;
pub mod reconciler;
pub mod reinforcement;
pub mod session_state;
pub mod spend;
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 MODEL_RELAY: &str = "model_relay";
    pub const MODEL_RELAY_WIRING: &str = "model-relay-wiring";
    pub const CLOUD_WORKER: &str = "cloud-worker";
    pub const OUTBOX_DRAIN: &str = "outbox-drain";
    pub const ALERTS_LONG_POLL: &str = "alerts-long-poll";
    pub const POLICY_POLLER: &str = "policy-poller";
    pub const RECONCILER: &str = "reconciler";
    pub const DEDUP_EVICTOR: &str = "dedup-evictor";
    pub const LOG_CLEANUP: &str = "log-cleanup";
    pub const EVENT_LOG_WRITER: &str = "event-log-writer";
    pub const TAMPER_LOG_WRITER: &str = "tamper-log-writer";
    pub const FALLBACK_REPLAY: &str = "fallback-replay";
    pub const UPDATE_CHECK: &str = "update-check";
    pub const AUTO_UPDATE_WORKER: &str = "auto-update-worker";
    pub const UPDATE_SENTINEL: &str = "update-sentinel";
    pub const EGRESS_MONITOR: &str = "egress-monitor";
    pub const CONFIG_MONITOR: &str = "config-monitor";
}

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

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

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

    fn invalidate(&self) {
        self.store.invalidate();
    }
}

// ---------------------------------------------------------------------------
// 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>,
    /// Typed below-floor state, kept in memory for fresh installs and metrics.
    pub floor: Arc<std::sync::RwLock<Option<ClientFloorState>>>,
    /// Caller-owned Tier-2 register files, bounded independently of bundles.
    pub sessions: Arc<session_state::SessionStateManager>,
    /// Bounded hashed delivery history for tool-dispatch actuators.
    pub dispatch_sessions: Arc<session_state::DispatchStateManager>,
    /// Pending Tier-3 decisions; the hook waits only on this loopback state.
    pub holds: Arc<hold_queue::HoldQueue>,
    /// Parsed reinforcement exchanges, never transcript polling.
    pub reinforcement: Arc<reinforcement::ReinforcementStore>,
}

/// Versions reported when the platform requires this client to upgrade.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClientFloorState {
    pub installed_version: String,
    pub minimum_version: String,
}

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::{project_document, store};

        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 floor = Arc::new(std::sync::RwLock::new(None));
        let resident = cached.and_then(|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);
            }
            if c.meta.state.as_deref() == Some("floor_blocked") {
                if let Some(minimum_version) = c.meta.min_client_version.clone() {
                    if let Ok(mut guard) = floor.write() {
                        *guard = Some(ClientFloorState {
                            installed_version: env!("CARGO_PKG_VERSION").to_string(),
                            minimum_version,
                        });
                    }
                }
            }
            match project_document(c.document) {
                Ok((_, mut resident)) => {
                    resident.digest = Some(c.meta.digest);
                    Some(resident)
                }
                Err(error) => {
                    store::discard(base_dir);
                    tracing::warn!(
                        target: "policy",
                        code = crate::error::ERR_BUNDLE_INVALID,
                        %error,
                        "cached policy bundle projection failed at startup; running with no policy until the next successful fetch"
                    );
                    None
                }
            }
        });

        Self {
            handle: crate::core::policy::new_handle(resident),
            last_fetch_ok,
            last_poll_ok_at,
            floor,
            sessions: Arc::new(session_state::SessionStateManager::default()),
            dispatch_sessions: Arc::new(session_state::DispatchStateManager::default()),
            holds: Arc::new(hold_queue::HoldQueue::default()),
            reinforcement: Arc::new(reinforcement::ReinforcementStore::default()),
        }
    }

    /// 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,
    /// Per-session run marks for backfilling the prompt events Cline Desktop
    /// never sends. See [`cline_prompt_backfill`].
    pub cline_prompts: cline_prompt_backfill::PromptBackfill,
    /// 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>,
    /// The same credential provider handed to the cloud worker. `None` if cloud
    /// forwarding is disabled (no credential store). `POST /admin/auth/refresh`
    /// calls `invalidate()` on it so a daemon that outlives an `openlatch system
    /// auth login` run in another process observes the new credential on its
    /// next poll instead of a memoized stale one (issue #306).
    pub credential_provider: Option<Arc<dyn CredentialProvider>>,
    /// 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>>,
    /// Live stage + byte counts of the running apply, shared with its
    /// pipeline. `GET /admin/update/status` reads it on every poll.
    pub update_progress: Arc<update::ApplyProgress>,
    /// 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>,
    /// The binary an auto-update swapped in, set by the apply pipeline just
    /// before it requests that drain. The serve loop returns it as
    /// [`Served::restart_into`]: acting on it is the foreground runner's job.
    pub restart_into: std::sync::OnceLock<std::path::PathBuf>,
    /// 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>,
    /// Live config-monitor readiness and request channel. Pending startup is
    /// observable, and its sender queues early hook/admin requests until the
    /// watcher and monitor loop are ready.
    pub config_monitor: Arc<config_monitor::ConfigMonitorRuntime>,
    /// 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-relay, D-09). The hook side
    /// (`handlers::process_envelope`) stamps it on `SessionStart` / tool-call
    /// hooks; the model relay listener (spawned in this same function when
    /// `spawn_model_relay`) reads the **same** `Arc` to resolve attribution +
    /// assurance in-process (B-2). Always present; empty until the first hook.
    pub registry: Arc<crate::model_relay::session::SessionRegistry>,
    /// Live state of every supervised in-process subsystem. Read by `/health`
    /// (which reports `degraded` when a `RestartPolicy::Always` task is not
    /// running), `/metrics`, and `openlatch status`. Written only by the
    /// supervisors in `core::supervision::task`.
    pub health: Arc<HealthRegistry>,
    /// Runtime egress health: the resolved route and whether it currently
    /// works. Written by every factory consumer at its own send site, read by
    /// `/admin/egress/status`, `/metrics` and `/health`.
    ///
    /// **Always present**, including on a host that has never authenticated:
    /// egress is not a cloud-credential capability, and a fresh install whose
    /// proxy is wrong needs the answer more than anyone.
    pub egress: crate::egress::EgressState,
}

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

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

/// Start the daemon HTTP server and run until a shutdown signal is received.
///
/// Binds to `127.0.0.1:{config.port}`. The server shuts down gracefully on:
/// - SIGTERM (Unix) or Ctrl+C (all platforms)
/// - HTTP POST /shutdown (authenticated)
///
/// After shutdown, prints a summary to stderr and waits for the event logger to drain.
///
/// # Parameters
///
/// - `credential_store`: optional credential store for cloud forwarding.
///   When `Some`, the cloud worker is spawned. When `None`, there is nothing to
///   authenticate with, so forwarding stays idle — the only way it does not run.
///
/// # Errors
///
/// Returns an error if the TCP listener cannot be bound (e.g., port in use).
pub async fn start_server(
    config: Config,
    token: String,
    credential_store: Option<Arc<dyn crate::auth::CredentialStore>>,
    spawn_model_relay: bool,
) -> anyhow::Result<Served> {
    // SECURITY: Bind to 127.0.0.1 by default.
    // Only bind 0.0.0.0 inside Docker/container environments where the network
    // model relay 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_model_relay,
        /* 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_model_relay: bool,
) -> anyhow::Result<Served> {
    // `reconcile_wiring = false`: this entry point serves an embedded or test
    // daemon that shares the machine with a real one. `spawn_model_relay = 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_model_relay,
        /* reconcile_wiring = */ false,
    )
    .await
}

/// Internal implementation: serve HTTP on the given listener.
///
/// `spawn_model_relay` decides whether the in-process model-relay listener is
/// co-launched here. It is `true` only on the full daemon path (`openlatch start`
/// / supervised / `init --foreground` with the model relay on) and `false` for tests
/// and for an explicit opt-out — the intentional asymmetry that keeps a
/// non-daemon process from binding the pinned model relay port.
///
/// `reconcile_wiring` says whether a `spawn_model_relay = 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_model_relay: bool,
    reconcile_wiring: bool,
) -> anyhow::Result<Served> {
    // Every long-lived task in this process runs under the in-process
    // supervisor: a panic restarts the subsystem instead of silently killing it
    // while `/health` keeps answering `ok`. `health` is the observable side of
    // that (read by `/health`, `/metrics`, `openlatch status`);
    // `tasks_shutdown_tx` is the single teardown broadcast every supervisor and
    // every supervised loop honours, so `openlatch stop` drains them all at once
    // rather than one bespoke channel at a time.
    let health = Arc::new(HealthRegistry::new());
    let (tasks_shutdown_tx, tasks_shutdown_rx) = tokio::sync::watch::channel(false);

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

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

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

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

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

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

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

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

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

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

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

    // `source` -> wire format, seeded ONCE here from `detect_agents()` and
    // shared with both the cloud worker and the wiring supervisor.
    //
    // The worker flushes batches, so a `detect_agents()` call per envelope would
    // stat the filesystem on every event; and `core::cloud` must not call
    // `crate::hooks::`, which would invert the core -> hooks direction. The
    // daemon does all the detecting and hands the answer down.
    //
    // **Seeded, not frozen.** A GUI-hosted agent's wire format is a function of
    // which provider the customer is on, so the wiring supervisor updates this
    // handle wherever it writes a plane's wiring — otherwise every envelope
    // from that agent carries the format of the provider they left until the
    // next daemon restart. Hoisted out of the cloud block for that: the wiring
    // supervisor runs whether or not there is a credential store.
    let source_formats: crate::cloud::worker::SourceFormats = crate::hooks::detect_agents()
        .iter()
        .filter_map(|a| {
            a.binding
                .model_relay_wiring()
                .map(|w| (a.binding.agent_type(), w.wire_format))
        })
        .collect();

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

            // The licensing host key, resolved ONCE for the process. Both
            // consumers below read this value: the cloud worker sends it as
            // `X-OpenLatch-Machine-Id`, and the policy poller sends it on the
            // bundle request. `identity::host_id()` beside it is the `hostid`
            // the single cloud egress fills into envelopes that carry none —
            // deliberately not `host_key`, whose `agent:<agent_id>` fallback is
            // a licensing key and not a machine identifier.
            //
            // Computed here rather than inside `core::cloud` because nothing in
            // `src/core` imports `crate::daemon`: the daemon resolves, core
            // carries.
            //
            // Filtered at this single source because the value is handed straight
            // to `req.header(...)` on the bundle poll. `agent_id` reaches here
            // from `config.toml`, so a hand-edit can put a newline or a
            // non-visible-ASCII byte in it, and the `agent:<agent_id>` fallback
            // carries that through verbatim. Left unfiltered, reqwest rejects the
            // request at build time on EVERY tick: the bundle never activates, the
            // client runs with no policy indefinitely, and the log blames the
            // transport rather than the file. `PolicyPoller::new` already applies
            // this same filter to `agent_id` itself, for this same reason — this
            // closes the second door into that failure.
            let host_key = identity::host_key(config.agent_id.as_deref().unwrap_or_default())
                .filter(|key| {
                    let usable = reqwest::header::HeaderValue::from_str(key).is_ok();
                    if !usable {
                        tracing::warn!(
                            "host key is not a usable header value — this host will not be \
                             identified to the platform; check agent_id in config.toml"
                        );
                    }
                    usable
                });

            // 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,
                host_key: host_key.clone(),
                host_id: identity::host_id(),
                agent_id: config.agent_id.clone(),
            };

            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();
            // Held on `AppState` so `POST /admin/auth/refresh` can invalidate
            // the SAME provider the worker polls — a login noticed anywhere
            // but here would refresh a copy nothing reads.
            let admin_provider = provider.clone();
            let worker_egress =
                crate::egress::EgressReporter::recording(&config.egress, egress_state.clone());

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

            // The daemon-wide `tasks_shutdown_tx` doubles as the cloud worker's
            // explicit flush signal. Channel closure alone is not a reliable
            // trigger: `cloud_tx` lives on the `Arc<AppState>` and is cloned
            // into the config monitor and the tamper reconciler's sinks, so the
            // mpsc may not close promptly on `openlatch stop` — the
            // `Arc::try_unwrap` below already warns about exactly that case.
            //
            // Receiver behind an `Arc<Mutex<_>>` for the same reason as the
            // event-log writer: a panicked run must be restartable onto the
            // same channel, buffered events included.
            let cloud_rx = Arc::new(tokio::sync::Mutex::new(rx));
            let worker_outbox = outbox.clone();
            // The same `Arc` the hook path upserts into and the model relay reads
            // (created above). The worker resolves a session for the producers
            // that cannot name one themselves — the config monitor and the
            // tamper reconciler — so their events land in a session the
            // platform can display instead of in none at all.
            let worker_sessions = registry.clone();
            // The swap handle, not a client: the worker re-reads it per flush, per drain
            // pass and per health probe, so a self-heal pass reaches the highest-volume
            // egress consumer on the host without restarting it.
            let worker_client = egress_clients.cloud.clone();
            // Cloned here, before the cloud worker's own factory closure below
            // takes ownership of `provider`, `cloud_config`, `worker_egress`,
            // `openlatch_dir` and `worker_client` — the outbox-drain task
            // (registered right after the worker) needs the same credential,
            // config, state, directory, formats, sessions and egress handle
            // the worker itself uses, since replay traffic takes the same
            // route as live traffic.
            let drain_provider = provider.clone();
            let drain_config = cloud_config.clone();
            let drain_state = cloud_state.clone();
            let drain_dir = openlatch_dir.clone();
            let drain_formats = source_formats.clone();
            let drain_sessions = registry.clone();
            let drain_egress = worker_egress.clone();
            let drain_client = worker_client.clone();
            let drain_shutdown = tasks_shutdown_rx.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();
                    // Cloned INTO the closure rather than moved: the same handle
                    // is also given to the wiring supervisor, which is what keeps
                    // the map current when an agent's provider changes.
                    let worker_formats = source_formats.clone();
                    move || {
                        let rx = cloud_rx.clone();
                        let provider = provider.clone();
                        let cloud_config = cloud_config.clone();
                        let worker_egress = worker_egress.clone();
                        let worker_state = worker_state.clone();
                        let openlatch_dir = openlatch_dir.clone();
                        let outbox = worker_outbox.clone();
                        let shutdown_rx = shutdown_rx.clone();
                        let worker_client = worker_client.clone();
                        let source_formats = worker_formats.clone();
                        let sessions = worker_sessions.clone();
                        async move {
                            let mut rx = rx.lock_owned().await;
                            crate::cloud::worker::run_cloud_worker_on(
                                &mut rx,
                                provider,
                                cloud_config,
                                worker_egress,
                                worker_state,
                                openlatch_dir,
                                outbox,
                                Some(shutdown_rx),
                                worker_client,
                                source_formats,
                                sessions,
                            )
                            .await
                        }
                    }
                },
            );

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

            // Durable-outbox drain: replays entries the worker spools when a
            // POST fails past its retries. Its own supervised subsystem
            // (issue #306, contract 3) — previously a bare `tokio::spawn`
            // nested inside `run_cloud_worker_on`, so a panic there ended
            // outbox replay for the rest of the daemon's lifetime while the
            // cloud worker itself kept reporting healthy (nothing joined the
            // nested task's handle). `Always`: like the worker whose route
            // and credential it shares, it is meant to run for the daemon's
            // whole lifetime and has no legitimate clean completion.
            if let Some(drain_outbox) = outbox.clone() {
                spawn_supervised(
                    &health,
                    TaskSpec::new(subsystem::OUTBOX_DRAIN, RestartPolicy::Always),
                    tasks_shutdown_rx.clone(),
                    move || {
                        crate::cloud::worker::run_outbox_drain_on(
                            drain_outbox.clone(),
                            drain_client.clone(),
                            drain_config.clone(),
                            drain_provider.clone(),
                            drain_state.clone(),
                            drain_dir.clone(),
                            drain_formats.clone(),
                            drain_sessions.clone(),
                            drain_egress.clone(),
                            drain_shutdown.clone(),
                            #[cfg(test)]
                            None,
                        )
                    },
                );
            }

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

            // Policy bundle poller. Shares the credential provider and cloud
            // state with the worker so token rotation and the auth-error pause
            // stay coordinated — the poller READS `is_auth_error` and never
            // writes it (D46). The resident bundle it publishes was already
            // loaded from disk above; the poller's job is only to keep it
            // fresh.
            if let Some(policy) = policy_runtime.as_ref() {
                // The shared handle, re-read per poll. `current()` answers the same
                // question the per-consumer build used to; a `None` means the one `apply`
                // above already reported why.
                match egress_clients.poller.if_routed() {
                    Some(client) => {
                        let policy_handle = policy.handle.clone();
                        let policy_last_fetch_ok = policy.last_fetch_ok.clone();
                        let policy_last_poll_ok_at = policy.last_poll_ok_at.clone();
                        let policy_floor = policy.floor.clone();
                        let policy_api_url = config.cloud.api_url.clone();
                        let policy_cfg = config.policy.clone();
                        policy
                            .holds
                            .set_source(Arc::new(hold_queue::CloudHoldAnswerSource::new(
                                &policy_api_url,
                                policy_provider.clone(),
                                client.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();
                        // Sent as `X-OpenLatch-Machine-Id` on the bundle
                        // request, the same name and the same value the cloud
                        // worker sends on ingest: which MACHINE is asking.
                        let policy_host_key = host_key.clone();
                        let policy_egress = crate::egress::EgressReporter::recording(
                            &config.egress,
                            egress_state.clone(),
                        );
                        spawn_supervised(
                            &health,
                            TaskSpec::new(subsystem::POLICY_POLLER, RestartPolicy::Always),
                            tasks_shutdown_rx.clone(),
                            move || {
                                policy_poller::run_policy_poller(
                                    policy_handle.clone(),
                                    policy_last_fetch_ok.clone(),
                                    policy_last_poll_ok_at.clone(),
                                    policy_floor.clone(),
                                    policy_state.clone(),
                                    policy_provider.clone(),
                                    policy_api_url.clone(),
                                    policy_cfg.clone(),
                                    policy_dir.clone(),
                                    client.clone(),
                                    policy_agent_id.clone(),
                                    policy_host_key.clone(),
                                    policy_egress.clone(),
                                )
                            },
                        );
                        tracing::info!(
                            target: "policy",
                            poll_interval_secs = config.policy.poll_interval_secs,
                            "policy bundle poller started"
                        );
                    }
                    None => {
                        // Disk-bundle-only: the resident bundle keeps
                        // enforcing, it just never refreshes.
                        tracing::warn!(
                            target: "policy",
                            code = crate::error::ERR_BUNDLE_FETCH_FAILED,
                            "no egress route is permitted, so the policy poller is not started; the resident bundle keeps enforcing but will not refresh"
                        );
                    }
                }
            }

            (
                Some(tx),
                Some(cloud_state),
                outbox,
                Some(cloud_worker),
                Some(admin_provider),
            )
        } else {
            tracing::info!(
                "no credential store provided — cloud forwarding cannot authenticate and is idle"
            );
            (None, 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 state is created now, but its synchronous
    // manifest/watcher initialization starts only after Axum is accepting
    // requests below. The pre-created sender queues hook/admin requests while
    // startup is pending rather than silently dropping them.
    let cache_max = config.inventory_monitor.cache_max_entries.max(64);
    let content_hash_cache = Arc::new(config_monitor::ContentHashCache::new(cache_max));
    let (config_monitor, config_monitor_request_rx, config_monitor_health) =
        if config.inventory_monitor.enabled {
            let (runtime, request_rx) = config_monitor::ConfigMonitorRuntime::pending();
            let task_health = health.register(subsystem::CONFIG_MONITOR, RestartPolicy::Always);
            (runtime, Some(request_rx), Some(task_health))
        } else {
            (config_monitor::ConfigMonitorRuntime::disabled(), None, None)
        };
    if !config.inventory_monitor.enabled {
        tracing::info!("config monitor disabled by config");
    } else {
        tracing::info!("config monitor initialization pending");
    }

    // Model-relay listener (plan 01 forward + plan 02 measurement). A SECOND
    // listener in THIS process, co-launched only on the full-daemon path
    // (`spawn_model_relay`) — NOT on init's background setup daemon or tests, so
    // neither binds the pinned model relay 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 system
    // model-relay 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 model relay 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 model relay 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 `model_relay_wiring` below, which owns it for the daemon's whole
    // life (see `run_wiring_supervisor`).
    #[cfg(feature = "model-relay")]
    #[allow(clippy::type_complexity)]
    let (model_relay_task, wiring_task, endpoint_listeners): (
        Option<tokio::task::JoinHandle<()>>,
        Option<tokio::task::JoinHandle<()>>,
        Option<Arc<crate::model_relay::endpoints::EndpointListeners>>,
    ) = if spawn_model_relay {
        use crate::model_relay;
        use crate::model_relay::wire_format::WireFormat;
        let model_relay_port = config.model_relay.port;

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

        // Outlives every serve attempt on purpose: a model relay restart rebuilds
        // `ModelRelayState`, 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(model_relay::preflight::WiringState::default());

        let model_relay_patterns = config.extra_patterns.clone();
        // D-28: whether an acting `prefix_reorder` (L-0) rule may rewrite the
        // forwarded request. Ships `false`; L-1/L-2 are unaffected by it.
        let model_relay_transforms_act = config.model_relay.transforms_act;
        let model_relay_registry = registry.clone();
        let model_relay_cloud_tx = cloud_tx.clone();
        let model_relay_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 model_relay_policy = policy_runtime.as_ref().map(|p| p.handle.clone());
        let model_relay_policy_sessions = policy_runtime
            .as_ref()
            .map(|p| p.sessions.clone())
            .unwrap_or_else(|| Arc::new(session_state::SessionStateManager::default()));
        let model_relay_wiring = wiring.clone();
        // Resolved once, out here: a per-attempt parse would let a restart
        // silently change where every model call on this host is going.
        // `upstream_for` returns a STRING (it owns the precedence, not the
        // parse), so the parse is explicit here and FALLS BACK rather than
        // failing: this is the one path where a startup error takes every
        // session on the machine down.
        let model_relay_upstream = reqwest::Url::parse(
            &config
                .model_relay
                .upstream_for(WireFormat::AnthropicMessages),
        )
        .unwrap_or_else(|_| model_relay::default_upstream());
        // Resolved ONCE, walking every variant, so the state's lookup can never
        // miss. Without this the per-format map is unreachable and D-04/D-05 are
        // inert no matter how correct the config parsing is.
        let model_relay_upstream_map: std::collections::BTreeMap<String, String> = WireFormat::ALL
            .iter()
            .map(|f| (f.as_str().to_string(), config.model_relay.upstream_for(*f)))
            .collect();
        // I-2 D-12: the formats the operator named. Inside an intercepted tunnel only these
        // override the authority the agent chose; learned upstreams and built-in defaults
        // never do.
        let model_relay_explicit: std::collections::BTreeSet<&'static str> = WireFormat::ALL
            .iter()
            .map(|f| f.as_str())
            .filter(|k| config.model_relay.upstream.contains_key(*k))
            .collect();
        // Cloned into the supervised factory below. The forward CLIENT no longer comes
        // from here — it rides the shared handle just below, so a restart cannot revert
        // the model relay to a route a self-heal pass has already replaced.
        let model_relay_egress = config.egress.clone();
        let model_relay_client = egress_clients.model_relay.clone();
        // Beside the client handle, and for the same reason: the serve factory
        // below runs again on every serve restart, and an upstream the wiring
        // supervisor learned from an agent's own config must survive those.
        // See [`crate::model_relay::UpstreamHandle`].
        let model_relay_upstreams: crate::model_relay::UpstreamHandle =
            Arc::new(std::sync::RwLock::new(std::collections::BTreeMap::new()));
        let model_relay_upstreams_for_serve = model_relay_upstreams.clone();
        let wiring_upstreams = model_relay_upstreams.clone();
        let model_relay_reporter =
            crate::egress::EgressReporter::recording(&config.egress, egress_state.clone());
        // The hook side registers sessions under this id. An intercepted agent's request cannot
        // carry it as a header, exactly as an endpoint agent's cannot, so the listener supplies
        // it for the main listener's tunnels as the factory below does for every endpoint.
        let model_relay_install_id = config.agent_id.clone();

        // The provider-slot endpoints. ONE set for the daemon's life, held out
        // here rather than in any serve factory: a main-listener restart must
        // not drop a port an editor is still dialling. Every endpoint's state is
        // the main listener's in all but the endpoint itself — same client,
        // registry, cloud rail, policy, wiring gate and egress reporter — and
        // they share one set of trackers so the materialisation cap does not
        // multiply by the number of providers.
        let endpoint_listeners = {
            let upstream = model_relay_upstream.clone();
            let patterns = model_relay_patterns.clone();
            let egress = model_relay_egress.clone();
            let registry = model_relay_registry.clone();
            let cloud_tx = model_relay_cloud_tx.clone();
            let policy = model_relay_policy.clone();
            let policy_sessions = model_relay_policy_sessions.clone();
            let wiring = model_relay_wiring.clone();
            let client = model_relay_client.clone();
            let reporter = model_relay_reporter.clone();
            let trackers = model_relay::RelayTrackers::new(model_relay::DEFAULT_INFLIGHT);
            let transforms_act = model_relay_transforms_act;
            let install_id = model_relay_install_id.clone();
            let factory: model_relay::endpoints::StateFactory = Arc::new(move |endpoint| {
                model_relay::ModelRelayState::new_with_egress(
                    upstream.clone(),
                    endpoint.port(),
                    model_relay::DEFAULT_INFLIGHT,
                    &patterns,
                    &egress,
                )
                .with_trackers(&trackers)
                .with_measurement(registry.clone(), cloud_tx.clone())
                .with_transforms_act(transforms_act)
                .with_policy(policy.clone())
                .with_policy_sessions(policy_sessions.clone())
                .with_wiring(wiring.clone())
                .with_client_handle(client.clone())
                .with_egress_reporter(reporter.clone())
                .with_install_id(install_id.clone())
            });
            Arc::new(model_relay::endpoints::EndpointListeners::new(factory))
        };
        let endpoint_listeners_for_serve = endpoint_listeners.clone();
        let endpoint_wiring = Arc::new(endpoint_wiring::EndpointWiring::new(
            endpoint_listeners.clone(),
            wiring.clone(),
            model_relay::endpoints::RelayPorts {
                daemon: config.port,
                main: model_relay_port,
            },
            reconciler::TamperSinks {
                logger: Some(tamper_logger.clone()),
                cloud_tx: cloud_tx.clone(),
                agent_id: config.agent_id.clone().unwrap_or_default(),
                client_version: env!("OPENLATCH_VERSION").to_string(),
            },
            model_relay_registry.clone(),
        ));
        let serve_task = spawn_supervised(
            &health,
            TaskSpec::new(subsystem::MODEL_RELAY, 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(
                    model_relay::ModelRelayState::new_with_egress(
                        model_relay_upstream.clone(),
                        model_relay_port,
                        model_relay::DEFAULT_INFLIGHT,
                        &model_relay_patterns,
                        &model_relay_egress,
                    )
                    .with_upstream_handle(model_relay_upstreams_for_serve.clone())
                    .with_upstream_map(model_relay_upstream_map.clone())
                    .with_measurement(model_relay_registry.clone(), model_relay_cloud_tx.clone())
                    .with_transforms_act(model_relay_transforms_act)
                    .with_policy(model_relay_policy.clone())
                    .with_policy_sessions(model_relay_policy_sessions.clone())
                    .with_wiring(model_relay_wiring.clone())
                    .with_client_handle(model_relay_client.clone())
                    .with_egress_reporter(model_relay_reporter.clone())
                    .with_install_id(model_relay_install_id.clone())
                    .with_explicit_upstreams(model_relay_explicit.clone())
                    .with_endpoints(endpoint_listeners_for_serve.clone())
                    .with_intercept(model_relay_wiring.intercept())
                    .with_refusals(model_relay_wiring.refusals()),
                );
                // The builder above just re-applied config over the shared
                // handle. Re-install what the wiring supervisor learned before
                // this attempt serves a single request, so a serve restart
                // cannot put a customer's local provider back on a cloud
                // default for a whole supervisor tick.
                apply_learned_upstreams(
                    &model_relay_upstreams_for_serve,
                    &model_relay_upstream_map,
                );
                model_relay::serve_attempt(
                    pre_bound.clone(),
                    bstate,
                    model_relay_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_formats = source_formats.clone();
        let wiring_endpoints = endpoint_wiring.clone();
        let wiring_task = spawn_supervised(
            &health,
            TaskSpec::new(subsystem::MODEL_RELAY_WIRING, RestartPolicy::Always),
            tasks_shutdown_rx.clone(),
            move || {
                run_wiring_supervisor(
                    wiring_config.clone(),
                    model_relay_port,
                    wiring_state.clone(),
                    wiring_formats.clone(),
                    wiring_upstreams.clone(),
                    Some(wiring_endpoints.clone()),
                    wiring_shutdown.clone(),
                )
            },
        );

        (
            Some(serve_task),
            Some(wiring_task),
            Some(endpoint_listeners),
        )
    } else {
        // Reconciliation. The model relay is off, so nothing in this process will
        // ever hold 7600 — any `ANTHROPIC_BASE_URL` still on disk is a leftover
        // from a SIGKILLed daemon or a since-flipped config, and it points every
        // agent at a dead port. Clearing it at startup is what makes the
        // invariant self-healing rather than dependent on a clean shutdown.
        if reconcile_wiring {
            unwire_every_agent(&config);
            // Cline.app's hub follows the wiring off the relay (06 §5, D-46).
            let cfg = config.clone();
            let _ = tokio::task::spawn_blocking(move || cline_hub::release_hub(&cfg)).await;
            // And every provider slot a relay endpoint served: with the relay
            // off nothing will ever serve those ports again. Only here — a
            // daemon that stops with the relay ON leaves them wired, because a
            // running editor keeps the URL it read at start either way.
            release_every_provider_endpoint(&config);
        }
        (None, None, None)
    };

    let state = Arc::new(AppState {
        config: Arc::new(config.clone()),
        token,
        dedup: dedup::DedupStore::new(),
        cline_prompts: cline_prompt_backfill::PromptBackfill::default(),
        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,
        credential_provider: credential_provider_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())),
        update_progress: Arc::default(),
        admin_shutdown_request: Arc::new(tokio::sync::Notify::new()),
        restart_into: std::sync::OnceLock::new(),
        last_hook_at_unix_secs: Arc::new(AtomicU64::new(0)),
        hooks_in_flight: Arc::new(AtomicU32::new(0)),
        content_hash_cache,
        config_monitor: config_monitor.clone(),
        pending_alerts,
        policy: policy_runtime,
        registry,
        health: health.clone(),
        egress: egress_state,
    });

    // 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. From here on this process is the daemon: it
    // flushes its own last events at exit instead of handing them off.
    crate::telemetry::mark_served_as_daemon();
    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
    // `fallback_drain_notify` signal fired by the cloud worker.
    //
    // `OnFailure`: the loop never returns on its own, so a completion can only
    // mean "cloud forwarding is disabled, nothing to replay" — restarting that
    // would spin. A panic mid-replay still gets the task back.
    {
        let state_for_replay = state.clone();
        spawn_supervised(
            &health,
            TaskSpec::new(subsystem::FALLBACK_REPLAY, RestartPolicy::OnFailure),
            tasks_shutdown_rx.clone(),
            move || fallback_replay::run(state_for_replay.clone()),
        );
    }

    // UPDT-01: Spawn async update check at startup (T-02-14: bounded, non-blocking)
    if config.update.check {
        let current = env!("CARGO_PKG_VERSION").to_string();
        let update_egress = config.egress.clone();
        let update_registry = config.update.registry_origin.clone();
        let state_for_update = state.clone();
        spawn_supervised(
            &health,
            TaskSpec::new(subsystem::UPDATE_CHECK, RestartPolicy::OnFailure),
            tasks_shutdown_rx.clone(),
            move || {
                let current = current.clone();
                let update_egress = update_egress.clone();
                let update_registry = update_registry.clone();
                let state_for_update = state_for_update.clone();
                async move {
                    if let Some(latest) =
                        update::check_for_update(&current, &update_registry, &update_egress).await
                    {
                        tracing::warn!(code = crate::error::ERR_VERSION_OUTDATED, latest_version = %latest, "Update available: run `openlatch update`");
                        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.
    //
    // The restart dropped the old process's status snapshot, so the one a
    // polling CLI sees is re-derived from the sentinel here, and only this
    // probe turns it `completed` — never the process that has yet to restart.
    //
    // Only a sentinel naming THIS version is reported. One a rollback left
    // behind describes an update that is not running, so the status stays
    // idle, as it does when the rollback cleaned its sentinel up.
    if let Some(sentinel) = update::read_sentinel() {
        let port = config.port;
        if sentinel.to == env!("CARGO_PKG_VERSION") {
            *state.update_status.lock().expect("status mutex poisoned") =
                update::UpdateStatusSnapshot::restarted(&sentinel);
        }
        let update_status = state.update_status.clone();
        spawn_supervised(
            &health,
            TaskSpec::new(subsystem::UPDATE_SENTINEL, RestartPolicy::OnFailure),
            tasks_shutdown_rx.clone(),
            move || {
                let sentinel = sentinel.clone();
                let update_status = update_status.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;
                    let healthy = probe_until_healthy(
                        || probe_self_health(port),
                        &POST_RESTART_PROBE_BACKOFF,
                    )
                    .await;
                    settle_restarted_update(&update_status, &sentinel, healthy);
                    if !healthy {
                        tracing::warn!(target: "update", from = %sentinel.from, to = %sentinel.to, "post-restart /health probe failed; leaving sentinel + .bak in place for restart-loop rollback");
                        return;
                    }
                    tracing::info!(target: "update", from = %sentinel.from, to = %sentinel.to, "post-restart healthz probe succeeded; cleaning up");
                    if let Err(e) = update::cleanup_bak_files() {
                        tracing::warn!(target: "update", error = %e, "cleanup of .bak siblings failed (non-fatal)");
                    }
                    if let Err(e) = update::delete_sentinel() {
                        tracing::warn!(target: "update", error = %e, "delete of update sentinel failed (non-fatal)");
                    }
                    crate::install_state::InstallState::stamp_for_running_binary(env!(
                        "CARGO_PKG_VERSION"
                    ));
                }
            },
        );
    }

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

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

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

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

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

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

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

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

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

    // Hook route — single generic CloudEvents v1.0.2 ingest endpoint. The
    // handler validates Content-Type inline (accepts
    // application/cloudevents+json, application/cloudevents-batch+json, and
    // application/json during the transition) and parses the envelope as
    // either a single object or a JSON array.
    let hook_routes = Router::new()
        .route("/hooks", post(handlers::ingest_cloudevent))
        .route("/holds/{tool_use_id}", get(handlers::wait_for_hold))
        .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();
    let (server_ready_tx, server_ready_rx) = tokio::sync::oneshot::channel();
    let server_task = tokio::spawn(async move {
        // The send occurs in the same poll that immediately enters `serve`.
        // Its receiver cannot resume until Axum has reached its accept wait.
        let _ = server_ready_tx.send(());
        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
    });
    server_ready_rx
        .await
        .map_err(|_| anyhow::anyhow!("daemon server task exited before startup"))?;

    let config_monitor_task = match (config_monitor_request_rx, config_monitor_health) {
        (Some(request_rx), Some(monitor_health)) => {
            let monitor_runtime = config_monitor.clone();
            let runtime_for_init = monitor_runtime.clone();
            let monitor_filter = state.privacy_filter.clone();
            let filter_for_init = monitor_filter.clone();
            let monitor_cache = state.content_hash_cache.clone();
            let monitor_cloud_tx = state.cloud_tx.clone();
            let monitor_event_logger = state.event_logger.clone();
            let monitor_config = state.config.clone();
            let monitor_shutdown = tasks_shutdown_rx.clone();
            Some(tokio::spawn(config_monitor::run_startup(
                monitor_runtime,
                monitor_health,
                monitor_filter,
                monitor_shutdown,
                move || async move {
                    let manifest =
                        tokio::task::spawn_blocking(config_monitor::manifest::load_embedded)
                            .await
                            .map_err(|error| {
                                format!("manifest initialization task failed: {error}")
                            })?
                            .map_err(|error| error.to_string())?;
                    runtime_for_init.mark_manifest_loaded();
                    let request_tx = runtime_for_init.request_tx().ok_or_else(|| {
                        "monitor request channel closed during startup".to_string()
                    })?;
                    config_monitor::ConfigMonitor::new(
                        Arc::new(manifest),
                        monitor_cache,
                        filter_for_init,
                        monitor_cloud_tx,
                        monitor_event_logger,
                        monitor_config,
                    )
                    .spawn_with_requests(request_tx, request_rx)
                    .await
                    .map_err(|error| error.to_string())
                },
                |handle, shutdown| handle.run_until_shutdown(shutdown),
            )))
        }
        _ => None,
    };

    let server_result = match server_task.await {
        Ok(result) => result.map_err(anyhow::Error::from),
        Err(error) => Err(anyhow::anyhow!("daemon server task failed: {error}")),
    };

    // 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, model relay's graceful drain) start
    // draining immediately.
    let _ = tasks_shutdown_tx.send(true);

    if let Some(config_monitor_task) = config_monitor_task {
        if tokio::time::timeout(std::time::Duration::from_secs(5), config_monitor_task)
            .await
            .is_err()
        {
            tracing::warn!("config monitor did not stop within 5s — abandoning it");
        }
    }

    // A serve error after an update has swapped the binary still hands over.
    // The old binary is gone from disk either way, so exiting on the error
    // leaves an unsupervised host with no daemon at all; the new binary gets
    // the post-restart probe and, failing it, the rollback.
    if let Err(error) = server_result {
        if state.restart_into.get().is_none() {
            return Err(error);
        }
        tracing::error!(
            target: "update",
            %error,
            "serve loop failed while draining for an update; handing over regardless"
        );
    }

    // 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 model relay supervisor so the pinned port (7600) is
    // released before the runtime drops. Without this the model relay 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 = "model-relay")]
    if let Some(model_relay_task) = model_relay_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!(
                    "model relay wiring supervisor did not stop within 5s — abandoning it"
                );
            }
        }
        if tokio::time::timeout(std::time::Duration::from_secs(5), model_relay_task)
            .await
            .is_err()
        {
            tracing::warn!("model relay listener did not shut down within 5s — abandoning it");
        }
        // The provider-slot endpoints stop with the relay. Their slots stay
        // written: a running editor keeps the URL it read at start whatever the
        // file says, and the next daemon start serves the same ports again
        // before anything else happens.
        if let Some(listeners) = endpoint_listeners {
            listeners.shutdown_all().await;
        }
        // The other half of the invariant. We are no longer holding the pinned
        // port, so the agent config must stop claiming we are — otherwise every
        // Claude Code session started after this stop dies on ECONNREFUSED,
        // which is precisely the failure this ownership move exists to end.
        //
        // Reached from every graceful teardown: `signal_handler` (ctrl_c /
        // SIGTERM / SIGHUP), `POST /shutdown`, and the auto-update restart —
        // they all funnel through the single `axum::serve` graceful-shutdown
        // above. SIGKILL escapes it by construction; `openlatch stop` and the
        // next daemon start reconcile that case.
        //
        // Isolated instances skip it: they never wrote the file, so removing
        // from it would revoke the canonical daemon's wiring. Asked per agent,
        // inside the loop — a bool captured before the bindings exist cannot
        // answer a per-binding question.
        unwire_every_agent(&state.config);
        // Cline.app's hub loses capture with the rest of the wiring (06 §5, D-46; REJECTED 5).
        let cfg = state.config.clone();
        let _ = tokio::task::spawn_blocking(move || cline_hub::release_hub(&cfg)).await;
    }

    // Capture final stats before releasing state
    let restart_into = state.restart_into.get().cloned();
    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, and there is deliberately no flush here. The
    // daemon's one final flush is `telemetry::finish_on_exit` at the end of
    // `main`, after `command_invoked` is captured: flushing here sent
    // `daemon_stopped` and then lost `command_invoked`, captured a moment later.

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

    if let Some(exe) = &restart_into {
        // A Unix hand-over replaces this process image, so `main` never
        // reaches its final flush: this is the one `update_completed` and
        // `daemon_stopped` get.
        crate::telemetry::flush_global().await;
        tracing::info!(target: "update", exe = %exe.display(), "drained; handing over to the updated binary");
    }
    Ok(Served {
        uptime_secs,
        events,
        restart_into,
    })
}

/// How a serve loop ended, once it has drained.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Served {
    pub uptime_secs: u64,
    pub events: u64,
    /// The binary an auto-update swapped in and asked this daemon to hand
    /// over to. Not acted on here: the serve loop's runtime is still up and
    /// its log writer unflushed. The foreground runner hands over once both
    /// are gone and `daemon.pid` is removed
    /// (`cli::commands::lifecycle::hand_over`).
    pub restart_into: Option<std::path::PathBuf>,
}

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

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

/// Pauses between the post-restart `/health` attempts: five attempts over
/// about 15 s, so one slow first answer does not fail an update for good.
const POST_RESTART_PROBE_BACKOFF: [std::time::Duration; 4] = [
    std::time::Duration::from_secs(1),
    std::time::Duration::from_secs(2),
    std::time::Duration::from_secs(4),
    std::time::Duration::from_secs(8),
];

/// Run `probe` once, then once more after each pause in `backoff`, until it
/// answers healthy.
async fn probe_until_healthy<F, Fut>(mut probe: F, backoff: &[std::time::Duration]) -> bool
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = bool>,
{
    if probe().await {
        return true;
    }
    for pause in backoff {
        tokio::time::sleep(*pause).await;
        if probe().await {
            return true;
        }
    }
    false
}

/// Close the update a restart left in progress: the new binary answered its
/// own `/health`, or it did not and the restart-loop rollback takes over.
///
/// Only the snapshot [`update::UpdateStatusSnapshot::restarted`] seeded for
/// this sentinel is settled. An apply started since has replaced it, and a
/// sentinel for another version never seeded one.
fn settle_restarted_update(
    status: &Mutex<update::UpdateStatusSnapshot>,
    sentinel: &update::UpdateSentinel,
    healthy: bool,
) {
    let mut snap = status.lock().expect("status mutex poisoned");
    let still_restarted = snap.status == update::UpdateStatusKind::InProgress
        && snap.stage == Some(update::ApplyStage::Healthz)
        && snap.started_at.as_deref() == Some(sentinel.applied_at.as_str());
    if !still_restarted {
        return;
    }
    if healthy {
        snap.status = update::UpdateStatusKind::Completed;
        snap.stage = None;
    } else {
        snap.status = update::UpdateStatusKind::Failed;
        snap.error = Some("post-restart /health probe failed".to_string());
    }
    snap.ended_at = Some(crate::install_state::now_rfc3339());
}

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

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

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

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

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

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

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

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

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

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

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

    let result = update::check(current_version, &registry_origin, &state.config.egress).await;

    // `last_check_at` records every probe, whatever it found, so `status`
    // and `doctor` see the worker is alive between releases too.
    let mut install = crate::install_state::InstallState::load_or_default();
    install.record_check();
    if let Err(e) = install.save() {
        tracing::warn!(target: "update", error = %e.message, "failed to write install-state.json");
    }

    let (latest, severity) = match result {
        update::CheckResult::Available {
            latest, severity, ..
        } => (latest, severity),
        update::CheckResult::UpToDate { .. } | update::CheckResult::Failed { .. } => {
            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.
    state.update_progress.begin();
    *state.update_status.lock().expect("status mutex poisoned") =
        update::UpdateStatusSnapshot::in_progress(current_version, &latest);

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

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

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

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

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

/// Whether this instance proves and delivers only the env-trusted `ProxyEnv` deliveries for
/// `binding`: an isolated instance (relocated directories, non-default relay port — the relay's own
/// definition, [`wire_model_relay_config`]) that owns a relocated agent. Its HOME hides the login keychain,
/// so the OS store can never be proven from it; the wrapper and Cline.app's hub carry
/// `NODE_EXTRA_CA_CERTS`, which `preflight::probe_intercept` proves on its own. A VS Code settings
/// key, trusted only through the OS store, is never delivered from one. The machine install is
/// always the full lane: store proof required, a refusal releases everything (D-19).
#[cfg(feature = "model-relay")]
pub(crate) fn env_trust_only(
    config: &Config,
    binding: &dyn crate::hooks::binding::AgentBinding,
) -> bool {
    env_trust_only_from(
        owns_wiring_for(config, binding),
        crate::supervision::owns_machine_supervision(),
        config.model_relay.port == crate::model_relay::default_model_relay_port(),
    )
}

#[cfg(feature = "model-relay")]
fn env_trust_only_from(owns_wiring: bool, owns_machine: bool, on_default_port: bool) -> bool {
    owns_wiring && !owns_machine && !on_default_port
}

/// How often the wiring supervisor comes back around.
#[cfg(feature = "model-relay")]
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 = "model-relay")]
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 = "model-relay")]
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 = "model-relay")]
fn wiring_tick() -> std::time::Duration {
    match std::env::var("OPENLATCH_MODEL_RELAY_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 model relay 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::model_relay::proxy::upstream_failures`] has grown since the last
/// look — real traffic failing is the signal, and a model relay 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 = "model-relay")]
/// Make the live handle's learned upstreams match the records: install every
/// upstream the wiring pass has learned, and put back the configured one for a
/// format whose record is gone.
///
/// The record is written by the writer at the moment it overwrites an agent's
/// base URL ([`crate::hooks::upstream_record_key`]) — the only point at which
/// where that agent's provider actually lives is still on disk. Reading it here
/// is what closes the loop: the listener forwards to the provider it replaced
/// instead of to the wire format's cloud default.
///
/// **Both directions.** This used to only ever insert, so a record forgotten on
/// uninstall kept forwarding that format to the old endpoint until the daemon
/// restarted — a customer who removed their gateway still had their traffic sent
/// to it. `configured` is the per-format map the listener was built from.
///
/// Idempotent and quiet. It runs every tick, and logs only a genuine change, so
/// a steady host says nothing.
fn apply_learned_upstreams(
    upstreams: &crate::model_relay::UpstreamHandle,
    configured: &std::collections::BTreeMap<String, String>,
) {
    for fmt in crate::model_relay::wire_format::WireFormat::ALL {
        let recorded =
            crate::hooks::model_relay_endpoints::peek(&crate::hooks::upstream_record_key(fmt));
        let learned = match recorded {
            Some(Some(recorded)) => match reqwest::Url::parse(&recorded) {
                Ok(url) => Some(url),
                Err(_) => {
                    // The agent's own config held something we cannot forward
                    // to. Leaving the entry alone keeps the format on its
                    // default, which is wrong but reachable — inventing a
                    // destination is not.
                    tracing::warn!(
                        format = fmt.as_str(),
                        value = %recorded,
                        "recorded upstream is not a valid URL — leaving this format on its default"
                    );
                    continue;
                }
            },
            _ => None,
        };
        let mut map = upstreams.write().unwrap_or_else(|e| e.into_inner());
        let target = match learned {
            Some(url) => url,
            None => {
                // No record: the format belongs to the configured map. Only a
                // handle entry that differs from it was learned, and only that
                // one is reverted.
                let Some(url) = configured
                    .get(fmt.as_str())
                    .and_then(|v| reqwest::Url::parse(v).ok())
                else {
                    continue;
                };
                if map.get(fmt.as_str()).is_none_or(|current| *current == url) {
                    continue;
                }
                url
            }
        };
        if map.get(fmt.as_str()) == Some(&target) {
            continue;
        }
        map.insert(fmt.as_str(), target.clone());
        tracing::info!(
            format = fmt.as_str(),
            upstream = %target,
            "model relay upstream for {} now {}",
            fmt.as_str(),
            target
        );
    }
}

/// The wall clock and a monotonic clock together, so [`react_to_trust_loss`] takes one
/// parameter for both rather than two — eight parameters trips clippy's `too_many_arguments`
/// under `-D warnings`. `Copy` so the supervisor can pass it by value each tick.
#[cfg(feature = "model-relay")]
#[derive(Debug, Clone, Copy)]
struct TrustClock {
    wall: time::OffsetDateTime,
    mono: std::time::Instant,
}

/// Agent → every host it intercepts, as [`proxy_env_declarations`] returns it.
#[cfg(feature = "model-relay")]
type ProxyEnvDeclarations =
    std::collections::BTreeMap<&'static str, std::collections::BTreeSet<String>>;

/// [`react_to_trust_loss_in`], computing the declarations itself. `#[cfg(test)]`: the supervisor
/// computes them once per tick and hands the same map to `wire_agents_in`.
#[cfg(all(test, feature = "model-relay"))]
fn react_to_trust_loss(
    config: &Config,
    agents: &[crate::hooks::DetectedAgent],
    ca: Option<&crate::model_relay::ca::CaInfo>,
    at: TrustClock,
    refusals: &crate::model_relay::intercept::RefusalLedger,
    holds: &mut std::collections::BTreeMap<String, std::time::Instant>,
    wiring: &crate::model_relay::preflight::WiringState,
) -> std::collections::BTreeSet<&'static str> {
    let declared = proxy_env_declarations(agents);
    let proxy_env = proxy_env_agents(agents, &declared);
    react_to_trust_loss_in(config, &proxy_env, ca, at, refusals, holds, wiring)
}

/// Each detected agent `declared` names, with its hosts.
#[cfg(feature = "model-relay")]
fn proxy_env_agents<'a>(
    agents: &'a [crate::hooks::DetectedAgent],
    declared: &'a ProxyEnvDeclarations,
) -> Vec<(
    &'a crate::hooks::DetectedAgent,
    &'a std::collections::BTreeSet<String>,
)> {
    agents
        .iter()
        .filter_map(|a| declared.get(a.agent_type()).map(|hosts| (a, hosts)))
        .collect()
}

/// D-31 (7-day release) and D-32 (refusal reaction), once per tick, BEFORE `wire_agents`, so an
/// agent released here is not re-armed by the same tick. Returns the agents to skip this tick.
/// Touches ProxyEnv bindings only — every other convention passes through untouched.
///
/// `proxy_env` pairs each ProxyEnv agent with its hosts from this tick's [`proxy_env_declarations`]
/// (I-4 plan 05 §5.1): static ∪ dynamic, the SAME map the interceptor and `sole_interceptor_for`
/// read, so a refused customer host releases its agent too.
#[cfg(feature = "model-relay")]
fn react_to_trust_loss_in(
    config: &Config,
    proxy_env: &[(
        &crate::hooks::DetectedAgent,
        &std::collections::BTreeSet<String>,
    )],
    ca: Option<&crate::model_relay::ca::CaInfo>,
    at: TrustClock,
    refusals: &crate::model_relay::intercept::RefusalLedger,
    holds: &mut std::collections::BTreeMap<String, std::time::Instant>, // host → when held (diagnostic)
    wiring: &crate::model_relay::preflight::WiringState,
) -> std::collections::BTreeSet<&'static str> {
    use crate::model_relay::{ca as ca_mod, ca_lifecycle as life, preflight::Verdict};

    let (now, clock) = (at.wall, at.mono);
    let mut skip = std::collections::BTreeSet::new();
    if proxy_env.is_empty() {
        return skip;
    }

    // D-31: from CA_RELEASE_DAYS before expiry nothing is pointed at a root about to lapse.
    // A missed rotation degrades to "not captured", never to "broken".
    if let Some(info) = ca {
        if life::days_left(info, now) <= ca_mod::CA_RELEASE_DAYS {
            let reason = life::expiring_reason(info.not_after);
            for (a, _) in proxy_env {
                // Release once, as D-32 does: again only if something re-wired it or the verdict
                // moved off this reason — not one release and one warning per tick for 7 days.
                if wiring.is_wired(a.agent_type())
                    || wiring.verdict(a.agent_type()) != Verdict::Failed(reason.clone())
                {
                    release_for_trust(config, a, reason.clone(), wiring);
                }
                skip.insert(a.agent_type());
            }
            return skip;
        }
    }
    // D-32 (amended 2026-09-23): a tripped host is HELD for the life of this supervisor. Holds
    // never lapse on a timer: a pinning agent passes our own proof, so any timed re-arm flaps.
    // Only a fresh daemon — a restart, which `doctor --fix` performs — starts with an empty map
    // and a fresh ledger, and its first tick's proof decides re-arming.
    let mut fresh = Vec::new();
    for host in refusals.tripped_hosts() {
        if !holds.contains_key(&host) {
            holds.insert(host.clone(), clock);
            fresh.push(host);
        }
    }
    for (a, hosts) in proxy_env {
        let Some(host) = hosts.iter().find(|h| holds.contains_key(h.as_str())) else {
            continue;
        };
        // Release once, on entry, or again if something re-wired it meanwhile.
        if fresh.iter().any(|f| f == host) || wiring.is_wired(a.agent_type()) {
            release_for_trust(config, a, life::refused_reason(host), wiring);
        }
        skip.insert(a.agent_type());
    }
    skip
}

/// The one release shape both reactions share. The ownership guard sits at the CALL SITE, as
/// `wire_agents`' unwire does: an olbox daemon must never touch the real config.
#[cfg(feature = "model-relay")]
fn release_for_trust(
    config: &Config,
    a: &crate::hooks::DetectedAgent,
    reason: String,
    wiring: &crate::model_relay::preflight::WiringState,
) {
    use crate::model_relay::preflight::Verdict;
    if owns_wiring_for(config, &*a.binding) {
        release_proxy_delivery(
            &*a.binding,
            crate::hooks::model_relay_endpoints::ReleasedBy::Wiring,
        );
    }
    tracing::warn!(
        agent = a.agent_type(),
        reason = %reason,
        "model relay released the agent's proxy delivery — its calls go direct"
    );
    wiring.set_wired(a.agent_type(), false);
    wiring.set_verdict(a.agent_type(), Verdict::Failed(reason)); // doctor renders OL-RELAY-CA
}

async fn run_wiring_supervisor(
    config: Arc<Config>,
    port: u16,
    wiring: Arc<crate::model_relay::preflight::WiringState>,
    source_formats: crate::cloud::worker::SourceFormats,
    upstreams: crate::model_relay::UpstreamHandle,
    endpoints: Option<Arc<endpoint_wiring::EndpointWiring>>,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
) {
    use crate::model_relay::proxy::upstream_failures;

    let mut last_failures = upstream_failures();
    let mut consecutive_failures: u32 = 0;
    let tick = wiring_tick();
    // The map every serve attempt is built from, so a format whose learned
    // upstream was forgotten goes back to exactly that.
    let configured_upstreams: std::collections::BTreeMap<String, String> =
        crate::model_relay::wire_format::WireFormat::ALL
            .iter()
            .map(|f| (f.as_str().to_string(), config.model_relay.upstream_for(*f)))
            .collect();
    // A change to a file a provider slot lives in wakes a slot-only pass within
    // seconds, rather than waiting out the tick: a running editor's save puts
    // the old URL back, and a developer setting up a new provider expects it
    // wired before their next editor start. Held in this future, so a
    // supervised restart builds a fresh watch.
    let file_changed = Arc::new(tokio::sync::Notify::new());
    let mut file_trigger: Option<watcher::FileTrigger> = None;
    // D-32's holds: kept across ticks, cleared only by a fresh supervisor (a restart).
    let mut holds: std::collections::BTreeMap<String, std::time::Instant> =
        std::collections::BTreeMap::new();

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

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

        // What the writer recorded when it overwrote an agent's base URL. Done
        // on every tick rather than once: the record is written by the wiring
        // pass itself, so on a FIRST install it does not exist yet when the
        // listener binds, and a start-only read would misroute that whole run.
        apply_learned_upstreams(&upstreams, &configured_upstreams);

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

        // D-31/D-32: react to trust loss BEFORE wiring, so an agent released this tick is not
        // re-armed by the same tick. Reads metadata off `ca.pem` only — never reloads the CA
        // itself (D-06's "read once per process" holds).
        //
        // S6: `react_to_trust_loss` itself returns immediately when no detected agent declares
        // `ProxyEnv` — hoisted here so a host with none skips the `ca.pem` read every tick too,
        // not just the reaction that follows it.
        let any_proxy_env = agents.iter().any(|a| {
            a.binding.model_relay_wiring().is_some_and(|w| {
                matches!(
                    w.endpoint,
                    crate::hooks::binding::EndpointConvention::ProxyEnv { .. }
                )
            })
        });
        let ca_info = if any_proxy_env {
            crate::model_relay::ca::inspect(&crate::model_relay::ca::ca_dir(
                &crate::config::openlatch_dir(),
            ))
        } else {
            None
        };
        let trust_clock = TrustClock {
            wall: time::OffsetDateTime::now_utc(),
            mono: std::time::Instant::now(),
        };
        // Once per tick (file reads: Cline's state files and the endpoint ledger), for both the
        // trust-loss reaction and the publication `wire_agents_in` makes.
        let declared = proxy_env_declarations(&agents);
        let skip = react_to_trust_loss_in(
            &config,
            &proxy_env_agents(&agents, &declared),
            ca_info.as_ref(),
            trust_clock,
            &wiring.refusals(),
            &mut holds,
            &wiring,
        );
        let wired_agents: Vec<crate::hooks::DetectedAgent> = agents
            .iter()
            .filter(|a| !skip.contains(a.agent_type()))
            .cloned()
            .collect();
        let declared: ProxyEnvDeclarations = declared
            .into_iter()
            .filter(|(a, _)| !skip.contains(a))
            .collect();

        let mut any_failed = wire_agents_in(
            wired_agents,
            port,
            forwarding_broke,
            &config,
            &wiring,
            &source_formats,
            declared,
        )
        .await;
        // Every provider slot, on its own endpoint. After the main-port agents:
        // the two never share a port or a record, so the order only decides
        // which log lines come first.
        if let Some(endpoints) = &endpoints {
            any_failed |= endpoints.reconcile(&config, &agents).await;
        }
        // Cline.app's hub (06 §5). Not in `any_failed`: a hub that cannot start is a Cline.app
        // detail, not a network outage, and it must not stretch the backoff.
        cline_hub::hub_tick(&config, port, &wiring, &agents).await;

        // AGAIN, after the writes. The tick-top pass ran before this tick's
        // wiring existed, and it is that wiring which records where an
        // agent's provider actually was. Without this second pass the
        // listener stays on the format's cloud default until the NEXT
        // tick — a minute of a customer's local-provider traffic leaving
        // for `api.openai.com`, on the very first install.
        apply_learned_upstreams(&upstreams, &configured_upstreams);
        // Backoff stays PROCESS-WIDE and counts a tick in which any agent
        // failed. Not hammering a dead network is a host-level concern; a timer
        // per agent would be three timers for one cause.
        if any_failed {
            consecutive_failures = consecutive_failures.saturating_add(1);
        } else {
            consecutive_failures = 0;
        }

        // (Re-)arm the watch whenever the set of files it should cover moved —
        // a lane or a settings directory the agent created since last tick.
        if let Some(endpoints) = &endpoints {
            let wanted: Vec<std::path::PathBuf> = endpoints
                .watch_files(&config, &agents)
                .into_iter()
                .filter(|f| f.parent().is_some_and(std::path::Path::is_dir))
                .collect();
            let armed = file_trigger.as_ref().map(|t| t.watched().to_vec());
            if armed.as_deref() != Some(&wanted[..]) {
                // Setting a watch up blocks — FSEvents restarts its stream for
                // every directory added, a third of a second each — and so
                // does tearing one down. Neither may run on a runtime worker:
                // a task woken from that worker (the tamper log's writer, a
                // relay connection) waits in its queue until the call returns.
                let previous = file_trigger.take();
                let notify = file_changed.clone();
                let armed = tokio::task::spawn_blocking(move || {
                    drop(previous);
                    watcher::spawn_file_trigger(&wanted, notify)
                })
                .await;
                file_trigger = match armed {
                    Ok(Ok(trigger)) => trigger,
                    Ok(Err(e)) => {
                        tracing::warn!(
                            error = %e,
                            "cannot watch provider settings files — changes are picked up on the next tick"
                        );
                        None
                    }
                    Err(e) => {
                        tracing::warn!(
                            error = %e,
                            "arming the provider settings watch failed — changes are picked up on the next tick"
                        );
                        None
                    }
                };
            }
        }

        let delay = wiring_delay(tick, consecutive_failures);
        let tick_done = tokio::time::sleep(delay);
        tokio::pin!(tick_done);
        loop {
            tokio::select! {
                _ = &mut tick_done => break,
                _ = shutdown.changed() => return,
                _ = file_changed.notified(), if file_trigger.is_some() => {
                    // Let a save burst settle, then run the slots, and republish the intercept hosts.
                    tokio::select! {
                        _ = tokio::time::sleep(FILE_QUIET_PERIOD) => {}
                        _ = shutdown.changed() => return,
                    }
                    if let Some(endpoints) = &endpoints {
                        let agents = crate::hooks::detect_agents();
                        // I-4 D-40 (amended): a new customer host joins the interceptor NOW. Only
                        // agents the last tick kept (04's trust-loss filter) are re-published.
                        let kept: Vec<_> = agents
                            .iter()
                            .filter(|a| wiring.declares_interceptor(a.agent_type()))
                            .cloned()
                            .collect();
                        publish_intercept_hosts(&wiring, &kept);
                        endpoints.reconcile(&config, &agents).await;
                    }
                }
            }
        }
    }
}

/// How long the wiring loop waits after a provider settings file changes before
/// it reads the file. An editor's save is a burst — a temp file, a rename, and
/// often a second save half a second later — and one pass over the settled
/// bytes is the goal.
#[cfg(feature = "model-relay")]
const FILE_QUIET_PERIOD: std::time::Duration = std::time::Duration::from_secs(2);

/// [`wire_agents_in`], computing the declarations itself. `#[cfg(test)]`: the supervisor computes
/// them once per tick and shares them with [`react_to_trust_loss_in`].
#[cfg(all(test, feature = "model-relay"))]
async fn wire_agents(
    agents: Vec<crate::hooks::DetectedAgent>,
    port: u16,
    forwarding_broke: bool,
    config: &Config,
    wiring: &crate::model_relay::preflight::WiringState,
    source_formats: &crate::cloud::worker::SourceFormats,
) -> bool {
    let declared = proxy_env_declarations(&agents);
    wire_agents_in(
        agents,
        port,
        forwarding_broke,
        config,
        wiring,
        source_formats,
        declared,
    )
    .await
}

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

    let interceptor = publish_declarations(wiring, declared);

    let mut any_failed = false;
    for agent in agents {
        let a = agent.agent_type();
        let plane = agent.binding.model_relay_wiring();

        // No request plane at all — nothing to probe and nothing to write.
        // Not an error: it is a question that does not apply to this agent. An
        // agent with provider slots is wired per slot instead
        // ([`endpoint_wiring`]).
        let Some(w) = plane else {
            wiring.set_wired(a, false);
            continue;
        };
        // Shut gate → probe until it opens. Open gate → only when live traffic
        // has started failing. Per agent, because one agent's open gate says
        // nothing about another's.
        if wiring.is_wired(a) && !forwarding_broke {
            continue;
        }

        // The probe speaks THIS agent's format against THIS format's upstream.
        // Probing as Anthropic and then wiring Codex would prove a Claude round
        // trip and open the gate for a plane nobody checked — a disabled
        // subsystem rendering healthy, which is the failure the gate exists to
        // stop.
        let upstream = config.model_relay.upstream_for(w.wire_format);
        let verdict = match &w.endpoint {
            crate::hooks::binding::EndpointConvention::ProxyEnv {
                intercept_hosts, ..
            } => {
                prove_proxy_env(
                    config,
                    port,
                    w.wire_format,
                    &upstream,
                    intercept_hosts,
                    interceptor.clone(),
                    !env_trust_only(config, &*agent.binding),
                )
                .await
            }
            _ => {
                preflight::probe(port, w.wire_format, &upstream, preflight::PREFLIGHT_TIMEOUT).await
            }
        };
        match verdict {
            Ok(()) => {
                wiring.set_verdict(a, Verdict::Ok);
                // ONLY when this agent is not already wired. The loop re-probes
                // a wired agent on every `forwarding_broke` tick on purpose;
                // writing on every green re-probe would rewrite the customer's
                // config file and re-emit the wiring log lines on every
                // upstream hiccup.
                if !wiring.is_wired(a) {
                    // THE DAEMON WRAPPER, never `hooks::write_model_relay_config`:
                    // the wrapper carries the isolated-instance guard and the
                    // install-id resolution, and skipping it is how an `olbox`
                    // daemon starts writing the developer's real agent config.
                    wire_model_relay_config(config, &*agent.binding, port, wiring, source_formats);
                }
            }
            Err(reason) => {
                // The backoff is for a dead network (WIRING_BACKOFF_MAX's doc). A ProxyEnv verdict
                // carrying CA_REASON_MARKER is local — the store, the CA, the relay's own chain — and
                // no probing cadence changes it (I-4, plan 05 §5.4). Every other failure is unchanged.
                let ca_refusal =
                    w.endpoint.is_proxy_env() && reason.starts_with(preflight::CA_REASON_MARKER);
                if !ca_refusal {
                    any_failed = true;
                }
                // These two lines are the ONLY operator-visible signal that a
                // request plane was torn off disk. `agent` is on both, because
                // on a two-agent host "which one" is the whole question.
                if wiring.is_wired(a) {
                    tracing::error!(
                        port,
                        agent = a,
                        reason = %reason,
                        "model_relay preflight failed on a wired listener — removing the agent \
                         wiring so sessions fall back to a direct provider connection"
                    );
                } else {
                    tracing::warn!(
                        port,
                        agent = a,
                        reason = %reason,
                        "model_relay preflight failed — the agent stays unwired and model calls \
                         go straight to the provider (nothing is captured)"
                    );
                }
                // The old `attempt = consecutive_failures` field is gone: the
                // counter is the supervisor's, and this tick may carry more
                // than one agent's failure.
                wiring.set_verdict(a, Verdict::Failed(reason));
                // Per-agent unwire, replacing a blanket one. The self-healing
                // reason still holds — a SIGKILLed daemon leaves its wiring
                // behind and this process starts with `is_wired` false while
                // the file still points at us — it is just scoped to the agent
                // whose probe failed.
                //
                // The guard is at the CALL SITE: `unwire_model_relay_config` has
                // none of its own, and an unguarded unwire in an `olbox` daemon
                // strips the model relay wiring out of the developer's real config.
                if owns_wiring_for(config, &*agent.binding) {
                    // I-2 D-26: a ProxyEnv agent's release path is the settings-key /
                    // env-file / marker teardown, never the EnvVars/TomlProvider
                    // config rewrite — recorded as the wiring pass's own release, not
                    // a teardown.
                    if matches!(
                        w.endpoint,
                        crate::hooks::binding::EndpointConvention::ProxyEnv { .. }
                    ) {
                        release_proxy_delivery(
                            &*agent.binding,
                            crate::hooks::model_relay_endpoints::ReleasedBy::Wiring,
                        );
                    } else {
                        unwire_model_relay_config(config, &*agent.binding);
                    }
                    // Both statements, always. Drop this one and a failed probe
                    // unwires the file while leaving the in-memory gate open, so
                    // the loop's own `if wiring.is_wired(a)` line skips the
                    // agent on every later tick: permanently unwired on disk,
                    // permanently "wired" in state, never re-probed.
                    wiring.set_wired(a, false);
                    // And the slot with it: the entry is no longer ours, so a
                    // later tick comparing against it would be comparing
                    // against a wiring that is not on disk.
                }
            }
        }
    }
    any_failed
}

/// D-40 (amended) / D-41: agent → every host it intercepts, for every DETECTED ProxyEnv binding:
/// its static `intercept_hosts` ∪ its `provider_endpoints().intercepted_hosts()` (the non-loopback
/// hosts in its own base-URL keys), lowercased.
#[cfg(feature = "model-relay")]
fn proxy_env_declarations(agents: &[crate::hooks::DetectedAgent]) -> ProxyEnvDeclarations {
    agents
        .iter()
        .filter_map(|a| match a.binding.model_relay_wiring()?.endpoint {
            crate::hooks::binding::EndpointConvention::ProxyEnv {
                intercept_hosts, ..
            } => {
                let mut hosts: std::collections::BTreeSet<String> = intercept_hosts
                    .iter()
                    .map(|h| h.to_ascii_lowercase())
                    .collect();
                if let Some(ep) = a.binding.provider_endpoints() {
                    hosts.extend(ep.intercepted_hosts());
                }
                Some((a.agent_type(), hosts))
            }
            _ => None,
        })
        .collect()
}

/// 02's `proxy_env_hosts`, re-bodied (its signature and doc line are unchanged): the union of
/// `proxy_env_declarations`' values. 02's `the_interceptor_is_built_only_for_proxy_env` still holds.
/// Its fakes declare no `provider_endpoints`, so their dynamic half is empty.
///
/// `#[cfg(test)]`: `wire_agents_in` publishes through [`publish_declarations`], from the one
/// [`proxy_env_declarations`] map the supervisor computes per tick, so the interceptor
/// and the attribution map can never disagree (I-4 plan 05 §5.1). Only tests still call this name.
#[cfg(all(test, feature = "model-relay"))]
fn proxy_env_hosts(agents: &[crate::hooks::DetectedAgent]) -> std::collections::BTreeSet<String> {
    proxy_env_declarations(agents)
        .into_values()
        .flatten()
        .collect()
}

/// One publication of the host set: fill or refresh the interceptor (02's `refresh_interceptor`,
/// unchanged) and replace the attribution declarations, from the SAME computation.
#[cfg(feature = "model-relay")]
fn publish_intercept_hosts(
    wiring: &crate::model_relay::preflight::WiringState,
    agents: &[crate::hooks::DetectedAgent],
) -> Option<Arc<crate::model_relay::ca::Interceptor>> {
    publish_declarations(wiring, proxy_env_declarations(agents))
}

/// [`publish_intercept_hosts`] for declarations already computed.
#[cfg(feature = "model-relay")]
fn publish_declarations(
    wiring: &crate::model_relay::preflight::WiringState,
    declared: ProxyEnvDeclarations,
) -> Option<Arc<crate::model_relay::ca::Interceptor>> {
    let ic = refresh_interceptor(
        &wiring.intercept(),
        declared.values().flatten().cloned().collect(),
    );
    wiring.set_interceptors(declared);
    ic
}

/// Fill the slot lazily on the first tick that has a host (only-when-needed: no `ProxyEnv`
/// binding → no CA is ever generated), and refresh the host set every tick after — an agent
/// installed after the daemon started gets its leaf on the next tick. An empty set on a filled
/// slot calls `set_hosts(empty)`: the CA stays in memory (D-06), nothing is intercepted.
#[cfg(feature = "model-relay")]
fn refresh_interceptor(
    slot: &crate::model_relay::InterceptSlot,
    hosts: std::collections::BTreeSet<String>,
) -> Option<Arc<crate::model_relay::ca::Interceptor>> {
    let filled = slot.read().map(|g| g.clone()).unwrap_or(None);
    match filled {
        Some(ic) => {
            if let Err(e) = ic.set_hosts(hosts) {
                tracing::warn!(code = %e.code, error = %e.message,
                    "could not refresh the model relay's intercepted host set");
            }
            Some(ic)
        }
        None if hosts.is_empty() => None,
        None => {
            let dir = crate::model_relay::ca::ca_dir(&crate::config::openlatch_dir());
            match crate::model_relay::ca::Interceptor::new(&dir, hosts) {
                Ok(ic) => {
                    let ic = Arc::new(ic);
                    match slot.write() {
                        Ok(mut g) => *g = Some(ic.clone()),
                        Err(poisoned) => *poisoned.into_inner() = Some(ic.clone()),
                    }
                    Some(ic)
                }
                Err(e) => {
                    tracing::error!(code = %e.code, error = %e.message,
                        "the model relay could not create its certificate authority — \
                         ProxyEnv agents stay unwired");
                    None
                }
            }
        }
    }
}

/// D-09r: every gate, in order; a proxy value is written only when all pass.
#[cfg(feature = "model-relay")]
async fn prove_proxy_env(
    config: &Config,
    port: u16,
    fmt: crate::model_relay::wire_format::WireFormat,
    upstream: &str,
    hosts: &[&str],
    ic: Option<Arc<crate::model_relay::ca::Interceptor>>,
    store_required: bool,
) -> Result<(), String> {
    use crate::model_relay::preflight;

    let Some(host) = hosts.first().copied() else {
        return Err("ProxyEnv declares no intercept host".to_string());
    };
    let Some(ic) = ic else {
        return Err(format!(
            "{}the relay has no certificate authority — see the daemon log",
            preflight::CA_REASON_MARKER
        ));
    };
    let dir = crate::model_relay::ca::ca_dir(&crate::config::openlatch_dir());
    let Some(info) = crate::model_relay::ca::inspect(&dir) else {
        return Err(format!(
            "{}{} is missing or unreadable",
            preflight::CA_REASON_MARKER,
            crate::model_relay::ca::ca_pem_path(&dir).display()
        ));
    };
    if ca_too_close_to_expiry(info.not_after, time::OffsetDateTime::now_utc()) {
        let reason = format!(
            "{}{}{}",
            preflight::CA_REASON_MARKER,
            preflight::CA_EXPIRING_PREFIX,
            info.not_after.date()
        );
        tracing::warn!(code = crate::error::ERR_MODEL_RELAY_CA_UNTRUSTED, host, reason = %reason,
            "the model relay's certificate authority is close to expiry");
        return Err(reason);
    }
    // I-4 (plan 05 §5.3): the store half runs BEFORE the network probe. This unit ships the first
    // real `ProxyEnv` binding, so every `tests/` daemon that detects it now runs this on every
    // tick, where `trust_store::store()` is the REAL platform arm (no `cfg(test)` there) — probing
    // first would tunnel to the real host on every such tick. A store that does not trust the CA
    // fails LOCALLY, with no network round trip, and every gate below still blocks every write.
    //
    // An env-trust-only instance ([`env_trust_only`]) skips it: nothing it delivers reads the OS
    // store, and the probe below proves the `NODE_EXTRA_CA_CERTS` chain its deliveries carry.
    if store_required {
        let store = crate::model_relay::trust_store::store();
        let host_owned = host.to_string();
        let ic_for_blocking = ic.clone();
        let store_result = tokio::task::spawn_blocking(move || {
            preflight::prove_store(&*store, &ic_for_blocking, &host_owned)
        })
        .await
        .map_err(|e| {
            format!(
                "{}the trust-store check did not complete: {e}",
                preflight::CA_REASON_MARKER
            )
        })?;
        if let Err(e) = store_result {
            tracing::warn!(code = crate::error::ERR_MODEL_RELAY_CA_UNTRUSTED, host, reason = %e,
                "the user trust store did not accept the relay's certificate authority");
            return Err(e);
        }
    } else {
        static LOGGED: std::sync::Once = std::sync::Once::new();
        LOGGED.call_once(|| {
            tracing::info!(host, "isolated instance — env-trusted deliveries only");
        });
    }
    let ca_pem = crate::model_relay::ca::ca_pem_path(&dir);
    if let Err(e) = preflight::probe_intercept(
        config,
        port,
        fmt,
        upstream,
        &ca_pem,
        host,
        preflight::PREFLIGHT_TIMEOUT,
    )
    .await
    {
        if e.starts_with(preflight::CA_REASON_MARKER) {
            tracing::warn!(code = crate::error::ERR_MODEL_RELAY_CA_UNTRUSTED, host, reason = %e,
                "the model relay's intercepted tunnel could not be proven");
        }
        return Err(e);
    }
    if store_required && !crate::model_relay::trust_store::platform_proven() {
        tracing::warn!(
            code = crate::error::ERR_MODEL_RELAY_TRUST_UNPROVEN,
            host,
            "user trust store verified on a platform whose trust path is not yet proven end to end"
        );
    }
    Ok(())
}

/// Pure, so it is tested without minting a CA with a custom validity. 02-32: the same boundary
/// as plan 04's `days_left(..) <= CA_RELEASE_DAYS`: one 7-day decision.
#[cfg(feature = "model-relay")]
fn ca_too_close_to_expiry(not_after: time::OffsetDateTime, now: time::OffsetDateTime) -> bool {
    (not_after - now).whole_days() <= crate::model_relay::ca::CA_RELEASE_DAYS
}

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

    // An isolated instance (non-default port) binds but does not touch a
    // machine-global agent config: that file has one owner, the daemon on the
    // default port, and a second writer would take the wiring out from under
    // it. Asked PER BINDING — one relocated agent beside one machine-global
    // agent on the same daemon get different answers.
    if !owns_wiring_for(config, binding) {
        tracing::info!(
            port,
            agent = binding.agent_type(),
            "isolated model_relay instance — {path_label} left untouched; route sessions through \
             this listener explicitly ({})",
            crate::hooks::isolated_wiring_hint(binding, port),
        );
        return;
    }
    // PII-free per-install id (F-22) — reuses the existing `agent_id`,
    // provisioning one if a pre-`init` config lacks it.
    let install_id = match config.agent_id.clone() {
        Some(id) => id,
        None => crate::config::ensure_agent_id(&crate::config::openlatch_dir().join("config.toml"))
            .unwrap_or_default(),
    };
    match crate::hooks::write_model_relay_config_trusting(
        binding,
        port,
        &install_id,
        !env_trust_only(config, binding),
    ) {
        Ok(()) => {
            wiring.set_wired(binding.agent_type(), true);
            // The protocol we just configured this agent to speak. Recorded
            // here, at the write, so it can never claim a format the file does
            // not actually name — it is what lets an unattributable request be
            // traced back to the sole agent on this install speaking its format
            // instead of being reported as `unknown`.
            if let Some(w) = binding.model_relay_wiring() {
                if w.endpoint.is_proxy_env() {
                    // I-4 D-41/D-42: an intercepted agent speaks every format its providers speak, so
                    // it has no single one to record. It is attributed by CONNECT host instead
                    // (`WiringState::sole_interceptor_for`). Recording `w.wire_format` here would make
                    // it a second speaker in `sole_wired_agent_for`, and every unattributed Claude Code
                    // turn on the main port would become `unknown`.
                    // D-42: a NEW wiring starts unproven. `wire_agents` calls this only for an agent
                    // that is not wired (a first write, or a re-establishment after a release), so
                    // every call here is a new wiring. An unchanged tick never gets here, and keeps
                    // the proof.
                    wiring.clear_intercept_proof(binding.agent_type());
                } else {
                    wiring.set_wired_format(binding.agent_type(), w.wire_format);
                    // The SAME fact, at the SAME moment, to the egress worker's map.
                    // It is stamped on hook, config AND economics envelopes, so a
                    // map that only learned this at daemon start labels every event
                    // a switched-over agent produces with the format of the provider
                    // it left. Set here rather than in the worker because this is
                    // the one place the daemon knows the format changed, and
                    // `core::cloud` may not call `crate::hooks::` to find out.
                    source_formats.set(binding.agent_type(), w.wire_format);
                }
            }
            tracing::info!(
                port,
                agent = binding.agent_type(),
                path = %path_label,
                "agent wired to the model relay — model calls route via http://127.0.0.1:{port}"
            );
            // I-2 D-27: the shell-profile line is MACHINE-GLOBAL (it edits the developer's own
            // rc files), so it needs the second predicate as well as the `owns_wiring_for` that
            // passed above.
            if let Some(crate::hooks::binding::EndpointConvention::ProxyEnv { delivery, .. }) =
                binding.model_relay_wiring().map(|w| w.endpoint)
            {
                if profile_line_wanted(delivery, crate::supervision::owns_machine_supervision()) {
                    if let Some(home) = dirs::home_dir() {
                        let ol = crate::config::openlatch_dir();
                        let documents = if cfg!(windows) {
                            dirs::document_dir()
                        } else {
                            None
                        };
                        match crate::hooks::shell_profile::ensure_line(
                            &home,
                            documents.as_deref(),
                            &crate::hooks::proxy_env_file::env_sh_path(&ol),
                            &crate::hooks::proxy_env_file::env_ps1_path(&ol),
                        ) {
                            Ok(files) => tracing::info!(
                                agent = binding.agent_type(),
                                files = ?files,
                                "shell profile sources the model relay wrapper — open a new \
                                 shell to pick it up"
                            ),
                            Err(e) => tracing::warn!(
                                code = %e.code,
                                error = %e.message,
                                "could not update the shell profile — shell-launched sessions \
                                 are not captured"
                            ),
                        }
                    }
                }
            }
            // P7: surfaced where the write actually happens, so the note can
            // never outlive the thing it describes.
            //
            // Gated on the convention, not on the agent name: Remote Control is
            // disabled by `ANTHROPIC_BASE_URL` being set, and only the `EnvVars`
            // arm sets it. Emitting this after wiring Codex named a feature that
            // agent does not have, in the log the operator reads to find out
            // what we just did to their machine.
            //
            // A `ProxyEnv` agent has no such note; this stays `EnvVars`-only on purpose (I-2 D-18).
            if matches!(
                binding.model_relay_wiring().map(|w| w.endpoint),
                Some(crate::hooks::binding::EndpointConvention::EnvVars { .. })
            ) {
                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) => {
            // I-2 D-33/D-34: every delivery skipped for a settings cause. The message starts
            // with `SETTINGS_REASON_MARKER` (`hooks::write_model_relay_config`'s `ProxyEnv`
            // arm); as the verdict it reaches doctor as `PreflightFailed`.
            if e.code == crate::error::ERR_MODEL_RELAY_SETTINGS {
                wiring.set_verdict(
                    binding.agent_type(),
                    crate::model_relay::preflight::Verdict::Failed(e.message.clone()),
                );
            }
            // Adversarial review: a `ProxyEnv` binding's deliveries are written one at a
            // time (`hooks::write_model_relay_config`'s `ProxyEnv` arm), and an earlier one
            // can already have written a settings key or the env-file wrapper before a
            // LATER delivery hard-errors and unwinds the whole call with `?`. This agent is
            // never marked wired below, so nothing would otherwise take that value back —
            // it would stay pointed at 127.0.0.1 once the daemon stops. `owns_wiring_for`
            // already gated this call (the early return above this match), so releasing
            // here is safe.
            if matches!(
                binding.model_relay_wiring().map(|w| w.endpoint),
                Some(crate::hooks::binding::EndpointConvention::ProxyEnv { .. })
            ) {
                release_proxy_delivery(
                    binding,
                    crate::hooks::model_relay_endpoints::ReleasedBy::Wiring,
                );
            }
            tracing::error!(
                code = %e.code,
                error = %e.message,
                agent = binding.agent_type(),
                path = %path_label,
                "failed to wire agent to the model relay — agents will talk to the provider directly"
            );
        }
    }
}

/// Pure: the profile line is written only by the machine's own install, and only for a binding
/// that has a wrapper to source. Extracted so the `true` branch is testable — every in-crate
/// fixture relocates `OPENLATCH_DIR`, which makes `owns_machine_supervision()` false.
#[cfg(feature = "model-relay")]
fn profile_line_wanted(
    delivery: &[crate::hooks::binding::ProxyDelivery],
    owns_machine: bool,
) -> bool {
    owns_machine
        && delivery
            .iter()
            .any(|d| matches!(d, crate::hooks::binding::ProxyDelivery::EnvFile { .. }))
}

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

/// The single release path for D-26, D-31 and D-32. NO guard of its own — like
/// [`unwire_model_relay_config`], `owns_wiring_for` is asked at every call site.
#[cfg(feature = "model-relay")]
pub(crate) fn release_proxy_delivery(
    binding: &dyn crate::hooks::binding::AgentBinding,
    by: crate::hooks::model_relay_endpoints::ReleasedBy,
) {
    match crate::hooks::release_proxy_env(binding, by) {
        Ok(()) => tracing::info!(
            agent = binding.agent_type(),
            ?by,
            "proxy delivery released — the agent's surfaces go direct"
        ),
        Err(e) => tracing::warn!(
            code = %e.code,
            error = %e.message,
            agent = binding.agent_type(),
            ?by,
            "could not release every proxy value — the agent may still name 127.0.0.1"
        ),
    }
}

/// D-29: install the relay CA into the user's trust store — only when needed, only from the
/// machine's own install. `owns_machine` is a parameter so both branches are testable.
#[cfg(feature = "model-relay")]
pub(crate) fn ensure_proxy_env_trust(
    config: &Config,
    agents: &[crate::hooks::DetectedAgent],
    owns_machine: bool,
) -> Option<crate::model_relay::trust_store::InstallOutcome> {
    if !owns_machine {
        return None;
    }
    let needed = agents.iter().any(|a| {
        matches!(
            a.binding.model_relay_wiring().map(|w| w.endpoint),
            Some(crate::hooks::binding::EndpointConvention::ProxyEnv { .. })
        ) && owns_wiring_for(config, &*a.binding)
    });
    if !needed {
        return None;
    }
    let dir = crate::model_relay::ca::ca_dir(&crate::config::openlatch_dir());
    let ca = match crate::model_relay::ca::LocalCa::load_or_generate(&dir) {
        Ok(ca) => ca,
        Err(e) => {
            return Some(crate::model_relay::trust_store::InstallOutcome::Refused(
                format!("the relay CA could not be created: {}", e.message),
            ))
        }
    };
    Some(
        crate::model_relay::trust_store::store()
            .install(&crate::model_relay::ca::ca_pem_path(&dir), &ca.sha256_hex()),
    )
}

/// Hand back every provider slot a relay endpoint served, for every agent this
/// daemon owns the wiring for — the relay-off half of
/// [`crate::hooks::provider_endpoints::release_all`].
#[cfg(feature = "model-relay")]
fn release_every_provider_endpoint(config: &Config) {
    for agent in crate::hooks::detect_agents() {
        let Some(endpoints) = agent.binding.provider_endpoints() else {
            continue;
        };
        // An isolated instance never wrote the machine's own slots, and must
        // not restore them.
        if !owns_wiring_for(config, &*agent.binding) {
            continue;
        }
        let summary = crate::hooks::provider_endpoints::release_all(
            endpoints,
            crate::hooks::model_relay_endpoints::ReleasedBy::Teardown,
        );
        if summary.released > 0 || !summary.failed.is_empty() {
            tracing::info!(
                agent = agent.agent_type(),
                released = summary.released,
                restored = summary.restored,
                failed = summary.failed.len(),
                "model relay is off — provider slots handed back"
            );
        }
    }
}

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

/// Wait for an OS shutdown signal (SIGTERM/SIGHUP on Unix, Ctrl+C everywhere).
///
/// **SIGHUP is handled deliberately.** Its default disposition is an immediate
/// terminate: closing the terminal that ran `openlatch start --foreground`
/// killed the daemon dead — no drain, no PID-file cleanup, and the next `start`
/// reporting `Cleared stale PID file`. There is no config-reload path in this
/// codebase, so a graceful shutdown is the honest minimal behaviour, and it
/// turns an instant kill into a clean drain that any OS supervisor then
/// restarts. (`spawn_daemon_background` additionally `setsid()`s, so the
/// background daemon has no controlling terminal to be hung up on at all.)
async fn signal_handler() {
    #[cfg(unix)]
    {
        use tokio::signal::unix::{signal, SignalKind};
        let mut sigterm =
            signal(SignalKind::terminate()).expect("failed to register SIGTERM handler");
        let mut sighup = signal(SignalKind::hangup()).expect("failed to register SIGHUP handler");
        let stopped_by = tokio::select! {
            _ = tokio::signal::ctrl_c() => SIGNAL_OPERATOR,
            _ = sigterm.recv() => SIGNAL_SUPERVISOR,
            _ = sighup.recv() => {
                tracing::info!("received SIGHUP — draining (no config-reload path exists)");
                SIGNAL_SUPERVISOR
            }
        };
        SIGNAL.store(stopped_by, std::sync::atomic::Ordering::SeqCst);
    }
    #[cfg(not(unix))]
    {
        // Both console interrupts, for the reason the unix branch above listens to three
        // signals rather than one: they are the same gesture, and a daemon that drains for
        // one and dies mid-write for the other is arbitrary. Ctrl+Break used to reach the
        // default handler, which terminates the process outright — no drain of the cloud
        // worker's in-flight batch, no boundary unwiring, no PID file removed.
        //
        // `ctrl_close` (the console window closed) is deliberately NOT here: Windows gives
        // roughly five seconds before killing the process anyway, which is less than the
        // cloud flush window alone, so promising a graceful teardown there would be a
        // promise this shutdown path cannot keep.
        use tokio::signal::windows;
        let mut ctrl_break = windows::ctrl_break().expect("failed to register ctrl_break handler");
        tokio::select! {
            _ = tokio::signal::ctrl_c() => {}
            _ = ctrl_break.recv() => {}
        }
        // Both console interrupts are the operator's own gesture; Windows has no
        // supervisor-side equivalent of SIGTERM reaching this handler.
        SIGNAL.store(SIGNAL_OPERATOR, std::sync::atomic::Ordering::SeqCst);
    }
}

/// Set by [`signal_handler`] the moment a signal actually arrives, and read by the
/// foreground callers once the serve loop has returned.
///
/// **This exists because CLI-09's `ctrlc` handler cannot fire on a foreground daemon.**
/// `main` registers one that hard-exits 130, but `tokio::signal::ctrl_c` registers its own
/// afterwards, and both platforms run the newest registration first: Windows walks its
/// console-handler chain in reverse and stops at the first that returns `TRUE`, and on unix
/// tokio's `sigaction` simply replaces what was there. Tokio's handler wins, resolves the
/// future, and never exits — so the process sails past the interrupt with the CLI-09
/// promise unkept and, in `init --foreground`, runs its remaining steps against a daemon
/// the operator just stopped. That is what printed an `Overall: BROKEN` install report on a
/// deliberate Ctrl+C.
///
/// Fighting the handler chain would be the wrong repair — the graceful drain that handler
/// buys is the point of a foreground daemon. Recording the signal and letting the caller
/// act on it keeps the drain and restores the contract.
static SIGNAL: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(SIGNAL_NONE);

/// The serve loop ended some other way: `/shutdown`, the auto-update handoff, a bind
/// failure, a panic.
const SIGNAL_NONE: u8 = 0;
/// SIGINT, Ctrl+C, Ctrl+Break — a person at a terminal ended the session.
const SIGNAL_OPERATOR: u8 = 1;
/// SIGTERM, SIGHUP — a supervisor stopping the service, or the controlling terminal going
/// away. Nobody typed anything, and nothing went wrong.
///
/// Only the unix arm can produce it: Windows has no supervisor-side signal reaching this
/// handler, so a Windows build sees the constant as dead. Gated with `cfg_attr` rather than
/// `cfg` because these three constants are one enumeration of what stopped the loop —
/// compiling one out would make the SET read as platform-dependent when only the producer
/// is.
#[cfg_attr(not(unix), allow(dead_code))]
const SIGNAL_SUPERVISOR: u8 = 2;

/// Did this process's daemon stop because a signal told it to?
///
/// False for every other way a serve loop ends: `/shutdown`, the auto-update handoff, a
/// bind failure, a panic in the loop.
pub fn interrupted() -> bool {
    SIGNAL.load(std::sync::atomic::Ordering::SeqCst) != SIGNAL_NONE
}

/// Was it the operator, rather than a supervisor?
///
/// Only this answer earns exit `130`. `.claude/rules/error-handling.md` fixes that code as
/// "interrupted by Ctrl+C / SIGINT", and the distinction is load-bearing rather than
/// pedantic: SIGTERM is precisely how systemd and launchd stop a service, so answering it
/// with a failure code makes `Restart=on-failure` restart the daemon it just asked to stop.
/// SIGHUP is the controlling terminal going away, which the drain exists to survive
/// cleanly.
pub fn stopped_by_operator() -> bool {
    SIGNAL.load(std::sync::atomic::Ordering::SeqCst) == SIGNAL_OPERATOR
}

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

    /// A restart surfaces as `in_progress` / `healthz` until the new binary
    /// has answered its own health probe; only then is it `completed`.
    #[test]
    fn a_restarted_update_completes_only_on_a_healthy_probe() {
        let sentinel = update::UpdateSentinel {
            from: "0.1.0".into(),
            to: "0.2.0".into(),
            applied_at: "2026-09-23T00:00:00Z".into(),
        };
        let status = Mutex::new(update::UpdateStatusSnapshot::restarted(&sentinel));
        assert_eq!(
            status.lock().unwrap().status,
            update::UpdateStatusKind::InProgress
        );

        settle_restarted_update(&status, &sentinel, true);
        let snap = status.lock().unwrap().clone();
        assert_eq!(snap.status, update::UpdateStatusKind::Completed);
        assert_eq!(snap.stage, None);
        assert!(snap.ended_at.is_some());

        let status = Mutex::new(update::UpdateStatusSnapshot::restarted(&sentinel));
        settle_restarted_update(&status, &sentinel, false);
        let snap = status.lock().unwrap().clone();
        assert_eq!(snap.status, update::UpdateStatusKind::Failed);
        assert_eq!(snap.stage, Some(update::ApplyStage::Healthz));
    }

    /// A `POST /admin/update` that won the lock inside the probe window owns
    /// the snapshot now; the late probe must not overwrite its progress.
    #[test]
    fn settling_leaves_a_newer_apply_and_an_unseeded_status_alone() {
        let sentinel = update::UpdateSentinel {
            from: "0.1.0".into(),
            to: "0.2.0".into(),
            applied_at: "2026-09-23T00:00:00Z".into(),
        };
        for healthy in [true, false] {
            let status = Mutex::new(update::UpdateStatusSnapshot::in_progress("0.2.0", "0.3.0"));
            settle_restarted_update(&status, &sentinel, healthy);
            let snap = status.lock().unwrap().clone();
            assert_eq!(snap.status, update::UpdateStatusKind::InProgress);
            assert_eq!(snap.stage, Some(update::ApplyStage::Check));
            assert_eq!(snap.to.as_deref(), Some("0.3.0"));
            assert_eq!(snap.ended_at, None);
        }

        // A sentinel for another version never seeded a snapshot: idle stays idle.
        let status = Mutex::new(update::UpdateStatusSnapshot::idle());
        settle_restarted_update(&status, &sentinel, true);
        assert_eq!(
            status.lock().unwrap().status,
            update::UpdateStatusKind::Idle
        );
    }

    #[tokio::test]
    async fn the_post_restart_probe_retries_before_giving_up() {
        use std::sync::atomic::AtomicUsize;
        let no_pause = [std::time::Duration::ZERO; 4];

        let calls = AtomicUsize::new(0);
        let healthy = probe_until_healthy(
            || async { calls.fetch_add(1, Ordering::SeqCst) >= 2 },
            &no_pause,
        )
        .await;
        assert!(healthy, "a daemon that answers on the third try is healthy");
        assert_eq!(
            calls.load(Ordering::SeqCst),
            3,
            "stops at the first success"
        );

        let calls = AtomicUsize::new(0);
        let healthy = probe_until_healthy(
            || async {
                calls.fetch_add(1, Ordering::SeqCst);
                false
            },
            &no_pause,
        )
        .await;
        assert!(!healthy);
        assert_eq!(
            calls.load(Ordering::SeqCst),
            5,
            "one try, then one per pause"
        );

        let budget: std::time::Duration = POST_RESTART_PROBE_BACKOFF.iter().sum();
        assert_eq!(budget, std::time::Duration::from_secs(15));
    }

    /// **What the wiring recorded reaches the live listener.**
    ///
    /// The other half of `wiring_records_where_the_provider_actually_was`: the
    /// record is written when an agent's `baseUrl` is overwritten, and this is
    /// what turns it into a forward. Applied on EVERY supervisor tick rather
    /// than at startup, because on a first install the record does not exist
    /// yet when the listener binds — a start-only read misroutes that whole run,
    /// which is exactly how a customer's Ollama traffic reached api.openai.com.
    #[test]
    fn a_recorded_upstream_reaches_the_live_handle() {
        use crate::model_relay::wire_format::WireFormat;

        let _dir_lock = crate::config::OPENLATCH_DIR_ENV_LOCK
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let dir = tempfile::tempdir().expect("tempdir");
        let previous = std::env::var_os("OPENLATCH_DIR");
        std::env::set_var("OPENLATCH_DIR", dir.path());

        crate::hooks::model_relay_endpoints::record(
            &crate::hooks::upstream_record_key(WireFormat::OpenAiChatCompletions),
            Some("http://127.0.0.1:11434".to_string()),
        )
        .expect("record the endpoint the wiring replaced");

        let handle: crate::model_relay::UpstreamHandle =
            Arc::new(std::sync::RwLock::new(std::collections::BTreeMap::new()));
        let configured = std::collections::BTreeMap::new();
        apply_learned_upstreams(&handle, &configured);

        let installed = handle
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .get(WireFormat::OpenAiChatCompletions.as_str())
            .map(|u| u.to_string());
        match previous {
            Some(v) => std::env::set_var("OPENLATCH_DIR", v),
            None => std::env::remove_var("OPENLATCH_DIR"),
        }
        assert_eq!(
            installed.as_deref(),
            Some("http://127.0.0.1:11434/"),
            "the relay must forward this format to the provider it replaced"
        );
    }

    /// A record forgotten on uninstall stops routing the format: the handle goes
    /// back to the configured upstream instead of forwarding to the old endpoint
    /// until the next restart.
    #[test]
    fn a_forgotten_upstream_record_reverts_the_live_handle() {
        use crate::model_relay::wire_format::WireFormat;

        let _dir_lock = crate::config::OPENLATCH_DIR_ENV_LOCK
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let dir = tempfile::tempdir().expect("tempdir");
        let _env = crate::hooks::cline::EnvOverride::apply([(
            "OPENLATCH_DIR",
            Some(dir.path().as_os_str().to_os_string()),
        )]);
        let key = crate::hooks::upstream_record_key(WireFormat::AnthropicMessages);
        let configured: std::collections::BTreeMap<String, String> = [(
            WireFormat::AnthropicMessages.as_str().to_string(),
            "https://api.anthropic.com".to_string(),
        )]
        .into();
        let handle: crate::model_relay::UpstreamHandle =
            Arc::new(std::sync::RwLock::new(std::collections::BTreeMap::new()));
        let current = || {
            handle
                .read()
                .unwrap_or_else(|e| e.into_inner())
                .get(WireFormat::AnthropicMessages.as_str())
                .map(|u| u.to_string())
        };

        crate::hooks::model_relay_endpoints::record(&key, Some("https://gw.corp".into()))
            .expect("record");
        apply_learned_upstreams(&handle, &configured);
        assert_eq!(current().as_deref(), Some("https://gw.corp/"));

        crate::hooks::model_relay_endpoints::forget(&key);
        apply_learned_upstreams(&handle, &configured);
        assert_eq!(current().as_deref(), Some("https://api.anthropic.com/"));
    }

    /// 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 = "model-relay")]
    #[test]
    fn wiring_delay_backs_off_and_caps() {
        let t = WIRING_TICK;
        assert_eq!(wiring_delay(t, 0), t, "a green gate idles at the tick");
        assert_eq!(
            wiring_delay(t, 1),
            t,
            "the first failure retries at the tick"
        );
        assert_eq!(wiring_delay(t, 2), t * 2);
        assert_eq!(wiring_delay(t, 3), t * 4);
        // 60s * 8 = 480s, past the 300s ceiling.
        assert_eq!(wiring_delay(t, 4), WIRING_BACKOFF_MAX);
        // Still capped — and still finite — far past any plausible streak, so a
        // long offline stretch keeps probing rather than drifting to never.
        assert_eq!(wiring_delay(t, 50), WIRING_BACKOFF_MAX);
        assert_eq!(wiring_delay(t, u32::MAX), WIRING_BACKOFF_MAX);
    }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        let fx = WiringFixture::new().await;
        let cfg = fx.config();
        let wiring = WiringState::default();
        let formats = crate::cloud::worker::SourceFormats::new();

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

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

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

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

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

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

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

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

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

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

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

        let mut cfg = Config::defaults();
        cfg.model_relay.port = crate::model_relay::default_model_relay_port();

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

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

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

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

    // -------------------------------------------------------------------
    // I-2 plan 04 §3 — react_to_trust_loss (D-31 release, D-32 reaction).
    // -------------------------------------------------------------------

    /// Relocates `OPENLATCH_DIR` under the documented lock (this test never touches an agent's
    /// own config-dir seams, so `OPENLATCH_DIR_ENV_LOCK` alone is enough), and writes one
    /// `ProxyEnv` wrapper block per `(agent, command)` pair so `release_for_trust`'s real
    /// `release_proxy_env` call has something to remove.
    // S4: field order is drop order for a struct (declaration order, first-to-last — the
    // opposite of local variables). `_env` must drop BEFORE `_dir_lock` releases the lock, and
    // `ol_dir` must outlive `_env` (its path is what `_env` restores away from), so `_env` is
    // declared first, `_dir_lock` second, `ol_dir` last.
    #[cfg(feature = "model-relay")]
    struct TrustLossFixture {
        _env: crate::hooks::cline::EnvOverride,
        _dir_lock: std::sync::MutexGuard<'static, ()>,
        ol_dir: tempfile::TempDir,
    }

    #[cfg(feature = "model-relay")]
    impl TrustLossFixture {
        fn new() -> Self {
            let _dir_lock = crate::config::OPENLATCH_DIR_ENV_LOCK
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let ol_dir = tempfile::tempdir().expect("tempdir");
            let _env = crate::hooks::cline::EnvOverride::apply([(
                "OPENLATCH_DIR",
                Some(ol_dir.path().as_os_str().to_os_string()),
            )]);
            Self {
                _env,
                _dir_lock,
                ol_dir,
            }
        }

        fn write_wrapper(&self, agent: &str, command: &str) {
            let ca_pem = self
                .ol_dir
                .path()
                .join("model-relay")
                .join("ca")
                .join("ca.pem");
            crate::hooks::proxy_env_file::write_block(
                agent,
                &[command],
                "http://127.0.0.1:7600",
                &ca_pem,
                "127.0.0.1:7600",
            )
            .expect("write wrapper block");
        }

        fn env_sh(&self) -> String {
            std::fs::read_to_string(crate::hooks::proxy_env_file::env_sh_path(
                self.ol_dir.path(),
            ))
            .unwrap_or_default()
        }

        fn proxy_env_agent(
            &self,
            agent_type: &'static str,
            delivery: &'static [crate::hooks::binding::ProxyDelivery],
            hosts: &'static [&'static str],
        ) -> crate::hooks::DetectedAgent {
            use crate::hooks::binding::test_support::{proxy_env_wiring, FakeBinding};
            crate::hooks::DetectedAgent {
                kind: crate::hooks::AgentKind::ClaudeCode,
                binding: Arc::new(FakeBinding {
                    agent_type,
                    model_relay_wiring: Some(proxy_env_wiring(hosts, delivery)),
                    ..Default::default()
                }),
            }
        }
    }

    #[test]
    #[cfg(feature = "model-relay")]
    fn seven_days_before_expiry_every_proxy_env_delivery_is_released() {
        use crate::hooks::binding::{EndpointConvention, ModelRelayWiring, ProxyDelivery};
        use crate::model_relay::ca::CaInfo;
        use crate::model_relay::ca_lifecycle::expiring_reason;
        use crate::model_relay::intercept::RefusalLedger;
        use crate::model_relay::preflight::{Verdict, WiringState};
        use crate::model_relay::wire_format::WireFormat;

        const DELIVERY_A: &[ProxyDelivery] = &[ProxyDelivery::EnvFile {
            commands: &["fake-a-cli"],
        }];
        const DELIVERY_B: &[ProxyDelivery] = &[ProxyDelivery::EnvFile {
            commands: &["fake-b-cli"],
        }];

        let cfg = Config::defaults();
        let at = TrustClock {
            wall: time::OffsetDateTime::now_utc(),
            mono: std::time::Instant::now(),
        };

        // Scoped so `fx`'s `OPENLATCH_DIR_ENV_LOCK` guard drops before the control below
        // takes it again — the mutex is not reentrant.
        {
            let fx = TrustLossFixture::new();
            fx.write_wrapper("fake-a", "fake-a-cli");
            fx.write_wrapper("fake-b", "fake-b-cli");

            let agent_a = fx.proxy_env_agent("fake-a", DELIVERY_A, &["a.test"]);
            let agent_b = fx.proxy_env_agent("fake-b", DELIVERY_B, &["b.test"]);
            let envvars_agent = crate::hooks::DetectedAgent {
                kind: crate::hooks::AgentKind::ClaudeCode,
                binding: Arc::new(crate::hooks::binding::test_support::FakeBinding {
                    agent_type: "claude-shaped",
                    model_relay_wiring: Some(ModelRelayWiring {
                        wire_format: WireFormat::AnthropicMessages,
                        endpoint: EndpointConvention::EnvVars {
                            base_url: "ANTHROPIC_BASE_URL",
                            headers: "ANTHROPIC_CUSTOM_HEADERS",
                        },
                        install_id_header: "x-openlatch-install-id",
                    }),
                    ..Default::default()
                }),
            };
            let agents = vec![agent_a, agent_b, envvars_agent];

            let wiring = WiringState::default();
            wiring.set_wired("fake-a", true);
            wiring.set_wired("fake-b", true);
            let ca = CaInfo {
                sha256_hex: "deadbeef".to_string(),
                not_after: time::OffsetDateTime::now_utc() + time::Duration::days(6),
            };
            let refusals = RefusalLedger::default();
            let mut holds = std::collections::BTreeMap::new();

            let skip =
                react_to_trust_loss(&cfg, &agents, Some(&ca), at, &refusals, &mut holds, &wiring);

            assert!(skip.contains("fake-a"));
            assert!(skip.contains("fake-b"));
            assert!(!skip.contains("claude-shaped"));

            let content = fx.env_sh();
            assert!(
                !content.contains("fake-a"),
                "fake-a's block must be gone: {content}"
            );
            assert!(
                !content.contains("fake-b"),
                "fake-b's block must be gone: {content}"
            );

            let reason = expiring_reason(ca.not_after);
            assert_eq!(wiring.verdict("fake-a"), Verdict::Failed(reason.clone()));
            assert_eq!(wiring.verdict("fake-b"), Verdict::Failed(reason));
            assert!(!wiring.is_wired("fake-a"));
            assert!(!wiring.is_wired("fake-b"));
        }

        // Control: with 8 days left, nothing is released and `skip` is empty.
        {
            let fx2 = TrustLossFixture::new();
            fx2.write_wrapper("fake-a", "fake-a-cli");
            let agent_a2 = fx2.proxy_env_agent("fake-a", DELIVERY_A, &["a.test"]);
            let agents2 = vec![agent_a2];
            let wiring2 = WiringState::default();
            wiring2.set_wired("fake-a", true);
            let ca8 = CaInfo {
                sha256_hex: "deadbeef".to_string(),
                not_after: time::OffsetDateTime::now_utc() + time::Duration::days(8),
            };
            let refusals2 = RefusalLedger::default();
            let mut holds2 = std::collections::BTreeMap::new();
            let skip2 = react_to_trust_loss(
                &cfg,
                &agents2,
                Some(&ca8),
                at,
                &refusals2,
                &mut holds2,
                &wiring2,
            );
            assert!(
                skip2.is_empty(),
                "outside CA_RELEASE_DAYS, nothing is released"
            );
            assert!(fx2.env_sh().contains("fake-a"), "the block must survive");
        }
    }

    #[test]
    #[cfg(feature = "model-relay")]
    fn a_tripped_host_releases_only_the_bindings_that_name_it() {
        use crate::hooks::binding::ProxyDelivery;
        use crate::model_relay::ca_lifecycle::refused_reason;
        use crate::model_relay::intercept::{RefusalKind, RefusalLedger};
        use crate::model_relay::preflight::{Verdict, WiringState};

        const DELIVERY_A: &[ProxyDelivery] = &[ProxyDelivery::EnvFile {
            commands: &["fake-a-cli"],
        }];
        const DELIVERY_B: &[ProxyDelivery] = &[ProxyDelivery::EnvFile {
            commands: &["fake-b-cli"],
        }];

        let fx = TrustLossFixture::new();
        fx.write_wrapper("fake-a", "fake-a-cli");
        fx.write_wrapper("fake-b", "fake-b-cli");
        let agent_a = fx.proxy_env_agent("fake-a", DELIVERY_A, &["a.test"]);
        let agent_b = fx.proxy_env_agent("fake-b", DELIVERY_B, &["b.test"]);
        let agents = vec![agent_a, agent_b];
        let cfg = Config::defaults();
        let wiring = WiringState::default();
        wiring.set_wired("fake-a", true);
        wiring.set_wired("fake-b", true);

        let refusals = RefusalLedger::default();
        for _ in 0..3 {
            refusals.record_refusal("a.test", RefusalKind::HandshakeEof);
        }
        let mut holds = std::collections::BTreeMap::new();
        let at = TrustClock {
            wall: time::OffsetDateTime::now_utc(),
            mono: std::time::Instant::now(),
        };

        let skip = react_to_trust_loss(&cfg, &agents, None, at, &refusals, &mut holds, &wiring);

        assert!(skip.contains("fake-a"));
        assert!(!skip.contains("fake-b"));
        assert_eq!(
            wiring.verdict("fake-a"),
            Verdict::Failed(refused_reason("a.test"))
        );
        assert!(!wiring.is_wired("fake-a"));
        assert!(wiring.is_wired("fake-b"), "B is untouched");
        assert_eq!(
            wiring.verdict("fake-b"),
            Verdict::Pending,
            "B's verdict was never set by this reaction"
        );
        assert!(!fx.env_sh().contains("fake-a"), "A's block is gone");
        assert!(fx.env_sh().contains("fake-b"), "B's block survives");
    }

    #[test]
    #[cfg(feature = "model-relay")]
    fn a_hold_blocks_rearming_until_restart() {
        use crate::hooks::binding::ProxyDelivery;
        use crate::model_relay::intercept::{RefusalKind, RefusalLedger};
        use crate::model_relay::preflight::WiringState;

        const DELIVERY_A: &[ProxyDelivery] = &[ProxyDelivery::EnvFile {
            commands: &["fake-a-cli"],
        }];

        let fx = TrustLossFixture::new();
        fx.write_wrapper("fake-a", "fake-a-cli");
        let agent_a = fx.proxy_env_agent("fake-a", DELIVERY_A, &["a.test"]);
        let agents = vec![agent_a];
        let cfg = Config::defaults();
        let wiring = WiringState::default();
        wiring.set_wired("fake-a", true);

        let refusals = RefusalLedger::default();
        for _ in 0..3 {
            refusals.record_refusal("a.test", RefusalKind::HandshakeEof);
        }
        let mut holds = std::collections::BTreeMap::new();
        let base = std::time::Instant::now();

        let at1 = TrustClock {
            wall: time::OffsetDateTime::now_utc(),
            mono: base,
        };
        let skip1 = react_to_trust_loss(&cfg, &agents, None, at1, &refusals, &mut holds, &wiring);
        assert!(skip1.contains("fake-a"));
        assert!(holds.contains_key("a.test"));
        assert!(!wiring.is_wired("fake-a"));

        // clock + 11 min: still held, no second release.
        let at2 = TrustClock {
            wall: time::OffsetDateTime::now_utc(),
            mono: base + std::time::Duration::from_secs(11 * 60),
        };
        let skip2 = react_to_trust_loss(&cfg, &agents, None, at2, &refusals, &mut holds, &wiring);
        assert!(skip2.contains("fake-a"));
        assert!(holds.contains_key("a.test"));

        // clock + 2 h: still held.
        let at3 = TrustClock {
            wall: time::OffsetDateTime::now_utc(),
            mono: base + std::time::Duration::from_secs(2 * 60 * 60),
        };
        let skip3 = react_to_trust_loss(&cfg, &agents, None, at3, &refusals, &mut holds, &wiring);
        assert!(skip3.contains("fake-a"));
        assert!(holds.contains_key("a.test"));

        // A fresh daemon (a restart): a fresh ledger and empty holds re-arm it.
        let fresh_refusals = RefusalLedger::default();
        let mut fresh_holds = std::collections::BTreeMap::new();
        let skip4 = react_to_trust_loss(
            &cfg,
            &agents,
            None,
            at3,
            &fresh_refusals,
            &mut fresh_holds,
            &wiring,
        );
        assert!(!skip4.contains("fake-a"), "a restart re-arms");
    }

    /// 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 = "model-relay")]
    #[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.schema_version, 1);
            assert_eq!(
                bundle.digest.as_deref(),
                Some(store::digest_of(BODY.as_bytes()).as_str())
            );
            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 digest_valid_but_unsupported_cache_is_discarded_before_polling() {
            let tmp = tempfile::tempdir().expect("tempdir");
            let valid: PolicyBundle = serde_json::from_str(BODY).expect("fixture parses");
            let unsupported = br#"{"schema_version":3}"#;
            let meta = BundleMeta::activated(
                &valid,
                store::digest_of(unsupported),
                Some("\"sha256:stale\"".to_string()),
            );
            store::store(tmp.path(), unsupported, &meta).expect("cache writes");

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

            assert!(runtime.handle.load().is_none());
            assert!(!store::bundle_path(tmp.path()).exists());
            assert!(!store::meta_path(tmp.path()).exists());
        }

        #[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));
        }
    }

    // -- I-2 plan 02: ProxyEnv trust and proxy delivery ----------------------

    #[cfg(feature = "model-relay")]
    mod proxy_env {
        use super::*;
        use crate::hooks::binding::test_support::{proxy_env_wiring, FakeBinding};
        use crate::hooks::binding::{EndpointConvention, ModelRelayWiring, ProxyDelivery};
        use crate::hooks::model_relay_endpoints::ReleasedBy;
        use crate::hooks::{proxy_env_file, proxy_settings};
        use crate::model_relay::preflight::{Verdict, WiringState};
        use crate::model_relay::trust_store::{
            self, test_support::FakeStore, InstallOutcome, TrustStore,
        };
        use crate::model_relay::wire_format::WireFormat;
        use crate::model_relay::{mock, serve_ephemeral, ModelRelayState};

        const HOST: &str = "a.test";
        const AGENT: &str = "fake-a";
        const KEY: &str = "fake.proxy";
        /// A customer file carrying comments, a trailing comma and no `fake.proxy` — the byte-
        /// identity table's shape 2, so a write-then-release round trip is exact.
        const SEED: &str = "{\n  \"a\": 1,\n}\n";
        const FAKE_DELIVERY: &[ProxyDelivery] = &[
            ProxyDelivery::SettingsKey {
                file: fake_settings_file,
                key: KEY,
            },
            ProxyDelivery::EnvFile {
                commands: &["fakecli"],
            },
        ];
        const FOREIGN_ONLY_DELIVERY: &[ProxyDelivery] = &[ProxyDelivery::SettingsKey {
            file: fake_settings_file,
            key: KEY,
        }];
        const ENV_FILE_ONLY_DELIVERY: &[ProxyDelivery] = &[ProxyDelivery::EnvFile {
            commands: &["fakecli"],
        }];

        /// The path every `SettingsKey` delivery in this module points at — inside the locked
        /// temp `OPENLATCH_DIR`, so a plain `fn` pointer (the shape `ProxyDelivery::SettingsKey`
        /// requires) needs no lock of its own.
        fn fake_settings_file() -> Option<std::path::PathBuf> {
            Some(
                crate::config::openlatch_dir()
                    .join("fake-agent")
                    .join("settings.json"),
            )
        }

        fn settings_path() -> std::path::PathBuf {
            fake_settings_file().expect("fake_settings_file always answers Some")
        }

        fn seed_settings(content: &str) {
            let path = settings_path();
            std::fs::create_dir_all(path.parent().expect("parent")).expect("settings dir");
            std::fs::write(&path, content).expect("seed settings");
        }

        /// The `Config` both `ProxyEnvFixture::config` and the owner acceptance test need: the
        /// agent id pinned so nothing can fall back to a real one, and the ownership guard
        /// opted in on the default port — what a sandbox looks like (`WiringFixture::config`'s
        /// twin, shared here rather than there since only this module's tests need it).
        fn proxy_env_test_config(agent_id: &str) -> Config {
            let mut cfg = Config::defaults();
            cfg.agent_id = Some(agent_id.to_string());
            cfg.model_relay.own_agent_wiring = Some(true);
            cfg.model_relay.port = crate::model_relay::default_model_relay_port();
            cfg
        }

        fn env_sh_exists() -> bool {
            proxy_env_file::env_sh_path(&crate::config::openlatch_dir()).exists()
        }

        fn marker_exists() -> bool {
            proxy_env_file::live_marker_path(&crate::config::openlatch_dir()).exists()
        }

        fn settings_record(
            agent: &str,
            key: &str,
        ) -> Option<crate::hooks::model_relay_endpoints::EndpointRecord> {
            let prefix = format!("{}{agent}:", proxy_settings::RECORD_PREFIX);
            let recs = crate::hooks::model_relay_endpoints::endpoint_records(&prefix)
                .expect("read the model-relay-endpoints ledger");
            recs.into_iter()
                .find(|(k, _)| k.starts_with(&format!("{prefix}{key}@")))
                .map(|(_, r)| r)
        }

        /// Everything a `ProxyEnv` wiring test needs: a real intercepting relay (plan 01's
        /// tunnel) over a mock upstream, an isolated `OPENLATCH_DIR` and `HOME`, and the OS trust
        /// store replaced by [`FakeStore`] so no test ever reaches a real keychain.
        struct ProxyEnvFixture {
            port: u16,
            _openlatch_dir: tempfile::TempDir,
            home_dir: tempfile::TempDir,
            wiring: Arc<WiringState>,
            fake_store: Arc<FakeStore>,
            previous_dir: Option<std::ffi::OsString>,
            previous_home: Option<std::ffi::OsString>,
            // Declaration order is drop order (LIFO against acquisition): the trust-store lock
            // was taken LAST (`install_for_tests`), so it is released FIRST; `_dir_lock`
            // (`OPENLATCH_DIR_ENV_LOCK`) was taken first and is released last.
            _trust_guard: trust_store::TrustStoreGuard,
            _home_lock: std::sync::MutexGuard<'static, ()>,
            _dir_lock: std::sync::MutexGuard<'static, ()>,
        }

        impl Drop for ProxyEnvFixture {
            fn drop(&mut self) {
                match self.previous_home.take() {
                    Some(v) => std::env::set_var("HOME", v),
                    None => std::env::remove_var("HOME"),
                }
                match self.previous_dir.take() {
                    Some(v) => std::env::set_var("OPENLATCH_DIR", v),
                    None => std::env::remove_var("OPENLATCH_DIR"),
                }
            }
        }

        impl ProxyEnvFixture {
            async fn new() -> Self {
                Self::build(mock::spawn_always_200().await).await
            }

            /// The fixture over a relay whose explicit Anthropic upstream is `upstream`.
            async fn build(upstream: u16) -> Self {
                let wiring = Arc::new(WiringState::default());
                let map: std::collections::BTreeMap<String, String> = [(
                    WireFormat::AnthropicMessages.as_str().to_string(),
                    format!("http://127.0.0.1:{upstream}"),
                )]
                .into_iter()
                .collect();
                let state = Arc::new(
                    ModelRelayState::new(
                        reqwest::Url::parse(&format!("http://127.0.0.1:{upstream}")).unwrap(),
                        0,
                        8,
                        &[],
                    )
                    .with_upstream_map(map)
                    .with_explicit_upstreams([WireFormat::AnthropicMessages.as_str()].into())
                    .with_wiring(wiring.clone())
                    .with_intercept(wiring.intercept())
                    .with_refusals(wiring.refusals()),
                );
                let port = serve_ephemeral(state).await;

                // LOCK ORDER: `config::OPENLATCH_DIR_ENV_LOCK` -> `claude_code::CONFIG_DIR_ENV_LOCK`
                // -> ... -> `trust_store::TEST_SLOT_LOCK` (via `install_for_tests`), always
                // acquired last — all AFTER every `.await` above, never held across one.
                let _dir_lock = crate::config::OPENLATCH_DIR_ENV_LOCK
                    .lock()
                    .unwrap_or_else(|e| e.into_inner());
                let _home_lock = crate::hooks::claude_code::CONFIG_DIR_ENV_LOCK
                    .lock()
                    .unwrap_or_else(|e| e.into_inner());

                let _openlatch_dir = tempfile::tempdir().expect("tempdir");
                let previous_dir = std::env::var_os("OPENLATCH_DIR");
                std::env::set_var("OPENLATCH_DIR", _openlatch_dir.path());

                let home_dir = tempfile::tempdir().expect("tempdir");
                let previous_home = std::env::var_os("HOME");
                std::env::set_var("HOME", home_dir.path());

                let fake_store = Arc::new(FakeStore::default());
                let _trust_guard =
                    trust_store::install_for_tests(fake_store.clone() as Arc<dyn TrustStore>);

                seed_settings(SEED);

                Self {
                    port,
                    _openlatch_dir,
                    home_dir,
                    wiring,
                    fake_store,
                    previous_dir,
                    previous_home,
                    _trust_guard,
                    _home_lock,
                    _dir_lock,
                }
            }

            /// A `Config` that can never reach the developer's own files: the agent id is
            /// pinned and the ownership guard is opted in on the default port, which is what a
            /// sandbox looks like (`WiringFixture::config`'s twin).
            fn config(&self) -> Config {
                proxy_env_test_config("agt_test")
            }

            fn agent(
                &self,
                hosts: &'static [&'static str],
                delivery: &'static [ProxyDelivery],
            ) -> crate::hooks::DetectedAgent {
                crate::hooks::DetectedAgent {
                    kind: crate::hooks::AgentKind::ClaudeCode,
                    binding: Arc::new(FakeBinding {
                        agent_type: AGENT,
                        display_name: "Fake",
                        config_dir: crate::config::openlatch_dir().join("fake-agent"),
                        model_relay_wiring: Some(proxy_env_wiring(hosts, delivery)),
                        ..Default::default()
                    }),
                }
            }
        }

        fn env_vars_agent(agent_type: &'static str) -> crate::hooks::DetectedAgent {
            crate::hooks::DetectedAgent {
                kind: crate::hooks::AgentKind::ClaudeCode,
                binding: Arc::new(FakeBinding {
                    agent_type,
                    model_relay_wiring: Some(ModelRelayWiring {
                        wire_format: WireFormat::AnthropicMessages,
                        endpoint: EndpointConvention::EnvVars {
                            base_url: "ANTHROPIC_BASE_URL",
                            headers: "ANTHROPIC_CUSTOM_HEADERS",
                        },
                        install_id_header: "x-openlatch-install-id",
                    }),
                    ..Default::default()
                }),
            }
        }

        fn toml_provider_agent(agent_type: &'static str) -> crate::hooks::DetectedAgent {
            crate::hooks::DetectedAgent {
                kind: crate::hooks::AgentKind::ClaudeCode,
                binding: Arc::new(FakeBinding {
                    agent_type,
                    model_relay_wiring: Some(ModelRelayWiring {
                        wire_format: WireFormat::OpenAiResponses,
                        endpoint: EndpointConvention::TomlProvider {
                            provider_name: "fake-provider",
                            wire_api: "chat",
                        },
                        install_id_header: "x-openlatch-install-id",
                    }),
                    ..Default::default()
                }),
            }
        }

        #[test]
        fn the_interceptor_is_built_only_for_proxy_env() {
            let non_proxy_env = vec![
                env_vars_agent("envvars-agent"),
                toml_provider_agent("toml-agent"),
            ];
            assert!(proxy_env_hosts(&non_proxy_env).is_empty());

            let proxy_env_agent = crate::hooks::DetectedAgent {
                kind: crate::hooks::AgentKind::ClaudeCode,
                binding: Arc::new(FakeBinding {
                    agent_type: AGENT,
                    model_relay_wiring: Some(proxy_env_wiring(&[HOST], FAKE_DELIVERY)),
                    ..Default::default()
                }),
            };
            let hosts = proxy_env_hosts(&[proxy_env_agent]);
            assert_eq!(hosts, [HOST.to_string()].into_iter().collect());

            let _dir_lock = crate::config::OPENLATCH_DIR_ENV_LOCK
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let dir = tempfile::tempdir().expect("tempdir");
            let previous = std::env::var_os("OPENLATCH_DIR");
            std::env::set_var("OPENLATCH_DIR", dir.path());

            let slot: crate::model_relay::InterceptSlot = Default::default();
            let ic = refresh_interceptor(&slot, Default::default());
            assert!(ic.is_none(), "an empty host set must leave the slot empty");
            assert!(
                !crate::model_relay::ca::ca_dir(&crate::config::openlatch_dir()).exists(),
                "no ProxyEnv host must mean no CA is ever generated"
            );

            match previous {
                Some(v) => std::env::set_var("OPENLATCH_DIR", v),
                None => std::env::remove_var("OPENLATCH_DIR"),
            }
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn proxy_env_agent_is_wired_only_after_both_proofs() {
            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(true, Ordering::SeqCst);
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();

            wire_agents(
                vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;

            assert_eq!(fx.wiring.verdict(AGENT), Verdict::Ok);
            assert!(fx.wiring.is_wired(AGENT));
            let settings = std::fs::read_to_string(settings_path()).expect("settings file");
            assert!(settings.contains(&format!("http://127.0.0.1:{}", fx.port)));
            assert!(env_sh_exists());
            assert!(marker_exists());
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn an_untrusted_store_leaves_every_delivery_unwritten() {
            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(false, Ordering::SeqCst);
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();

            wire_agents(
                vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;

            match fx.wiring.verdict(AGENT) {
                Verdict::Failed(reason) => assert!(
                    reason.starts_with(crate::model_relay::preflight::CA_REASON_MARKER),
                    "reason must carry the CA marker: {reason}"
                ),
                other => panic!("expected Failed, got {other:?}"),
            }
            assert!(!fx.wiring.is_wired(AGENT));
            assert_eq!(std::fs::read_to_string(settings_path()).unwrap(), SEED);
            assert!(!env_sh_exists());
            assert!(!marker_exists());
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn a_lone_foreign_settings_key_fails_the_verdict_with_the_marker() {
            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(true, Ordering::SeqCst);
            let foreign_seed = "{\n  \"fake.proxy\": \"http://corp:3128\"\n}\n";
            seed_settings(foreign_seed);
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();

            wire_agents(
                vec![fx.agent(&[HOST], FOREIGN_ONLY_DELIVERY)],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;

            match fx.wiring.verdict(AGENT) {
                Verdict::Failed(reason) => {
                    assert!(
                        reason.starts_with(crate::model_relay::preflight::SETTINGS_REASON_MARKER)
                    );
                    assert!(reason.contains(crate::model_relay::preflight::SETTINGS_FOREIGN));
                }
                other => panic!("expected Failed, got {other:?}"),
            }
            assert!(!fx.wiring.is_wired(AGENT));
            assert_eq!(
                std::fs::read_to_string(settings_path()).unwrap(),
                foreign_seed
            );
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn a_garbage_ca_releases_what_was_written() {
            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(true, Ordering::SeqCst);
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();

            // Tick 1: green, the slot is filled and everything is written.
            wire_agents(
                vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;
            assert_eq!(fx.wiring.verdict(AGENT), Verdict::Ok);
            assert!(fx.wiring.is_wired(AGENT));

            // Corrupt ca.pem on disk WITHOUT touching the in-memory slot.
            let ca_pem = crate::model_relay::ca::ca_pem_path(&crate::model_relay::ca::ca_dir(
                &crate::config::openlatch_dir(),
            ));
            std::fs::write(&ca_pem, "garbage").expect("corrupt ca.pem");

            // Tick 2, forced to re-probe the already-wired agent — `forwarding_broke` is the
            // only thing that re-probes a wired agent.
            wire_agents(
                vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                fx.port,
                true,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;

            match fx.wiring.verdict(AGENT) {
                Verdict::Failed(reason) => assert!(
                    reason.starts_with(crate::model_relay::preflight::CA_REASON_MARKER),
                    "reason must carry the CA marker: {reason}"
                ),
                other => panic!("expected Failed, got {other:?}"),
            }
            assert!(!fx.wiring.is_wired(AGENT));
            assert_eq!(std::fs::read_to_string(settings_path()).unwrap(), SEED);
            assert!(!env_sh_exists());
            assert!(!marker_exists());
            let rec = settings_record(AGENT, KEY)
                .expect("the settings record must survive as a tombstone");
            assert_eq!(rec.released_by, Some(ReleasedBy::Wiring));
        }

        /// Adversarial review: `FAKE_DELIVERY`'s `SettingsKey` (declared first) writes and
        /// wires successfully; its `EnvFile` (declared second) then hard-fails because a
        /// directory sits where `env.sh` must be written — `write_block`'s rename-over-path
        /// fails with a raw IO error, not a settings skip, and unwinds
        /// `write_model_relay_config` with `?` before the settings delivery's own write is
        /// ever taken back. Without the fix, this agent is simply never marked wired and
        /// nothing releases the settings key it already wrote.
        #[tokio::test(flavor = "multi_thread")]
        async fn a_failed_later_delivery_releases_the_earlier_ones() {
            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(true, Ordering::SeqCst);
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();

            let env_sh = proxy_env_file::env_sh_path(&crate::config::openlatch_dir());
            std::fs::create_dir_all(&env_sh).expect("plant a directory at env.sh's path");

            wire_agents(
                vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;

            assert!(!fx.wiring.is_wired(AGENT));
            assert_eq!(
                std::fs::read_to_string(settings_path()).unwrap(),
                SEED,
                "the settings key a later delivery's hard failure left behind must be released"
            );
            let rec = settings_record(AGENT, KEY)
                .expect("the settings record must survive as a tombstone");
            assert_eq!(rec.released_by, Some(ReleasedBy::Wiring));
            assert!(!marker_exists());
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn teardown_releases_every_proxy_value() {
            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(true, Ordering::SeqCst);
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();
            let agent = fx.agent(&[HOST], FAKE_DELIVERY);
            let binding = agent.binding.clone();

            wire_agents(vec![agent], fx.port, false, &cfg, &fx.wiring, &formats).await;
            assert!(fx.wiring.is_wired(AGENT));

            unwire_model_relay_config(&cfg, &*binding);

            assert_eq!(std::fs::read_to_string(settings_path()).unwrap(), SEED);
            assert!(!env_sh_exists());
            assert!(!marker_exists());
            let rec = settings_record(AGENT, KEY)
                .expect("the settings record must survive as a tombstone");
            assert_eq!(rec.released_by, Some(ReleasedBy::Teardown));
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn release_proxy_delivery_records_who_released() {
            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(true, Ordering::SeqCst);
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();
            let agent = fx.agent(&[HOST], FAKE_DELIVERY);
            let binding = agent.binding.clone();
            wire_agents(vec![agent], fx.port, false, &cfg, &fx.wiring, &formats).await;

            release_proxy_delivery(&*binding, ReleasedBy::Wiring);
            let rec = settings_record(AGENT, KEY).expect("record");
            assert_eq!(rec.released_by, Some(ReleasedBy::Wiring));

            let ledger_path = crate::config::openlatch_dir().join("model-relay-endpoints.json");
            let bytes_before = std::fs::read(&ledger_path).expect("ledger bytes");
            release_proxy_delivery(&*binding, ReleasedBy::Wiring);
            let bytes_after = std::fs::read(&ledger_path).expect("ledger bytes");
            assert_eq!(bytes_before, bytes_after, "a second release is a no-op");
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn no_machine_global_write_from_a_relocated_instance() {
            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(true, Ordering::SeqCst);
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();

            wire_agents(
                vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;
            assert!(fx.wiring.is_wired(AGENT));

            for rc in [".zshrc", ".bashrc", ".bash_profile", ".profile"] {
                assert!(
                    !fx.home_dir.path().join(rc).exists(),
                    "{rc} must not be created by a relocated (non-machine-owning) instance"
                );
            }
            assert_eq!(fx.fake_store.install_calls.load(Ordering::SeqCst), 0);
        }

        #[test]
        fn machine_global_gates_need_both_predicates() {
            assert!(profile_line_wanted(ENV_FILE_ONLY_DELIVERY, true));
            assert!(!profile_line_wanted(ENV_FILE_ONLY_DELIVERY, false));
            assert!(!profile_line_wanted(FOREIGN_ONLY_DELIVERY, true));
            assert!(!profile_line_wanted(FOREIGN_ONLY_DELIVERY, false));
        }

        #[test]
        fn env_trust_only_is_an_isolated_instance_that_owns_the_wiring() {
            // (owns_wiring, owns_machine, on_default_port) -> env_trust_only
            let rows = [
                ((true, false, false), true),
                // The machine install, and anything on its port, keeps the store lane (D-19).
                ((true, true, false), false),
                ((true, true, true), false),
                ((true, false, true), false),
                ((false, false, false), false),
                ((false, true, false), false),
                ((false, false, true), false),
                ((false, true, true), false),
            ];
            for ((w, m, d), want) in rows {
                assert_eq!(
                    env_trust_only_from(w, m, d),
                    want,
                    "wiring={w} machine={m} default_port={d}"
                );
            }
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn an_isolated_instance_wires_the_wrapper_without_the_store() {
            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(false, Ordering::SeqCst);
            // What olbox exports: a non-default relay port, the explicit opt-in, a relocated
            // agent (FakeBinding is not machine-global) and a relocated OPENLATCH_DIR.
            let mut cfg = fx.config();
            cfg.model_relay.port = fx.port;
            assert!(!crate::supervision::owns_machine_supervision());
            let formats = crate::cloud::worker::SourceFormats::new();

            wire_agents(
                vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;

            assert_eq!(fx.wiring.verdict(AGENT), Verdict::Ok);
            assert!(fx.wiring.is_wired(AGENT));
            assert!(env_sh_exists());
            assert!(marker_exists());
            assert_eq!(
                std::fs::read_to_string(settings_path()).unwrap(),
                SEED,
                "the store-trusted settings key is never delivered from an isolated instance"
            );
            assert_eq!(fx.fake_store.install_calls.load(Ordering::SeqCst), 0);
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn ensure_proxy_env_trust_runs_only_when_needed() {
            let fx = ProxyEnvFixture::new().await;
            let cfg = fx.config();

            let envvars_only = vec![env_vars_agent("envvars-agent")];
            assert!(ensure_proxy_env_trust(&cfg, &envvars_only, true).is_none());
            assert_eq!(fx.fake_store.install_calls.load(Ordering::SeqCst), 0);

            let proxy_agents = vec![fx.agent(&[HOST], FAKE_DELIVERY)];
            assert!(ensure_proxy_env_trust(&cfg, &proxy_agents, false).is_none());
            assert_eq!(fx.fake_store.install_calls.load(Ordering::SeqCst), 0);

            let outcome = ensure_proxy_env_trust(&cfg, &proxy_agents, true);
            assert_eq!(outcome, Some(InstallOutcome::Installed));
            assert_eq!(fx.fake_store.install_calls.load(Ordering::SeqCst), 1);
            let dir = crate::model_relay::ca::ca_dir(&crate::config::openlatch_dir());
            assert!(crate::model_relay::ca::ca_pem_path(&dir).is_file());
            let sha = crate::model_relay::ca::LocalCa::load_or_generate(&dir)
                .expect("the CA now exists")
                .sha256_hex();
            assert!(fx.fake_store.is_trusted(&sha).expect("is_trusted"));

            let second = ensure_proxy_env_trust(&cfg, &proxy_agents, true);
            assert_eq!(second, Some(InstallOutcome::AlreadyTrusted));
        }

        #[test]
        fn ca_expiry_gate_is_seven_days() {
            let now = time::OffsetDateTime::now_utc();
            assert!(
                ca_too_close_to_expiry(now + time::Duration::days(6), now),
                "6 days left must be too close"
            );
            assert!(
                ca_too_close_to_expiry(
                    now + time::Duration::days(7) + time::Duration::hours(23),
                    now
                ),
                "7 days 23 hours left is still whole_days() == 7 — 04's own boundary"
            );
            assert!(
                !ca_too_close_to_expiry(now + time::Duration::days(8), now),
                "8 days left must not be too close"
            );
        }

        /// D-36 / §11 — the owner's own Mac, real `HOME`, real login keychain, real VS Code.
        /// Never run by an agent or by CI: `#[ignore]`, macOS-only, and it refuses outright
        /// unless every precondition below holds — none of which hold in this sandbox or in CI.
        #[test]
        #[ignore]
        #[cfg(target_os = "macos")]
        fn owner_acceptance_real_store_on_real_home() {
            fn run(program: &str, args: &[&str]) -> (bool, String) {
                let out = std::process::Command::new(program)
                    .args(args)
                    .output()
                    .unwrap_or_else(|e| panic!("could not run {program}: {e}"));
                (
                    out.status.success(),
                    String::from_utf8_lossy(&out.stdout).into_owned(),
                )
            }
            fn acceptance_settings_file() -> Option<std::path::PathBuf> {
                std::env::var("OPENLATCH_ACCEPT_SETTINGS")
                    .ok()
                    .map(std::path::PathBuf::from)
            }

            if std::env::var("OPENLATCH_OWNER_ACCEPTANCE").as_deref() != Ok("1") {
                panic!(
                    "refuses: set OPENLATCH_OWNER_ACCEPTANCE=1 to run this owner acceptance check"
                );
            }
            let (ok, keychains) = run("security", &["list-keychains", "-d", "user"]);
            if !ok || !keychains.contains("login.keychain-db") {
                panic!(
                    "refuses: `security list-keychains -d user` does not name login.keychain-db \
                     — this looks like a sandbox HOME"
                );
            }
            let (ok, manager) = run("launchctl", &["managername"]);
            if !ok || manager.trim() != "Aqua" {
                panic!("refuses: no desktop session (`launchctl managername` != Aqua)");
            }
            let settings_path = acceptance_settings_file().unwrap_or_else(|| {
                panic!("refuses: set OPENLATCH_ACCEPT_SETTINGS to an absolute settings.json path")
            });
            if !settings_path.is_absolute() {
                panic!("refuses: OPENLATCH_ACCEPT_SETTINGS must be an absolute path");
            }
            let snapshot = std::fs::read_to_string(&settings_path).unwrap_or_else(|e| {
                panic!("refuses: could not read {}: {e}", settings_path.display())
            });
            let parsed = jsonc_parser::parse_to_ast(
                &snapshot,
                &jsonc_parser::CollectOptions {
                    comments: jsonc_parser::CommentCollectionStrategy::Off,
                    tokens: false,
                },
                &jsonc_parser::ParseOptions::default(),
            )
            .unwrap_or_else(|e| {
                panic!(
                    "refuses: {} does not parse as JSONC: {e}",
                    settings_path.display()
                )
            });
            match parsed.value {
                Some(jsonc_parser::ast::Value::Object(o)) if !o.properties.is_empty() => {}
                _ => panic!(
                    "refuses: {} must parse as a non-empty JSON object",
                    settings_path.display()
                ),
            }

            let _dir_lock = crate::config::OPENLATCH_DIR_ENV_LOCK
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let openlatch_dir = tempfile::tempdir().expect("tempdir");
            let previous_dir = std::env::var_os("OPENLATCH_DIR");
            std::env::set_var("OPENLATCH_DIR", openlatch_dir.path());

            // Deliberately the REAL trust-store arm — the only place in this crate that installs
            // one under `cfg(test)`, and only because every precondition above just proved this
            // is the owner's own desktop session, never a sandbox or a CI runner.
            let _trust_guard = trust_store::install_for_tests(trust_store::platform());

            let rt = tokio::runtime::Runtime::new().expect("tokio runtime");
            rt.block_on(async {
                let host = "marketplace.visualstudio.com";
                let wiring = Arc::new(WiringState::default());
                // WITHOUT `with_explicit_upstreams`: the tunnel's upstream is the CONNECT
                // authority (01, D-12), never a configured one.
                let state = Arc::new(
                    ModelRelayState::new(
                        reqwest::Url::parse(&format!("https://{host}")).unwrap(),
                        0,
                        8,
                        &[],
                    )
                    .with_wiring(wiring.clone())
                    .with_intercept(wiring.intercept())
                    .with_refusals(wiring.refusals()),
                );
                let port = serve_ephemeral(state).await;

                let binding: Arc<dyn crate::hooks::binding::AgentBinding> = Arc::new(FakeBinding {
                    agent_type: "vscode-acceptance",
                    model_relay_wiring: Some(proxy_env_wiring(
                        &["marketplace.visualstudio.com"],
                        &[ProxyDelivery::SettingsKey {
                            file: acceptance_settings_file,
                            key: "http.proxy",
                        }],
                    )),
                    ..Default::default()
                });
                let agents = vec![crate::hooks::DetectedAgent {
                    kind: crate::hooks::AgentKind::ClaudeCode,
                    binding: binding.clone(),
                }];
                let cfg = proxy_env_test_config("agt_owner_acceptance");

                let dir = crate::model_relay::ca::ca_dir(&crate::config::openlatch_dir());
                let sha = crate::model_relay::ca::LocalCa::load_or_generate(&dir)
                    .expect("load or generate the CA")
                    .sha256_hex();

                // A-1: install.
                match ensure_proxy_env_trust(&cfg, &agents, true) {
                    Some(InstallOutcome::Installed) | Some(InstallOutcome::AlreadyTrusted) => {}
                    other => panic!("A-1: expected Installed or AlreadyTrusted, got {other:?}"),
                }
                assert!(
                    trust_store::store().is_trusted(&sha).expect("is_trusted"),
                    "A-1 read-back"
                );
                println!("A-1: the relay CA is trusted in the login keychain");

                // A-2: prove and write.
                let formats = crate::cloud::worker::SourceFormats::new();
                wire_agents(agents, port, false, &cfg, &wiring, &formats).await;
                assert_eq!(
                    wiring.verdict("vscode-acceptance"),
                    Verdict::Ok,
                    "A-2: the tick must prove green"
                );
                let written = std::fs::read_to_string(&settings_path).expect("read settings after A-2");
                assert!(
                    written.contains(&format!("\"http.proxy\": \"http://127.0.0.1:{port}\"")),
                    "A-2: the settings file must now carry our proxy URL"
                );
                println!("A-2: {} now names http://127.0.0.1:{port}", settings_path.display());

                // Step 3: launch VS Code and hold for the manual check.
                let hold_secs: u64 = std::env::var("OPENLATCH_ACCEPT_HOLD_SECS")
                    .ok()
                    .and_then(|s| s.parse().ok())
                    .unwrap_or(180);
                println!(
                    "Launch VS Code now: code --user-data-dir \"$X\" --extensions-dir \"$Y\" --new-window"
                );
                println!(
                    "Open the Extensions view and search. Holding {hold_secs}s (relay log with \
                     RUST_LOG=info --nocapture should show intercepted CONNECTs for {host} with \
                     completed handshakes and no \"tls handshake eof\")."
                );
                std::thread::sleep(std::time::Duration::from_secs(hold_secs));

                // A-3: the D-26 live-pickup check.
                release_proxy_delivery(&*binding, ReleasedBy::Teardown);
                let after_release =
                    std::fs::read_to_string(&settings_path).expect("read settings after A-3");
                assert_eq!(
                    after_release, snapshot,
                    "A-3: release must restore the pre-test bytes exactly"
                );
                println!(
                    "A-3: released at {:?}. Holding 60s more with the relay still up — search \
                     again WITHOUT reloading and record whether new CONNECT lines keep arriving \
                     (reload needed) or not (live pickup).",
                    std::time::SystemTime::now()
                );
                std::thread::sleep(std::time::Duration::from_secs(60));

                // A-4: removal read-back.
                assert!(
                    trust_store::store().remove(&sha).expect("remove"),
                    "A-4: remove"
                );
                assert!(
                    !trust_store::store().is_trusted(&sha).expect("is_trusted"),
                    "A-4: read-back"
                );
                println!("A-4: the relay CA has been removed from the login keychain");
            });

            match previous_dir {
                Some(v) => std::env::set_var("OPENLATCH_DIR", v),
                None => std::env::remove_var("OPENLATCH_DIR"),
            }
        }

        // ---------------------------------------------------------------
        // I-4 plan 05 §5 — no format, a proof that survives an unchanged
        // tick, the static ∪ dynamic host set, store-first, no backoff on
        // a CA refusal, and a tripped dynamic host.
        // ---------------------------------------------------------------

        impl ProxyEnvFixture {
            /// `new()`'s twin with the relay's explicit Anthropic upstream at `upstream` (a
            /// closed port, for the store-first tests).
            async fn with_upstream(upstream: u16) -> Self {
                Self::build(upstream).await
            }
        }

        /// A temp Cline lane holding `openAiBaseUrl: <url>`, and the endpoints reading it,
        /// leaked (a binding holds `&'static dyn ProviderEndpoints`).
        fn lane_endpoints(
            dir: &std::path::Path,
            url: &str,
        ) -> (
            std::path::PathBuf,
            &'static dyn crate::hooks::provider_endpoints::ProviderEndpoints,
        ) {
            let path = dir.join("globalState.json");
            write_lane(&path, url);
            let lanes = crate::hooks::cline_providers::state_lanes_from(
                Some(path.clone()),
                Some(path.clone()),
            );
            let ep: &'static crate::hooks::cline_providers::ClineProviderEndpoints =
                Box::leak(Box::new(
                    crate::hooks::cline_providers::ClineProviderEndpoints::at(lanes, None),
                ));
            (
                path,
                ep as &'static dyn crate::hooks::provider_endpoints::ProviderEndpoints,
            )
        }

        fn write_lane(path: &std::path::Path, url: &str) {
            std::fs::write(path, format!("{{\"openAiBaseUrl\": \"{url}\"}}\n")).expect("lane");
        }

        fn agent_with_endpoints(
            hosts: &'static [&'static str],
            delivery: &'static [ProxyDelivery],
            endpoints: &'static dyn crate::hooks::provider_endpoints::ProviderEndpoints,
        ) -> crate::hooks::DetectedAgent {
            crate::hooks::DetectedAgent {
                kind: crate::hooks::AgentKind::ClaudeCode,
                binding: Arc::new(FakeBinding {
                    agent_type: AGENT,
                    display_name: "Fake",
                    config_dir: crate::config::openlatch_dir().join("fake-agent"),
                    model_relay_wiring: Some(proxy_env_wiring(hosts, delivery)),
                    provider_endpoints: Some(endpoints),
                    ..Default::default()
                }),
            }
        }

        fn interceptor_of(wiring: &WiringState) -> Arc<crate::model_relay::ca::Interceptor> {
            wiring
                .intercept()
                .read()
                .expect("slot")
                .clone()
                .expect("the tick filled the interceptor")
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn a_proxy_env_agent_records_no_wire_format() {
            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(true, Ordering::SeqCst);
            fx.wiring.set_wired("claude-code", true);
            fx.wiring
                .set_wired_format("claude-code", WireFormat::AnthropicMessages);
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();

            wire_agents(
                vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;

            assert_eq!(
                fx.wiring.verdict(AGENT),
                Verdict::Ok,
                "premise: a green tick"
            );
            assert!(fx.wiring.is_wired(AGENT));
            assert_eq!(fx.wiring.wired_format(AGENT), None);
            assert_eq!(
                fx.wiring
                    .sole_wired_agent_for(WireFormat::AnthropicMessages),
                Some("claude-code")
            );
            assert!(!fx.wiring.intercept_proven(AGENT));
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn an_unchanged_tick_keeps_the_intercept_proof() {
            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(true, Ordering::SeqCst);
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();
            async fn tick(
                fx: &ProxyEnvFixture,
                cfg: &Config,
                formats: &crate::cloud::worker::SourceFormats,
                broke: bool,
            ) {
                wire_agents(
                    vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                    fx.port,
                    broke,
                    cfg,
                    &fx.wiring,
                    formats,
                )
                .await;
            }

            tick(&fx, &cfg, &formats, false).await;
            assert!(fx.wiring.is_wired(AGENT), "premise: tick 1 is green");
            fx.wiring.note_intercepted(HOST);
            assert!(fx.wiring.intercept_proven(AGENT));
            let written = std::fs::read(settings_path()).expect("settings");

            tick(&fx, &cfg, &formats, false).await;
            assert_eq!(std::fs::read(settings_path()).expect("settings"), written);
            assert!(fx.wiring.intercept_proven(AGENT), "tick 2: nothing changed");

            tick(&fx, &cfg, &formats, true).await;
            assert_eq!(fx.wiring.verdict(AGENT), Verdict::Ok);
            assert!(
                fx.wiring.intercept_proven(AGENT),
                "tick 3: a green re-probe of a wired agent writes nothing and keeps the proof"
            );

            fx.wiring.set_wired(AGENT, false);
            tick(&fx, &cfg, &formats, false).await;
            assert!(
                fx.wiring.is_wired(AGENT),
                "tick 4 re-establishes the wiring"
            );
            assert!(
                !fx.wiring.intercept_proven(AGENT),
                "a new wiring starts unproven"
            );
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn the_tick_publishes_static_and_dynamic_hosts() {
            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(true, Ordering::SeqCst);
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();
            let lane_dir = tempfile::tempdir().expect("lane");
            let (lane, ep) = lane_endpoints(lane_dir.path(), "https://qwen.gpu.test/v1");
            let agent = agent_with_endpoints(&[HOST], FAKE_DELIVERY, ep);

            wire_agents(
                vec![agent.clone()],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;
            assert_eq!(fx.wiring.sole_interceptor_for(HOST), Some(AGENT));
            assert_eq!(fx.wiring.sole_interceptor_for("qwen.gpu.test"), Some(AGENT));
            assert!(interceptor_of(&fx.wiring).intercepts("qwen.gpu.test"));

            // The file-trigger path: the customer moved the slot to a loopback URL.
            write_lane(&lane, "http://127.0.0.1:8123/v1");
            publish_intercept_hosts(&fx.wiring, std::slice::from_ref(&agent));
            assert_eq!(fx.wiring.sole_interceptor_for("qwen.gpu.test"), None);
            assert!(!interceptor_of(&fx.wiring).intercepts("qwen.gpu.test"));
            assert_eq!(fx.wiring.sole_interceptor_for(HOST), Some(AGENT));
            assert!(interceptor_of(&fx.wiring).intercepts(HOST));

            wire_agents(
                vec![env_vars_agent("envvars-agent")],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;
            assert_eq!(fx.wiring.sole_interceptor_for(HOST), None);
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn the_store_is_asked_before_any_network_probe() {
            let dead = mock::closed_port().await;
            let fx = ProxyEnvFixture::with_upstream(dead).await;
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();

            fx.fake_store.verify.store(false, Ordering::SeqCst);
            wire_agents(
                vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;
            match fx.wiring.verdict(AGENT) {
                Verdict::Failed(reason) => {
                    assert!(
                        reason.starts_with(crate::model_relay::preflight::CA_REASON_MARKER),
                        "{reason}"
                    );
                    assert!(
                        reason.contains("does not trust the relay's CA"),
                        "the store answered first, before any probe: {reason}"
                    );
                }
                other => panic!("expected Failed, got {other:?}"),
            }

            // Control: a trusting store lets the probe run, and the probe names the dead upstream.
            fx.fake_store.verify.store(true, Ordering::SeqCst);
            wire_agents(
                vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;
            match fx.wiring.verdict(AGENT) {
                Verdict::Failed(reason) => {
                    assert!(
                        !reason.contains("does not trust"),
                        "the store is not the failure: {reason}"
                    );
                    assert!(
                        !reason.starts_with(crate::model_relay::preflight::CA_REASON_MARKER),
                        "an unreachable upstream is not a CA refusal: {reason}"
                    );
                }
                other => panic!("expected Failed, got {other:?}"),
            }
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn a_store_refusal_is_not_a_network_failure() {
            let dead = mock::closed_port().await;
            let fx = ProxyEnvFixture::with_upstream(dead).await;
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();

            fx.fake_store.verify.store(false, Ordering::SeqCst);
            let failed = wire_agents(
                vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;
            assert!(!failed, "a CA refusal is local: no backoff");

            fx.fake_store.verify.store(true, Ordering::SeqCst);
            let failed = wire_agents(
                vec![fx.agent(&[HOST], FAKE_DELIVERY)],
                fx.port,
                false,
                &cfg,
                &fx.wiring,
                &formats,
            )
            .await;
            assert!(failed, "control: a dead upstream still backs off");
        }

        fn fake_b_settings_file() -> Option<std::path::PathBuf> {
            Some(
                crate::config::openlatch_dir()
                    .join("fake-b")
                    .join("settings.json"),
            )
        }

        #[tokio::test(flavor = "multi_thread")]
        async fn a_tripped_dynamic_host_releases_its_agent() {
            use crate::model_relay::ca_lifecycle::refused_reason;
            use crate::model_relay::intercept::RefusalKind;
            const B_DELIVERY: &[ProxyDelivery] = &[ProxyDelivery::SettingsKey {
                file: fake_b_settings_file,
                key: KEY,
            }];

            let fx = ProxyEnvFixture::new().await;
            fx.fake_store.verify.store(true, Ordering::SeqCst);
            let cfg = fx.config();
            let formats = crate::cloud::worker::SourceFormats::new();
            let lane_dir = tempfile::tempdir().expect("lane");
            let (_lane, ep) = lane_endpoints(lane_dir.path(), "https://qwen.gpu.test/v1");
            let a = agent_with_endpoints(&[HOST], FAKE_DELIVERY, ep);
            let b_file = fake_b_settings_file().expect("path");
            std::fs::create_dir_all(b_file.parent().expect("parent")).expect("mkdir");
            std::fs::write(&b_file, SEED).expect("seed b");
            let b = crate::hooks::DetectedAgent {
                kind: crate::hooks::AgentKind::ClaudeCode,
                binding: Arc::new(FakeBinding {
                    agent_type: "fake-b",
                    display_name: "Fake B",
                    config_dir: crate::config::openlatch_dir().join("fake-b"),
                    model_relay_wiring: Some(proxy_env_wiring(&["b.test"], B_DELIVERY)),
                    ..Default::default()
                }),
            };
            let agents = vec![a, b];

            wire_agents(agents.clone(), fx.port, false, &cfg, &fx.wiring, &formats).await;
            assert!(fx.wiring.is_wired(AGENT), "premise: fake-a is wired");
            assert!(fx.wiring.is_wired("fake-b"), "premise: fake-b is wired");
            assert!(env_sh_exists());

            let refusals = fx.wiring.refusals();
            for _ in 0..3 {
                refusals.record_refusal("qwen.gpu.test", RefusalKind::HandshakeEof);
            }
            let mut holds = std::collections::BTreeMap::new();
            let at = TrustClock {
                wall: time::OffsetDateTime::now_utc(),
                mono: std::time::Instant::now(),
            };
            let skip =
                react_to_trust_loss(&cfg, &agents, None, at, &refusals, &mut holds, &fx.wiring);

            assert!(skip.contains(AGENT), "the dynamic host names fake-a");
            assert!(!fx.wiring.is_wired(AGENT));
            assert_eq!(
                fx.wiring.verdict(AGENT),
                Verdict::Failed(refused_reason("qwen.gpu.test"))
            );
            assert_eq!(std::fs::read_to_string(settings_path()).unwrap(), SEED);
            assert!(!env_sh_exists());

            assert!(!skip.contains("fake-b"));
            assert!(fx.wiring.is_wired("fake-b"), "fake-b is untouched");
        }
    }
}