openlatch-client 0.6.1

//! Model relay preflight — proving the forwarder actually *forwards* before any
//! agent is pointed at it, and un-pointing them the moment it stops.
//!
//! Binding the pinned port proves only that the port is held. It says nothing
//! about the leg that actually breaks in the field: loopback → axum → the
//! observe/transform stage → reqwest → TLS → `api.anthropic.com` → back. A
//! model relay that binds and then cannot reach upstream — captive portal, VPN not
//! up yet, corporate TLS interception, a regression in the forward path — is
//! indistinguishable from a healthy one from the agent's side, and every Claude
//! Code session on the machine dies on it, because `ANTHROPIC_BASE_URL` is set.
//!
//! So the wiring hangs off a *round trip*, not off a bind:
//!
//! > **Gate.** `ANTHROPIC_BASE_URL` is written only after a synthetic request
//! > has travelled the full path through our own listener and come back with an
//! > answer that provably originated upstream.
//!
//! ## Why an unauthenticated request is the right probe
//!
//! [`probe`] sends a deliberately credential-less `POST /v1/messages`. Anthropic
//! answers `401`. That is the **success** case: the question is not "did the
//! call succeed" but "did an *upstream* response come back at all". OpenLatch
//! has no provider credential of its own — the model relay forwards the caller's
//! verbatim — so a probe that required one would be unrunnable at daemon start,
//! and a probe that spent tokens would bill the customer for our health check.
//! A 401 costs nothing, needs no key, and still exercises every hop.
//!
//! The one response that must NOT open the gate is the model relay's own synthetic
//! 502 (`proxy::synth_502`, C-5b) — which is exactly what an unreachable
//! upstream produces. It carries `x-openlatch-upstream: unreachable`, so the two
//! are told apart by header rather than by status code: a real upstream 502
//! still proves a live path, because it came from upstream.
//!
//! ## Why the probe carries a marker header
//!
//! [`PREFLIGHT_HEADER`] marks the request as ours. The observe/transform stage
//! still runs on it — that stage is where bugs live, and a panic there is worth
//! surfacing — but the resulting observation is dropped instead of being
//! promoted to an economics event. Our health check is not the customer's
//! traffic and must never land on their bill or in their usage data. The header
//! is stripped before the request leaves for upstream.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::Mutex;
use std::time::Duration;

use super::wire_format::WireFormat;

/// Marks a request as OpenLatch's own preflight probe.
///
/// Read in exactly two places: `proxy::proxy_any` drops the observation so the
/// probe never reaches the economics rail, and `proxy::forward_headers` strips
/// it so it never reaches the provider.
pub const PREFLIGHT_HEADER: &str = "x-openlatch-preflight";

/// The header the model relay stamps on its synthetic 502 when it could not reach
/// upstream at all (`proxy::synth_502`). Its presence is the single signal that
/// separates "our forwarder answered *for* the upstream" from "the upstream
/// answered".
const UPSTREAM_UNREACHABLE_HEADER: &str = "x-openlatch-upstream";

/// Total budget for one probe.
///
/// Deliberately far shorter than the forward path's own `HEADER_TIMEOUT` (60 s):
/// that budget is generous because a slow first token is legitimate on a real
/// turn, whereas an unauthenticated request is rejected at the provider's edge
/// and comes back in well under a second. A daemon start must not stall on a
/// silent upstream, and the supervisor's retry loop makes a tight budget safe —
/// a false negative costs one tick, not the wiring.
pub const PREFLIGHT_TIMEOUT: Duration = Duration::from_secs(5);

/// A credential-less `/v1/messages` body. `max_tokens: 1` so that even a
/// hypothetical future in which this request DID authenticate could not spend
/// meaningfully; as written it is rejected before a model is ever loaded.
const PREFLIGHT_BODY: &str =
    r#"{"model":"claude-sonnet-4-5","max_tokens":1,"messages":[{"role":"user","content":"ping"}]}"#;

/// The same claim in the OpenAI Responses shape — the minimal valid request,
/// with `max_output_tokens` playing `max_tokens`' role.
///
/// A probe that spoke `/v1/messages` at a Codex plane would prove a **Claude
/// Code** round trip and then open the gate for Codex: a disabled subsystem
/// rendering as healthy, which is the failure the gate exists to stop (PRD
/// D-13).
const PREFLIGHT_BODY_RESPONSES: &str =
    r#"{"model":"gpt-5-codex","input":"preflight","max_output_tokens":16}"#;

/// The same claim in the OpenAI **chat completions** shape.
///
/// A different route AND a different body from [`PREFLIGHT_BODY_RESPONSES`]:
/// `/v1/responses` takes `input` + `max_output_tokens`, `/v1/chat/completions`
/// takes `messages` + `max_tokens`, and an endpoint handed the wrong one
/// answers 400 before it ever reaches the leg this probe exists to exercise.
///
/// The model name is a placeholder that no on-prem gateway is obliged to know.
/// That is fine and is the point of the whole design: the question is *"did an
/// upstream response come back at all"*, and an unknown-model rejection is as
/// good a proof of a live path as a 401.
const PREFLIGHT_BODY_CHAT_COMPLETIONS: &str = r#"{"model":"openlatch-preflight","max_tokens":1,"messages":[{"role":"user","content":"ping"}]}"#;

/// The same claim in Google's Generative Language shape.
///
/// `contents`, not `messages`, and no token cap: `generationConfig` is optional
/// and every field in it is one more thing a Vertex deployment can reject for
/// its own reasons.
const PREFLIGHT_BODY_GENERATE_CONTENT: &str =
    r#"{"contents":[{"role":"user","parts":[{"text":"ping"}]}]}"#;

/// The same claim in Ollama's native chat shape. An empty conversation with
/// `stream: false`: a server without the placeholder model answers a 404 at
/// once, which proves the path as well as a reply would, and loads nothing.
const PREFLIGHT_BODY_OLLAMA_NATIVE: &str =
    r#"{"model":"openlatch-preflight","messages":[],"stream":false}"#;

/// The model segment Google's probe route carries.
///
/// **The route MUST have one, and MUST keep the colon suffix.**
/// `WireFormat::resolve` matches Google on the `:generateContent` /
/// `:streamGenerateContent` suffix alone, so a probe posted to a bare
/// `/v1beta/models` resolves `Unknown` — whose upstream is Anthropic's — and the
/// preflight becomes an instance of the very misroute the gate it opens exists
/// to prevent.
const PREFLIGHT_GOOGLE_MODEL: &str = "openlatch-preflight";

/// The value sent as `x-api-key`. Not a credential and not a redacted one — a
/// literal that cannot be mistaken for either in a log or a capture.
const PREFLIGHT_API_KEY: &str = "ol-preflight-not-a-key";

/// The outcome of the most recent probe.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub enum Verdict {
    /// No probe has completed yet — the daemon is up but the gate has not run.
    /// Distinct from `Failed` on purpose: `init` waits `Pending` out, and a
    /// caller that collapsed the two would report a healthy install as broken
    /// for the first second of its life.
    #[default]
    Pending,
    /// A response provably originating upstream came back through our listener.
    Ok,
    /// The round trip did not complete. Carries the reason, surfaced verbatim by
    /// `init` and `doctor` — a preflight failure the operator cannot act on is
    /// barely better than no check at all.
    Failed(String),
}

impl Verdict {
    /// Stable machine-readable label for the admin surface and `--json`.
    pub fn label(&self) -> &'static str {
        match self {
            Self::Pending => "pending",
            Self::Ok => "ok",
            Self::Failed(_) => "failed",
        }
    }

    /// The failure reason, when there is one.
    pub fn error(&self) -> Option<&str> {
        match self {
            Self::Failed(e) => Some(e.as_str()),
            _ => None,
        }
    }

    /// Whether the gate is open.
    pub fn is_ok(&self) -> bool {
        matches!(self, Self::Ok)
    }
}

/// Process-wide view of the wiring gate, **keyed per agent**, shared by the
/// three parties that must agree on it: the supervisor that opens and closes
/// it, the admin status endpoint that reports it, and — through that endpoint —
/// `init` and `doctor`.
///
/// Deliberately NOT a field on the per-attempt `ModelRelayState`: a model relay
/// restart rebuilds that struct, and the verdict has to survive one.
///
/// **Both fields are per agent, and neither may be left process-wide.** Two
/// agents on one host point at the same listener through different conventions
/// and are probed in different formats, so one can be wired while the other's
/// round trip fails. A single flag would report one plane's verdict for both:
/// `init` would pass on a host whose Codex plane never came up, and the failing
/// agent's own unwire would take the healthy one's wiring with it.
///
/// A missing key is `Pending` / not wired — the state an agent is in between
/// the supervisor seeding it and its first probe returning.
#[derive(Debug, Default)]
pub struct WiringState {
    wired: Mutex<BTreeMap<&'static str, bool>>,
    verdict: Mutex<BTreeMap<&'static str, Verdict>>,
    /// The format each agent's request plane was wired to speak, from its own
    /// [`ModelRelayWiring`](crate::hooks::binding::ModelRelayWiring).
    ///
    /// Recorded beside the flag rather than derived, because it is a fact about
    /// THIS install's configuration — the `wire_api` we wrote into Codex's
    /// `config.toml`, the `ANTHROPIC_BASE_URL` we set for Claude Code — and not
    /// a fixed agent→protocol table, which PRD D-7 says does not exist.
    format: Mutex<BTreeMap<&'static str, WireFormat>>,
    /// The wiring pass's verdict for each provider slot served by a relay
    /// endpoint, keyed by slot key.
    ///
    /// **A separate map, never folded into the per-agent ones above.**
    /// [`Self::sole_wired_agent_for`] reads `format` and `wired`; an endpoint
    /// counted there would make a Cline Anthropic slot a second speaker of
    /// Anthropic Messages, and every Claude Code turn on the main port would
    /// attribute to `unknown`. Only slots with something to say have an entry:
    /// a slot with no entry is judged from its record and its traffic.
    endpoints: std::sync::Mutex<BTreeMap<String, EndpointVerdict>>,
    /// The intercepting half's slot (plan 01's `InterceptSlot`), filled by the wiring supervisor from
    /// the detected ProxyEnv bindings and served from by the relay. Here for the reason this struct
    /// exists: a serve restart must not lose it.
    intercept: InterceptCell,
    /// Plan 01's D-32 refusal ledger, owned here for the same reason as the slot: the relay records
    /// into it and plan 04's supervisor reaction reads it through `refusals()`, so it must be ONE
    /// ledger that survives a serve restart (and `run_wiring_supervisor` keeps its 7 parameters).
    refusals: RefusalsCell,
    /// D-41: every DETECTED ProxyEnv binding → every host it intercepts, static AND dynamic
    /// (daemon `proxy_env_declarations`, lowercase), replaced by each publication (§5.1).
    /// Never read by `sole_wired_agent_for`: an intercepted agent is attributed by host, not format.
    interceptors: Mutex<BTreeMap<&'static str, BTreeSet<String>>>,
    /// D-42: whether a decrypted, non-preflight request on one of the agent's hosts has arrived since
    /// its current wiring. Cleared when the ProxyEnv wiring is new (§5.5), never by an unchanged tick.
    intercepted: Mutex<BTreeMap<&'static str, bool>>,
}

#[derive(Default)]
struct InterceptCell(crate::model_relay::InterceptSlot);
impl std::fmt::Debug for InterceptCell {
    // `WiringState` derives Debug; the Interceptor need not.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let filled = self.0.read().map(|g| g.is_some()).unwrap_or(false);
        f.debug_struct("InterceptCell")
            .field("filled", &filled)
            .finish()
    }
}

#[derive(Default)]
struct RefusalsCell(std::sync::Arc<crate::model_relay::intercept::RefusalLedger>);
impl std::fmt::Debug for RefusalsCell {
    // 01's RefusalLedger derives Default only, not Debug.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RefusalsCell").finish_non_exhaustive()
    }
}

/// Why the wiring pass could not wire, or keep wired, one provider slot.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EndpointVerdict {
    /// The `OL-RELAY-*` code.
    pub code: &'static str,
    /// What happened, for the report.
    pub detail: String,
}

/// `host`, ASCII-lowercased at most once: the relay's CONNECT authority already is
/// (`intercept::handle_connect`), so the common case borrows.
fn lowercase_host(host: &str) -> std::borrow::Cow<'_, str> {
    if host.bytes().any(|b| b.is_ascii_uppercase()) {
        std::borrow::Cow::Owned(host.to_ascii_lowercase())
    } else {
        std::borrow::Cow::Borrowed(host)
    }
}

impl WiringState {
    /// Read a map, degrading a poisoned lock to the value it was holding.
    ///
    /// This is read from the admin handler on the model relay's own runtime, and
    /// the gate's observability must never be able to take the listener down.
    fn read<T: Clone>(m: &Mutex<BTreeMap<&'static str, T>>) -> BTreeMap<&'static str, T> {
        match m.lock() {
            Ok(v) => v.clone(),
            Err(poisoned) => poisoned.into_inner().clone(),
        }
    }

    /// Mutate a map under the same poison-tolerant rule.
    fn write<T>(
        m: &Mutex<BTreeMap<&'static str, T>>,
        f: impl FnOnce(&mut BTreeMap<&'static str, T>),
    ) {
        match m.lock() {
            Ok(mut v) => f(&mut v),
            Err(poisoned) => f(&mut poisoned.into_inner()),
        }
    }

    /// Whether `agent`'s config currently points at this listener.
    pub fn is_wired(&self, agent: &str) -> bool {
        match self.wired.lock() {
            Ok(v) => v.get(agent).copied().unwrap_or(false),
            Err(poisoned) => poisoned.into_inner().get(agent).copied().unwrap_or(false),
        }
    }

    /// Record `agent`'s wiring state after a successful write / removal.
    pub fn set_wired(&self, agent: &'static str, wired: bool) {
        Self::write(&self.wired, |m| {
            m.insert(agent, wired);
        });
    }

    /// Record the format `agent`'s request plane was wired to speak.
    ///
    /// Separate from [`set_wired`](Self::set_wired) so an unwire clears the flag
    /// without erasing what the agent speaks — the entry is inert while the flag
    /// is false, and correct again the moment it is re-wired.
    pub fn set_wired_format(&self, agent: &'static str, format: WireFormat) {
        Self::write(&self.format, |m| {
            m.insert(agent, format);
        });
    }

    /// The format `agent` was last wired to speak, if it has ever been wired.
    ///
    /// The wiring supervisor compares this against the format the binding
    /// resolves NOW: a GUI-hosted agent's format follows the provider the
    /// customer is on, and a plane still wired for the provider they left is
    /// pointed at a route the relay no longer captures. That comparison is the
    /// only thing that notices, because the listener stays healthy throughout.
    pub fn wired_format(&self, agent: &str) -> Option<WireFormat> {
        match self.format.lock() {
            Ok(v) => v.get(agent).copied(),
            Err(poisoned) => poisoned.into_inner().get(agent).copied(),
        }
    }

    /// The single wired agent speaking `format`, when there is exactly one.
    ///
    /// This is the whole of the deduction, and the counting is the point.
    ///
    /// The model relay resolves a request's format from its ROUTE and never from
    /// agent identity (PRD D-7), so this cannot run backwards into "an Anthropic
    /// request means Claude Code". What it may say is narrower and checkable:
    /// *on this install, only one wired agent was configured to speak this
    /// protocol, so a request in it came from that agent.* Two speakers, or
    /// none, and there is no such sentence to say — the caller keeps `unknown`.
    ///
    /// `Unknown` is never attributable: it is the catch-all for every route the
    /// model relay does not capture, so "the sole agent speaking unknown" would be
    /// an answer about the routes rather than about an agent.
    pub fn sole_wired_agent_for(&self, format: WireFormat) -> Option<&'static str> {
        if !format.is_captured() {
            return None;
        }
        let wired = Self::read(&self.wired);
        let mut hit = None;
        for (agent, agent_format) in Self::read(&self.format) {
            if agent_format != format || !wired.get(agent).copied().unwrap_or(false) {
                continue;
            }
            if hit.is_some() {
                // A second speaker. Naming either would be a coin flip wearing
                // the platform's `(org, source, agent_id)` join key.
                return None;
            }
            hit = Some(agent);
        }
        hit
    }

    /// Record `key`'s endpoint verdict, or clear it with `None`.
    pub fn set_endpoint_verdict(&self, key: &str, verdict: Option<EndpointVerdict>) {
        let mut map = match self.endpoints.lock() {
            Ok(m) => m,
            Err(poisoned) => poisoned.into_inner(),
        };
        match verdict {
            Some(v) => {
                map.insert(key.to_string(), v);
            }
            None => {
                map.remove(key);
            }
        }
    }

    /// Every endpoint verdict, for the admin surface.
    pub fn endpoint_verdicts(&self) -> BTreeMap<String, EndpointVerdict> {
        match self.endpoints.lock() {
            Ok(m) => m.clone(),
            Err(poisoned) => poisoned.into_inner().clone(),
        }
    }

    /// The most recent probe verdict for `agent`.
    ///
    /// `Pending` when the agent has no entry — the honest answer for a plane
    /// the supervisor has not judged yet, and the same value a fresh daemon
    /// reported before the state was keyed.
    pub fn verdict(&self, agent: &str) -> Verdict {
        match self.verdict.lock() {
            Ok(v) => v.get(agent).cloned().unwrap_or_default(),
            Err(poisoned) => poisoned
                .into_inner()
                .get(agent)
                .cloned()
                .unwrap_or_default(),
        }
    }

    /// Store a fresh probe verdict for `agent`.
    pub fn set_verdict(&self, agent: &'static str, verdict: Verdict) {
        Self::write(&self.verdict, |m| {
            m.insert(agent, verdict);
        });
    }

    /// Every verdict, for the admin surface to render one entry per agent.
    pub fn verdicts(&self) -> BTreeMap<&'static str, Verdict> {
        Self::read(&self.verdict)
    }

    /// Every wiring flag, for the same reason.
    ///
    /// A separate snapshot from [`verdicts`](Self::verdicts) because the two
    /// facts differ by design: the verdict goes `Ok` when the probe returns,
    /// the flag goes true only once the agent's file was actually written.
    /// Collapsing them would report a plane as wired on the strength of a green
    /// probe whose write then failed.
    pub fn wired_agents(&self) -> BTreeMap<&'static str, bool> {
        Self::read(&self.wired)
    }

    /// Give `agent` its `Pending` / not-wired entry, unless it already has one.
    ///
    /// The supervisor seeds every agent that has a request plane before its
    /// first probe. Without it an unprobed agent has no key at all, and every
    /// consumer of the status JSON — `classify_model_relay`, `init`'s wait rule,
    /// `doctor` — falls through to its "some other reason" arm and renders a
    /// host that is merely still checking as one that is unwired.
    ///
    /// Never clobbers: a supervised restart re-seeds, and an agent already
    /// judged must keep its verdict.
    pub fn seed(&self, agent: &'static str) {
        Self::write(&self.wired, |m| {
            m.entry(agent).or_insert(false);
        });
        Self::write(&self.verdict, |m| {
            m.entry(agent).or_default();
        });
    }

    /// The shared intercept slot. Cloning the Arc — the relay and the supervisor hold the same one.
    pub fn intercept(&self) -> crate::model_relay::InterceptSlot {
        self.intercept.0.clone()
    }

    /// The shared refusal ledger (INDEX row "daemon + hooks"). Cloning the Arc, as [`Self::intercept`].
    pub fn refusals(&self) -> std::sync::Arc<crate::model_relay::intercept::RefusalLedger> {
        self.refusals.0.clone()
    }

    /// Replace every declarer's host set (I-4 §5.1's `publish_intercept_hosts`, on every wiring pass).
    pub fn set_interceptors(&self, declared: BTreeMap<&'static str, BTreeSet<String>>) {
        match self.interceptors.lock() {
            Ok(mut m) => *m = declared,
            Err(poisoned) => *poisoned.into_inner() = declared,
        }
    }

    /// The one detected ProxyEnv binding whose static or dynamic hosts hold `host` (compared
    /// lowercased); `None` for none or two — the `sole_wired_agent_for` rule, on hosts. A customer-set
    /// remote host is attributed exactly like a stock one. Scanned under the guard: no copy of the
    /// map per request.
    pub fn sole_interceptor_for(&self, host: &str) -> Option<&'static str> {
        let host = lowercase_host(host);
        let scan = |m: &BTreeMap<&'static str, BTreeSet<String>>| {
            let mut hit = None;
            for (agent, hosts) in m {
                if !hosts.contains(host.as_ref()) {
                    continue;
                }
                if hit.is_some() {
                    return None;
                }
                hit = Some(*agent);
            }
            hit
        };
        match self.interceptors.lock() {
            Ok(m) => scan(&m),
            Err(poisoned) => scan(&poisoned.into_inner()),
        }
    }

    /// Whether ANY detected ProxyEnv binding declares `host` (compared lowercased) — the set
    /// [`Self::sole_interceptor_for`] scans, without its "exactly one" rule. The owner's Amendment 2
    /// (2026-09-24) gate: a plain-http absolute-form request for such a host is measured.
    pub fn is_intercept_host(&self, host: &str) -> bool {
        let host = lowercase_host(host);
        let any = |m: &BTreeMap<&'static str, BTreeSet<String>>| {
            m.values().any(|hosts| hosts.contains(host.as_ref()))
        };
        match self.interceptors.lock() {
            Ok(m) => any(&m),
            Err(poisoned) => any(&poisoned.into_inner()),
        }
    }

    /// Whether the last publication kept `agent` (the file-trigger pass's filter, §5.1).
    pub fn declares_interceptor(&self, agent: &str) -> bool {
        match self.interceptors.lock() {
            Ok(m) => m.contains_key(agent),
            Err(poisoned) => poisoned.into_inner().contains_key(agent),
        }
    }

    /// Every declarer → its hosts, for the admin surface (§6.2) and the live proof.
    pub fn intercept_hosts(&self) -> BTreeMap<&'static str, BTreeSet<String>> {
        match self.interceptors.lock() {
            Ok(m) => m.clone(),
            Err(poisoned) => poisoned.into_inner().clone(),
        }
    }

    /// Marks `true` for EVERY declarer of `host` (two declarers are both proven: the request did use
    /// the relay, whoever sent it — D-44 says so to the operator).
    pub fn note_intercepted(&self, host: &str) {
        let host = lowercase_host(host);
        // Every request after the first: its one declarer is already proven, nothing to write.
        if self
            .sole_interceptor_for(&host)
            .is_some_and(|agent| self.intercept_proven(agent))
        {
            return;
        }
        let host = host.as_ref();
        let declarers: Vec<&'static str> = match self.interceptors.lock() {
            Ok(m) => m
                .iter()
                .filter(|(_, hosts)| hosts.contains(host))
                .map(|(a, _)| *a)
                .collect(),
            Err(poisoned) => poisoned
                .into_inner()
                .iter()
                .filter(|(_, hosts)| hosts.contains(host))
                .map(|(a, _)| *a)
                .collect(),
        };
        match self.intercepted.lock() {
            Ok(mut m) => {
                for agent in declarers {
                    m.insert(agent, true);
                }
            }
            Err(poisoned) => {
                let mut m = poisoned.into_inner();
                for agent in declarers {
                    m.insert(agent, true);
                }
            }
        }
    }

    /// Whether a decrypted, non-preflight request on one of `agent`'s hosts has arrived since its
    /// current wiring. Missing → `false`.
    pub fn intercept_proven(&self, agent: &str) -> bool {
        match self.intercepted.lock() {
            Ok(m) => m.get(agent).copied().unwrap_or(false),
            Err(poisoned) => poisoned.into_inner().get(agent).copied().unwrap_or(false),
        }
    }

    /// D-42: a NEW wiring starts unproven (§5.5).
    pub fn clear_intercept_proof(&self, agent: &'static str) {
        match self.intercepted.lock() {
            Ok(mut m) => {
                m.insert(agent, false);
            }
            Err(poisoned) => {
                poisoned.into_inner().insert(agent, false);
            }
        }
    }

    /// Every declarer → proven?, for the admin surface (keys = the interceptors map's).
    pub fn intercept_proofs(&self) -> BTreeMap<&'static str, bool> {
        let declarers: Vec<&'static str> = match self.interceptors.lock() {
            Ok(m) => m.keys().copied().collect(),
            Err(poisoned) => poisoned.into_inner().keys().copied().collect(),
        };
        let proven = match self.intercepted.lock() {
            Ok(m) => m.clone(),
            Err(poisoned) => poisoned.into_inner().clone(),
        };
        declarers
            .into_iter()
            .map(|a| (a, proven.get(a).copied().unwrap_or(false)))
            .collect()
    }
}

/// The route, credential header and body one format's probe speaks.
///
/// **Widened from a `bool` by this unit, and the widening is the point.** The
/// flag picked between an OpenAI-Responses body and an Anthropic one, and plan
/// 01 landed `false` — Anthropic — for both new formats as an explicit
/// placeholder. Neither is right: a chat-completions endpoint does not accept
/// an Anthropic body, and neither does `…:generateContent`. A probe of the
/// wrong shape either fails against a correct deployment, or — against a
/// permissive gateway — PASSES FOR THE WRONG REASON and opens the wiring gate
/// on a round trip that proved nothing about the plane being wired.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProbeShape {
    /// `POST /v1/messages`, `x-api-key` + `anthropic-version`.
    AnthropicMessages,
    /// `POST /v1/responses`, bearer.
    OpenAiResponses,
    /// `POST /v1/chat/completions`, bearer.
    OpenAiChatCompletions,
    /// `POST /v1beta/models/<model>:generateContent`, `x-goog-api-key`.
    GoogleGenerateContent,
    /// `POST /api/chat`, no credential — a local model server has none.
    OllamaNative,
    /// `GET /`, no credential and no body: whether ANY answer comes back from
    /// the origin, for a provider slot whose protocol no format names. The
    /// relay endpoint forwards every route to its one origin, so this still
    /// proves the leg the slot depends on.
    Reachability,
}

impl ProbeShape {
    /// The loopback path this shape posts to.
    ///
    /// Every one of these must resolve back to its OWN [`WireFormat`] through
    /// [`WireFormat::resolve`] — the relay routes the probe by path exactly as
    /// it routes the agent, so a path that resolves to something else sends the
    /// probe to the wrong upstream. `probe_paths_resolve_to_their_own_format`
    /// is the assertion.
    fn route(self) -> String {
        match self {
            Self::AnthropicMessages => "/v1/messages".to_string(),
            Self::OpenAiResponses => "/v1/responses".to_string(),
            Self::OpenAiChatCompletions => "/v1/chat/completions".to_string(),
            Self::GoogleGenerateContent => {
                format!("/v1beta/models/{PREFLIGHT_GOOGLE_MODEL}:generateContent")
            }
            Self::OllamaNative => "/api/chat".to_string(),
            Self::Reachability => "/".to_string(),
        }
    }

    /// The method this shape sends.
    fn method(self) -> reqwest::Method {
        match self {
            Self::Reachability => reqwest::Method::GET,
            _ => reqwest::Method::POST,
        }
    }

    /// The body, in this format's own shape.
    fn body(self) -> &'static str {
        match self {
            Self::AnthropicMessages => PREFLIGHT_BODY,
            Self::OpenAiResponses => PREFLIGHT_BODY_RESPONSES,
            Self::OpenAiChatCompletions => PREFLIGHT_BODY_CHAT_COMPLETIONS,
            Self::GoogleGenerateContent => PREFLIGHT_BODY_GENERATE_CONTENT,
            Self::OllamaNative => PREFLIGHT_BODY_OLLAMA_NATIVE,
            Self::Reachability => "",
        }
    }

    /// The credential header pair this format's provider reads.
    ///
    /// The VALUE is [`PREFLIGHT_API_KEY`] in every arm — a literal that cannot
    /// be mistaken for a credential or for a redacted one. Only the header NAME
    /// (and, for Anthropic, the extra version header) differs, and it differs
    /// because a provider that does not see its own auth header answers
    /// something other than the clean rejection this probe reads as success.
    fn auth_headers(self) -> &'static [(&'static str, &'static str)] {
        match self {
            Self::AnthropicMessages => &[
                ("anthropic-version", "2023-06-01"),
                ("x-api-key", PREFLIGHT_API_KEY),
            ],
            Self::OpenAiResponses | Self::OpenAiChatCompletions => &[(
                "authorization",
                concat!("Bearer ", "ol-preflight-not-a-key"),
            )],
            Self::GoogleGenerateContent => &[("x-goog-api-key", PREFLIGHT_API_KEY)],
            Self::OllamaNative | Self::Reachability => &[],
        }
    }

    /// Which shape a format's probe takes.
    ///
    /// Every variant named, never a wildcard: a fifth format must be a compile
    /// error here rather than a silent Anthropic probe against a plane that
    /// speaks something else.
    fn of(fmt: WireFormat) -> Self {
        match fmt {
            WireFormat::AnthropicMessages => Self::AnthropicMessages,
            WireFormat::OpenAiResponses => Self::OpenAiResponses,
            WireFormat::OpenAiChatCompletions => Self::OpenAiChatCompletions,
            WireFormat::GoogleGenerateContent => Self::GoogleGenerateContent,
            // Cline's Ollama provider speaks the native API, and its slot is
            // probed through the slot's own endpoint — probing as Anthropic
            // against Ollama would prove nothing.
            WireFormat::OllamaNative => Self::OllamaNative,
            // A provider slot whose protocol no format names. Only ever probed
            // through an endpoint, which forwards every route to its one
            // origin; on the main port `Unknown` goes to Anthropic, and nothing
            // probes the main port in this format.
            WireFormat::Unknown => Self::Reachability,
        }
    }
}

/// Prefixes every reason a CA-trust failure produces, so the renderers name the CA step
/// without a second verdict variant (plan 04, D-11).
pub const CA_REASON_MARKER: &str = "ca-untrusted: ";

/// Prefixes the reason when EVERY delivery of a ProxyEnv binding was skipped for a settings
/// cause (D-33/D-34), so plan 04 renders `OL-RELAY-SETTINGS` from the verdict alone. A PARTIAL skip
/// (some delivery live) is not a verdict: plan 04 renders it per surface through `observe_key`.
pub const SETTINGS_REASON_MARKER: &str = "settings: ";

/// The one phrase a foreign-value skip carries (`"{key}: {SETTINGS_FOREIGN}"`, hooks::proxy_settings).
/// Plan 04's `settings_remedy` matches on it to give D-34's `[proxy]` remedy rather than the syntax one.
pub const SETTINGS_FOREIGN: &str = "holds a proxy OpenLatch did not write";
/// The skip phrase for a `SettingsKey` whose file resolver returns `None` (that surface is not
/// installed here). Plan 04 renders an all-no-file verdict with `Resolution::NothingToDo`.
pub const SETTINGS_NO_FILE: &str = "the agent's settings file is not on this host";
/// The skip phrase when the settings file kept changing under every CAS attempt. Plan 04
/// renders it `Resolution::SelfResolving`: the next wiring pass retries.
pub const SETTINGS_CONTENDED: &str = "the file kept changing under the write";
/// The refusal when the post-splice read-back fails: both `write_proxy_key` and
/// `observe_key` return `Unparseable(SETTINGS_LAYOUT)`. Plan 04 gives it a layout remedy, not the syntax one.
pub const SETTINGS_LAYOUT: &str = "the settings file's layout cannot take the key safely";
/// The ONE expiry phrase: the write-side refusal (`prove_proxy_env`, `CA_REASON_MARKER` + this + date)
/// and plan 04's `ca_lifecycle::expiring_reason` build the same text.
pub const CA_EXPIRING_PREFIX: &str = "certificate authority expires ";

/// Find the first occurrence of `needle` in `haystack`, for the bounded head-read loops below.
/// No new parser: CONNECT and the probe's own response head are read as raw bytes until the
/// blank-line terminator appears.
fn find_subslice(haystack: &[u8], needle: &[u8]) -> Option<usize> {
    haystack
        .windows(needle.len())
        .position(|window| window == needle)
}

/// The numeric status code out of an HTTP/1.1 status line (`"HTTP/1.1 200 Connection
/// established"` → `200`), or the first line malformed.
fn parse_status_line(head: &str) -> Result<u16, String> {
    let first_line = head.lines().next().unwrap_or("");
    first_line
        .split_whitespace()
        .nth(1)
        .and_then(|code| code.parse::<u16>().ok())
        .ok_or_else(|| format!("could not parse a status line from {first_line:?}"))
}

/// D-09r: the tunnel half. Proves a handshake through the relay's own CONNECT listener,
/// trusting ONLY `ca_pem`, then sends the same request [`probe`] would send for `fmt` — over
/// the encrypted tunnel — and applies the same verdict. The CONNECT carries [`PREFLIGHT_HEADER`],
/// so plan 01's refusal ledger (`RefusalLedger`) records neither success nor refusal for it: our
/// own probe must never mask an agent's real refusals.
///
/// Every step runs under the one `timeout`. Only the CA arms (trust-store roots, the TLS
/// handshake itself) carry [`CA_REASON_MARKER`]; reachability and route errors do not.
pub async fn probe_intercept(
    cfg: &crate::config::Config,
    port: u16,
    fmt: WireFormat,
    upstream: &str,
    ca_pem: &std::path::Path,
    host: &str,
    timeout: Duration,
) -> Result<(), String> {
    match tokio::time::timeout(
        timeout,
        probe_intercept_body(cfg, port, fmt, upstream, ca_pem, host),
    )
    .await
    {
        Ok(inner) => inner,
        Err(_) => Err(format!(
            "the intercept probe for {host} did not complete within {}s",
            timeout.as_secs()
        )),
    }
}

async fn probe_intercept_body(
    cfg: &crate::config::Config,
    port: u16,
    fmt: WireFormat,
    upstream: &str,
    ca_pem: &std::path::Path,
    host: &str,
) -> Result<(), String> {
    use tokio::io::AsyncWriteExt;

    // 0. An https:// or Negotiate proxy route cannot carry an opaque tunnel yet (01), so this
    //    agent stays unwired and direct; the reason is rendered as-is, with no marker.
    crate::core::egress::tunnel::route_supported(&cfg.egress)?;

    // 1. Reach the relay's own loopback listener.
    let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", port))
        .await
        .map_err(|e| format!("relay unreachable on 127.0.0.1:{port}: {e}"))?;

    // 2. CONNECT, carrying PREFLIGHT_HEADER so 01's ledger ignores it.
    let connect_req =
        format!("CONNECT {host}:443 HTTP/1.1\r\nHost: {host}:443\r\n{PREFLIGHT_HEADER}: 1\r\n\r\n");
    stream
        .write_all(connect_req.as_bytes())
        .await
        .map_err(|e| format!("could not send CONNECT to the relay: {e}"))?;
    let head = read_head(&mut stream)
        .await
        .map_err(|e| format!("could not read the relay's CONNECT response: {e}"))?;
    let status = parse_status_line(&head)?;
    if status != 200 {
        return Err(format!("relay refused CONNECT {host}:443 with {status}"));
    }

    // 3. Trust ONLY our CA, then complete the TLS handshake through the tunnel.
    // `rustls::pki_types::pem::PemObject` — never `rustls_pki_types` by name (it is optional,
    // gated by `proxy-negotiate`, and unreachable from a plain `model-relay` build).
    use rustls::pki_types::pem::PemObject;
    let mut roots = rustls::RootCertStore::empty();
    let certs: Vec<rustls::pki_types::CertificateDer<'static>> =
        rustls::pki_types::CertificateDer::pem_file_iter(ca_pem)
            .map_err(|e| {
                format!(
                    "{CA_REASON_MARKER}{} is missing or holds no certificate: {e}",
                    ca_pem.display()
                )
            })?
            .collect::<Result<Vec<_>, _>>()
            .map_err(|e| {
                format!(
                    "{CA_REASON_MARKER}{} is missing or holds no certificate: {e}",
                    ca_pem.display()
                )
            })?;
    if certs.is_empty() {
        return Err(format!(
            "{CA_REASON_MARKER}{} is missing or holds no certificate",
            ca_pem.display()
        ));
    }
    for cert in certs {
        roots.add(cert).map_err(|e| {
            format!(
                "{CA_REASON_MARKER}{} could not be trusted as a root: {e}",
                ca_pem.display()
            )
        })?;
    }
    let client_config = rustls::ClientConfig::builder_with_provider(std::sync::Arc::new(
        rustls::crypto::ring::default_provider(),
    ))
    .with_safe_default_protocol_versions()
    .map_err(|e| format!("{CA_REASON_MARKER}TLS client config: {e}"))?
    .with_root_certificates(roots)
    .with_no_client_auth();
    let connector = tokio_rustls::TlsConnector::from(std::sync::Arc::new(client_config));
    let server_name = rustls::pki_types::ServerName::try_from(host.to_string())
        .map_err(|e| format!("{host} is not a valid TLS server name: {e}"))?;
    let mut tls = connector.connect(server_name, stream).await.map_err(|e| {
        format!(
            "{CA_REASON_MARKER}the certificate the relay presents for {host} is not signed by {}: {e}",
            ca_pem.display()
        )
    })?;

    // 4. Over TLS, the SAME request `probe` sends for `fmt`.
    let shape = ProbeShape::of(fmt);
    let body = shape.body();
    let mut req = format!(
        "{} {} HTTP/1.1\r\nHost: {host}\r\n{PREFLIGHT_HEADER}: 1\r\n",
        shape.method(),
        shape.route(),
    );
    for (name, value) in shape.auth_headers() {
        req.push_str(&format!("{name}: {value}\r\n"));
    }
    if !body.is_empty() {
        req.push_str("content-type: application/json\r\n");
        req.push_str(&format!("content-length: {}\r\n", body.len()));
    }
    req.push_str("\r\n");
    req.push_str(body);
    tls.write_all(req.as_bytes())
        .await
        .map_err(|e| format!("could not send the probe request through the tunnel: {e}"))?;

    // 5. `probe`'s verdict: a head carrying UPSTREAM_UNREACHABLE_HEADER → Err; any other
    //    response, 401 included → Ok(()).
    let head = read_head(&mut tls)
        .await
        .map_err(|e| format!("could not read the probe response through the tunnel: {e}"))?;
    if head
        .to_ascii_lowercase()
        .lines()
        .any(|line| line.starts_with(&format!("{UPSTREAM_UNREACHABLE_HEADER}:")))
    {
        return Err(format!(
            "the model relay could not reach {upstream} — model calls would fail"
        ));
    }
    Ok(())
}

/// Read up to an 8 KiB head (terminated by a blank line) off `io`, as raw text — no new parser.
async fn read_head<R>(io: &mut R) -> Result<String, String>
where
    R: tokio::io::AsyncRead + Unpin,
{
    use tokio::io::AsyncReadExt;
    const HEAD_LIMIT: usize = 8192;
    let mut buf = Vec::with_capacity(512);
    loop {
        if buf.len() >= HEAD_LIMIT {
            return Err("response head exceeded 8 KiB".to_string());
        }
        let mut chunk = [0u8; 512];
        let n = io.read(&mut chunk).await.map_err(|e| e.to_string())?;
        if n == 0 {
            return Err("the connection closed before the head completed".to_string());
        }
        buf.extend_from_slice(&chunk[..n]);
        if let Some(pos) = find_subslice(&buf, b"\r\n\r\n") {
            return Ok(String::from_utf8_lossy(&buf[..pos]).into_owned());
        }
    }
}

/// The tail of `prove_store`'s refusal. The install command changes trust for the whole OS user,
/// so an isolated instance never prints it: only the machine's own install may ask for that.
fn untrusted_remedy(owns_machine: bool) -> String {
    if owns_machine {
        crate::model_relay::trust_store::install_command(&crate::model_relay::ca::ca_pem_path(
            &crate::model_relay::ca::ca_dir(&crate::config::openlatch_dir()),
        ))
    } else {
        "only the machine's own install trusts its relay certificate".to_string()
    }
}

/// The STORE half of D-09r: the OS trust store must accept a leaf the relay itself mints for
/// `host`. Synchronous and blocking (it shells out); the daemon calls it under `spawn_blocking`.
pub fn prove_store(
    store: &dyn crate::model_relay::trust_store::TrustStore,
    ic: &crate::model_relay::ca::Interceptor,
    host: &str,
) -> Result<(), String> {
    let leaf = ic.mint_leaf_pem(host).map_err(|e| {
        format!(
            "{CA_REASON_MARKER}could not mint a leaf for {host}: {}",
            e.message
        )
    })?;
    match store.verify_leaf(&leaf, host) {
        Ok(true) => Ok(()),
        Ok(false) => Err(format!(
            "{CA_REASON_MARKER}the user trust store does not trust the relay's CA for {host} — {}",
            untrusted_remedy(crate::supervision::owns_machine_supervision())
        )),
        Err(e) => Err(format!(
            "{CA_REASON_MARKER}the trust store could not be asked: {}",
            e.message
        )),
    }
}

/// Send one synthetic request through the model relay on `port` and classify the
/// round trip.
///
/// The listener must already be serving — this probes it over loopback exactly
/// as an agent would, rather than calling the handler in-process, because "the
/// handler works" and "the listener is reachable" are different claims and the
/// agent depends on both.
///
/// `upstream` is only ever named in the failure message — the probe cannot reach
/// it directly and must not try, since a check that bypassed the forwarder would
/// vouch for a path nobody uses. Passing it in keeps the message honest when the
/// daemon forwards somewhere other than the first-party API.
///
/// `Ok(())` means a response came back and it did not originate from our own
/// unreachable-upstream fallback. Every other outcome is `Err` with a reason
/// short enough for a CLI error and specific enough to act on.
///
/// **The probe speaks `fmt`** — its route, its credential header and its body.
/// Everything else is identical across formats: the same loopback target, the
/// same `UPSTREAM_UNREACHABLE_HEADER` check, the same error strings, and the
/// same non-credential.
pub async fn probe(
    port: u16,
    fmt: WireFormat,
    upstream: &str,
    timeout: Duration,
) -> Result<(), String> {
    let client = match crate::egress::client_builder().timeout(timeout).build() {
        Ok(c) => c,
        Err(e) => return Err(format!("could not build the preflight client: {e}")),
    };

    // Route, credential header and body all come off ONE mapping, so a format
    // cannot be given one format's route and another's body.
    let shape = ProbeShape::of(fmt);
    let url = format!("http://127.0.0.1:{port}{}", shape.route());
    let mut req = client
        .request(shape.method(), &url)
        .header(PREFLIGHT_HEADER, "1");
    if !shape.body().is_empty() {
        req = req
            .header("content-type", "application/json")
            .body(shape.body());
    }
    for (name, value) in shape.auth_headers() {
        req = req.header(*name, *value);
    }
    let sent = req.send().await;

    let resp = match sent {
        Ok(r) => r,
        Err(e) if e.is_timeout() => {
            return Err(format!(
                "no response from the model relay on 127.0.0.1:{port} within {}s",
                timeout.as_secs()
            ))
        }
        // `without_url` keeps the message to the source cause. The URL is
        // loopback and the body synthetic, so nothing sensitive is at stake —
        // but a reqwest Display that embeds the request is a habit worth not
        // forming on a path whose output lands in CLI errors and logs.
        Err(e) => {
            return Err(format!(
                "could not reach the model relay on 127.0.0.1:{port}: {}",
                e.without_url()
            ))
        }
    };

    if resp.headers().contains_key(UPSTREAM_UNREACHABLE_HEADER) {
        return Err(format!(
            "the model relay is listening but could not reach {upstream} — model calls would fail"
        ));
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::model_relay::{mock, serve_ephemeral, ModelRelayState};
    use std::sync::Arc;

    fn state_for(upstream_port: u16) -> Arc<ModelRelayState> {
        let base = reqwest::Url::parse(&format!("http://127.0.0.1:{upstream_port}")).unwrap();
        Arc::new(ModelRelayState::new(base, 0, 8, &[]))
    }

    #[test]
    fn only_the_machine_install_is_told_to_change_os_trust() {
        let isolated = untrusted_remedy(false);
        assert!(!isolated.contains("add-trusted-cert") && !isolated.contains("certutil"));
        #[cfg(target_os = "macos")]
        assert!(untrusted_remedy(true).contains("add-trusted-cert"));
    }

    /// I-4 D-41: a host declared by exactly one agent is attributed to it, dynamic hosts included;
    /// two declarers of one host name nobody.
    #[test]
    fn sole_interceptor_for_names_one_declarer_and_nobody_for_two() {
        let ws = WiringState::default();
        ws.set_interceptors(BTreeMap::from([(
            "cline",
            BTreeSet::from(["a.test".to_string(), "qwen.gpu.test".to_string()]),
        )]));
        assert_eq!(ws.sole_interceptor_for("A.TEST"), Some("cline"));
        assert_eq!(ws.sole_interceptor_for("qwen.gpu.test"), Some("cline"));

        let mut two = ws.intercept_hosts();
        two.insert("fake", BTreeSet::from(["a.test".to_string()]));
        ws.set_interceptors(two);
        assert_eq!(ws.sole_interceptor_for("a.test"), None);
        assert_eq!(ws.sole_interceptor_for("qwen.gpu.test"), Some("cline"));

        assert_eq!(ws.sole_interceptor_for("undeclared.test"), None);
        assert!(ws.declares_interceptor("cline"));
        assert!(!ws.declares_interceptor("codex"));
    }

    /// I-4 D-42: proofs start false, a decrypted request proves every declarer of its host, and a
    /// fresh wiring clears one agent's proof alone.
    #[test]
    fn a_decrypted_request_proves_every_declarer_until_rewired() {
        let ws = WiringState::default();
        ws.set_interceptors(BTreeMap::from([
            ("cline", BTreeSet::from(["a.test".to_string()])),
            ("fake", BTreeSet::from(["a.test".to_string()])),
        ]));
        assert!(!ws.intercept_proven("cline"));
        assert!(!ws.intercept_proven("fake"));

        ws.note_intercepted("a.test");
        assert!(ws.intercept_proven("cline"));
        assert!(ws.intercept_proven("fake"));

        ws.clear_intercept_proof("cline");
        assert!(!ws.intercept_proven("cline"));
        assert!(ws.intercept_proven("fake"));

        let proofs = ws.intercept_proofs();
        assert_eq!(
            proofs.keys().copied().collect::<Vec<_>>(),
            vec!["cline", "fake"]
        );
    }

    /// I-4 D-41/D-42: an interceptor's format is never recorded, so it never enters
    /// `sole_wired_agent_for` — Claude Code's Anthropic attribution stays byte-identical.
    #[test]
    fn interceptors_never_enter_the_format_rung() {
        let ws = WiringState::default();
        ws.set_wired("claude-code", true);
        ws.set_wired_format("claude-code", WireFormat::AnthropicMessages);
        ws.set_wired("cline", true);
        ws.set_interceptors(BTreeMap::from([(
            "cline",
            BTreeSet::from(["api.anthropic.com".to_string()]),
        )]));
        assert_eq!(
            ws.sole_wired_agent_for(WireFormat::AnthropicMessages),
            Some("claude-code")
        );

        // Control: recording cline's format too (what §5.5 prevents) makes it a second speaker.
        ws.set_wired_format("cline", WireFormat::AnthropicMessages);
        assert_eq!(ws.sole_wired_agent_for(WireFormat::AnthropicMessages), None);
    }

    /// **Every probe route must resolve back to its own format.**
    ///
    /// The relay routes the probe by path exactly as it routes the agent, so a
    /// route that resolves to something else sends the probe to a different
    /// vendor's upstream. Google's is the one that can go wrong quietly: the
    /// arm matches on the `:generateContent` suffix alone, so a probe posted to
    /// a bare `/v1beta/models` resolves `Unknown` — whose upstream is
    /// Anthropic's — and the preflight becomes an instance of the misroute the
    /// gate it opens exists to prevent.
    #[test]
    fn probe_paths_resolve_to_their_own_format() {
        for fmt in [
            WireFormat::AnthropicMessages,
            WireFormat::OpenAiResponses,
            WireFormat::OpenAiChatCompletions,
            WireFormat::GoogleGenerateContent,
            WireFormat::OllamaNative,
        ] {
            let route = ProbeShape::of(fmt).route();
            let (parts, _) = axum::http::Request::builder()
                .method(axum::http::Method::POST)
                .uri(format!("http://127.0.0.1:1{route}"))
                .body(())
                .expect("a request")
                .into_parts();
            assert_eq!(
                WireFormat::resolve(&parts),
                fmt,
                "{fmt:?}'s probe route {route} must resolve to {fmt:?}, or the probe \
                 is forwarded to another format's upstream"
            );
        }
        assert!(
            ProbeShape::of(WireFormat::GoogleGenerateContent)
                .route()
                .ends_with(":generateContent"),
            "the colon suffix is what the Google arm matches on — a bare \
             /v1beta/models resolves Unknown and goes to Anthropic"
        );
    }

    /// **The stub plan 01 landed is gone.** Each format's probe carries its own
    /// route AND its own body, asserted on the bytes the upstream actually
    /// received.
    ///
    /// An Anthropic-shaped body posted at a chat-completions route either fails
    /// against a correct deployment — leaving the plane unwired — or, against a
    /// permissive gateway, PASSES FOR THE WRONG REASON and opens the wiring gate
    /// on a round trip that proved nothing about the plane being wired.
    #[tokio::test(flavor = "multi_thread")]
    async fn preflight_probes_each_format_in_its_own_shape() {
        // (format, a substring that appears in THIS format's body and in no
        // other's, a substring that must NOT appear)
        let cases = [
            (
                WireFormat::OpenAiChatCompletions,
                "/v1/chat/completions",
                "\"messages\"",
                "\"input\"",
            ),
            (
                WireFormat::GoogleGenerateContent,
                ":generateContent",
                "\"contents\"",
                "\"messages\"",
            ),
            (
                WireFormat::OllamaNative,
                "POST /api/chat",
                "\"stream\":false",
                "max_tokens",
            ),
        ];

        for (fmt, route_marker, expected, forbidden) in cases {
            let upstream = mock::spawn_capture_200().await;
            let port = serve_ephemeral(state_for(upstream.port)).await;

            probe(port, fmt, "http://127.0.0.1", PREFLIGHT_TIMEOUT)
                .await
                .unwrap_or_else(|e| panic!("{fmt:?} probe: {e}"));

            let line = upstream
                .received_request_line
                .lock()
                .expect("the mock recorded a request line")
                .clone()
                .expect("a request arrived");
            assert!(
                line.contains(route_marker),
                "{fmt:?} must probe its own route, got: {line}"
            );

            let body = String::from_utf8(
                upstream
                    .received_body
                    .lock()
                    .expect("the mock recorded a body")
                    .clone()
                    .expect("a body arrived"),
            )
            .expect("utf8");
            assert!(
                body.contains(expected),
                "{fmt:?} must probe in its own body shape, got: {body}"
            );
            assert!(
                !body.contains(forbidden),
                "{fmt:?} must not carry another format's body, got: {body}"
            );
        }
    }

    /// A slot whose protocol no format names is probed for reachability alone:
    /// a bodiless GET that carries no credential header of any vendor's.
    #[tokio::test(flavor = "multi_thread")]
    async fn ollama_native_probe_posts_api_chat_and_unknown_probes_reachability() {
        let upstream = mock::spawn_capture_200().await;
        let port = serve_ephemeral(state_for(upstream.port)).await;
        probe(
            port,
            WireFormat::Unknown,
            "http://127.0.0.1",
            PREFLIGHT_TIMEOUT,
        )
        .await
        .expect("reachability probe");
        let line = upstream
            .received_request_line
            .lock()
            .expect("lock")
            .clone()
            .expect("a request arrived");
        assert!(line.starts_with("GET / "), "{line}");
        assert_eq!(upstream.header("x-api-key"), None);
        assert_eq!(upstream.header("authorization"), None);
    }

    /// The gate opens on a reachable upstream — including one that rejects the
    /// call. "Upstream answered" is the claim, not "the call succeeded": the
    /// probe carries no credential precisely so it cannot succeed.
    #[tokio::test(flavor = "multi_thread")]
    async fn probe_passes_when_upstream_answers() {
        let upstream = mock::spawn_capture_200().await;
        let port = serve_ephemeral(state_for(upstream.port)).await;

        assert_eq!(
            probe(
                port,
                WireFormat::AnthropicMessages,
                crate::model_relay::ANTHROPIC_BASE,
                PREFLIGHT_TIMEOUT
            )
            .await,
            Ok(())
        );
    }

    /// An upstream that cannot be reached produces the synthetic 502, and the
    /// gate must stay shut on it. This is the exact shape of the field bug: the
    /// bind succeeded, the listener is up, and every session would still die if
    /// the agent were wired here.
    #[tokio::test(flavor = "multi_thread")]
    async fn probe_fails_when_upstream_is_unreachable() {
        let dead = mock::closed_port().await;
        let port = serve_ephemeral(state_for(dead)).await;

        let err = probe(
            port,
            WireFormat::AnthropicMessages,
            crate::model_relay::ANTHROPIC_BASE,
            PREFLIGHT_TIMEOUT,
        )
        .await
        .unwrap_err();
        assert!(
            err.contains("could not reach"),
            "an unreachable upstream must be named as such, got: {err}"
        );
    }

    /// An upstream that accepts the connection and then says nothing is the hang
    /// the forward path's header timeout exists for. The probe must not wait it
    /// out — it has its own, much tighter budget, and a daemon start that blocks
    /// on a silent provider is its own outage.
    #[tokio::test(flavor = "multi_thread")]
    async fn probe_fails_fast_on_a_silent_upstream() {
        let hung = mock::spawn_hang_after_accept().await;
        let upstream = reqwest::Url::parse(&format!("http://127.0.0.1:{hung}")).unwrap();
        // Mirror production ordering: the forward path's own header wait is far
        // longer than the probe budget, so the probe's timeout is what fires.
        let state = Arc::new(
            ModelRelayState::new(upstream, 0, 8, &[]).with_header_timeout(Duration::from_secs(60)),
        );
        let port = serve_ephemeral(state).await;

        let started = std::time::Instant::now();
        let err = probe(
            port,
            WireFormat::AnthropicMessages,
            crate::model_relay::ANTHROPIC_BASE,
            Duration::from_millis(300),
        )
        .await
        .unwrap_err();
        assert!(
            err.contains("no response"),
            "a silent upstream must read as no response, got: {err}"
        );
        assert!(
            started.elapsed() < Duration::from_secs(5),
            "the probe must return on its own budget, not the forward path's"
        );
    }

    /// Nothing listening at all — what a supervisor restart passes through.
    #[tokio::test(flavor = "multi_thread")]
    async fn probe_fails_when_nothing_is_listening() {
        let port = mock::closed_port().await;
        assert!(probe(
            port,
            WireFormat::AnthropicMessages,
            crate::model_relay::ANTHROPIC_BASE,
            Duration::from_millis(500)
        )
        .await
        .is_err());
    }

    /// THE D-13 GATE. Without it the probe passes on a broken Codex plane: run
    /// unchanged it sends `POST /v1/messages` with `anthropic-version`, proving
    /// a **Claude Code** round trip and then opening the gate for Codex — a
    /// disabled subsystem rendering as healthy.
    #[tokio::test(flavor = "multi_thread")]
    async fn probe_sends_the_format_it_was_given() {
        let upstream = mock::spawn_capture_200().await;
        let port = serve_ephemeral(state_for(upstream.port)).await;

        assert_eq!(
            probe(
                port,
                WireFormat::OpenAiResponses,
                crate::model_relay::wire_format::OPENAI_BASE,
                PREFLIGHT_TIMEOUT
            )
            .await,
            Ok(())
        );

        let line = upstream
            .received_request_line
            .lock()
            .unwrap()
            .clone()
            .expect("the mock recorded the request line");
        assert!(
            line.starts_with("POST /v1/responses"),
            "a Responses probe must speak the Responses route, got: {line}"
        );
        assert_eq!(
            upstream.header("anthropic-version"),
            None,
            "an Anthropic protocol header on an OpenAI request is the exact defect this gate exists to catch"
        );
    }

    /// The probe's claim is "upstream answered", never "the call worked" — so it
    /// must keep carrying a literal that cannot authenticate, in BOTH formats.
    /// A real credential here would put OpenLatch's own synthetic traffic on the
    /// customer's bill.
    #[tokio::test(flavor = "multi_thread")]
    async fn probe_carries_no_real_credential() {
        let anthropic_up = mock::spawn_capture_200().await;
        let port = serve_ephemeral(state_for(anthropic_up.port)).await;
        assert_eq!(
            probe(
                port,
                WireFormat::AnthropicMessages,
                crate::model_relay::ANTHROPIC_BASE,
                PREFLIGHT_TIMEOUT
            )
            .await,
            Ok(())
        );
        assert_eq!(
            anthropic_up.header("x-api-key").as_deref(),
            Some(PREFLIGHT_API_KEY),
            "the Anthropic probe must send the not-a-key literal"
        );

        let responses_up = mock::spawn_capture_200().await;
        let port = serve_ephemeral(state_for(responses_up.port)).await;
        assert_eq!(
            probe(
                port,
                WireFormat::OpenAiResponses,
                crate::model_relay::wire_format::OPENAI_BASE,
                PREFLIGHT_TIMEOUT
            )
            .await,
            Ok(())
        );
        assert_eq!(
            responses_up.header("authorization").as_deref(),
            Some(format!("Bearer {PREFLIGHT_API_KEY}").as_str()),
            "the Responses probe must send the same not-a-key literal as a Bearer"
        );
    }

    #[test]
    fn verdict_labels_are_stable() {
        assert_eq!(Verdict::default(), Verdict::Pending);
        assert_eq!(Verdict::Pending.label(), "pending");
        assert_eq!(Verdict::Ok.label(), "ok");
        assert_eq!(Verdict::Failed("boom".into()).label(), "failed");
        assert_eq!(Verdict::Failed("boom".into()).error(), Some("boom"));
        assert_eq!(Verdict::Ok.error(), None);
        assert!(Verdict::Ok.is_ok());
        assert!(!Verdict::Pending.is_ok());
    }

    #[test]
    fn wiring_state_round_trips() {
        let st = WiringState::default();
        assert!(!st.is_wired("claude-code"));
        assert_eq!(st.verdict("claude-code"), Verdict::Pending);

        st.set_wired("claude-code", true);
        st.set_verdict("claude-code", Verdict::Ok);
        assert!(st.is_wired("claude-code"));
        assert_eq!(st.verdict("claude-code"), Verdict::Ok);
    }

    /// Two agents point at ONE listener through two conventions, and one can be
    /// wired while the other's round trip fails. A process-wide flag reports
    /// one plane's verdict for both — which is how `init` passes on a host
    /// whose Codex plane never came up, and how one failing probe unwires the
    /// agent that was working.
    ///
    /// `wiring_state_round_trips` above cannot gate this: it still compiles and
    /// passes against a single-valued `WiringState`.
    #[test]
    fn wiring_state_is_keyed_per_agent() {
        let st = WiringState::default();

        st.set_wired("claude-code", true);
        assert!(st.is_wired("claude-code"));
        assert!(
            !st.is_wired("codex-cli"),
            "one agent's wiring says nothing about another's"
        );

        st.set_verdict("claude-code", Verdict::Ok);
        assert_eq!(
            st.verdict("codex-cli"),
            Verdict::Pending,
            "an agent with no entry is Pending — not the other agent's verdict"
        );

        st.set_verdict("codex-cli", Verdict::Failed("no round trip".into()));
        let verdicts = st.verdicts();
        assert_eq!(verdicts.get("claude-code"), Some(&Verdict::Ok));
        assert_eq!(
            verdicts.get("codex-cli"),
            Some(&Verdict::Failed("no round trip".into())),
            "the snapshot the admin surface renders carries every agent"
        );

        let wired = st.wired_agents();
        assert_eq!(wired.get("claude-code"), Some(&true));
        assert_eq!(
            wired.get("codex-cli"),
            None,
            "a verdict is not a write: codex-cli was judged, never wired"
        );

        // Seeding gives an unprobed agent its entry without clobbering one
        // that has already been judged.
        st.seed("cline");
        st.seed("claude-code");
        assert_eq!(st.verdict("cline"), Verdict::Pending);
        assert_eq!(st.wired_agents().get("cline"), Some(&false));
        assert_eq!(
            st.verdict("claude-code"),
            Verdict::Ok,
            "re-seeding must never overwrite a verdict the supervisor recorded"
        );
        assert_eq!(st.wired_agents().get("claude-code"), Some(&true));
    }

    /// One wired speaker of the format, so the request has one possible author.
    #[test]
    fn a_lone_speaker_is_named() {
        let st = WiringState::default();
        st.set_wired("claude-code", true);
        st.set_wired_format("claude-code", WireFormat::AnthropicMessages);
        st.set_wired("codex-cli", true);
        st.set_wired_format("codex-cli", WireFormat::OpenAiResponses);

        assert_eq!(
            st.sole_wired_agent_for(WireFormat::AnthropicMessages),
            Some("claude-code")
        );
        assert_eq!(
            st.sole_wired_agent_for(WireFormat::OpenAiResponses),
            Some("codex-cli")
        );
    }

    /// TWO speakers, and the answer is silence.
    ///
    /// This is the case the whole design turns on. Cline and Claude Code both
    /// speak the Anthropic Messages API, so on a host running both, naming
    /// either is a coin flip — and it would be a coin flip wearing the
    /// platform's `(org, source, agent_id)` join key, which is exactly the
    /// unfalsifiable attribution `unknown` exists to prevent.
    #[test]
    fn two_speakers_of_one_format_name_nobody() {
        let st = WiringState::default();
        st.set_wired("claude-code", true);
        st.set_wired_format("claude-code", WireFormat::AnthropicMessages);
        st.set_wired("cline", true);
        st.set_wired_format("cline", WireFormat::AnthropicMessages);

        assert_eq!(st.sole_wired_agent_for(WireFormat::AnthropicMessages), None);
    }

    /// An agent that is no longer wired is not a candidate.
    ///
    /// `set_wired_format` deliberately does not clear on unwire, so without the
    /// flag check an uninstalled agent would keep answering for a protocol it
    /// no longer speaks — and worse, would keep a live second agent from being
    /// the lone speaker.
    #[test]
    fn an_unwired_agent_does_not_speak() {
        let st = WiringState::default();
        st.set_wired_format("claude-code", WireFormat::AnthropicMessages);
        st.set_wired("claude-code", false);
        assert_eq!(st.sole_wired_agent_for(WireFormat::AnthropicMessages), None);

        st.set_wired("cline", true);
        st.set_wired_format("cline", WireFormat::AnthropicMessages);
        assert_eq!(
            st.sole_wired_agent_for(WireFormat::AnthropicMessages),
            Some("cline"),
            "the unwired agent must not block the one that is actually wired"
        );
    }

    /// Nothing wired for the format at all.
    #[test]
    fn no_speaker_names_nobody() {
        let st = WiringState::default();
        st.set_wired("claude-code", true);
        st.set_wired_format("claude-code", WireFormat::AnthropicMessages);
        assert_eq!(st.sole_wired_agent_for(WireFormat::OpenAiResponses), None);
    }

    /// `Unknown` is never attributable, however few agents are wired.
    ///
    /// It is the catch-all for every route the model relay does not capture
    /// (`GET /v1/models`, count_tokens, the batch endpoints), so "the sole agent
    /// speaking Unknown" would be a statement about the route table, not about
    /// an agent. Those routes emit no economics event anyway; refusing here
    /// keeps the rule true rather than true-by-accident.
    #[test]
    fn the_uncaptured_format_is_never_attributable() {
        let st = WiringState::default();
        st.set_wired("claude-code", true);
        st.set_wired_format("claude-code", WireFormat::Unknown);
        assert_eq!(st.sole_wired_agent_for(WireFormat::Unknown), None);
    }

    // -- I-2 plan 02: `probe_intercept`, `prove_store`, the shared slot --------

    #[test]
    fn wiring_state_hands_out_one_slot_and_one_ledger() {
        let wiring = WiringState::default();
        assert!(
            Arc::ptr_eq(&wiring.intercept(), &wiring.intercept()),
            "the relay and the supervisor must share one intercept slot"
        );
        assert!(
            Arc::ptr_eq(&wiring.refusals(), &wiring.refusals()),
            "the relay and the supervisor must share one refusal ledger"
        );
    }

    /// A relay whose interceptor declares `hosts`, forwarding Anthropic Messages to a mock
    /// upstream that always answers 200. Returns (relay port, the CA's `ca.pem` path, the CA's
    /// tempdir — kept alive by the caller, the mock upstream's port).
    async fn intercepting_relay(
        hosts: &[&'static str],
    ) -> (u16, std::path::PathBuf, tempfile::TempDir, u16) {
        use crate::model_relay::ca::{ca_dir, ca_pem_path, Interceptor};

        let dir = tempfile::tempdir().expect("tempdir");
        let ca_root = ca_dir(dir.path());
        let interceptor =
            Arc::new(Interceptor::new(&ca_root, hosts.iter().copied()).expect("interceptor"));
        let ca_pem = ca_pem_path(&ca_root);
        let upstream = mock::spawn_always_200().await;
        let map: std::collections::BTreeMap<String, String> = [(
            WireFormat::AnthropicMessages.as_str().to_string(),
            format!("http://127.0.0.1:{upstream}"),
        )]
        .into_iter()
        .collect();
        let intercept_slot: crate::model_relay::InterceptSlot = Default::default();
        *intercept_slot.write().unwrap() = Some(interceptor);
        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_intercept(intercept_slot),
        );
        let port = serve_ephemeral(state).await;
        (port, ca_pem, dir, upstream)
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn probe_intercept_passes_with_the_on_disk_ca() {
        let (port, ca_pem, _dir, upstream) = intercepting_relay(&["a.test"]).await;
        let cfg = crate::config::Config::defaults();
        probe_intercept(
            &cfg,
            port,
            WireFormat::AnthropicMessages,
            &format!("http://127.0.0.1:{upstream}"),
            &ca_pem,
            "a.test",
            PREFLIGHT_TIMEOUT,
        )
        .await
        .expect("probe_intercept must pass with the on-disk CA");
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn probe_intercept_fails_with_the_marker_when_the_ca_is_wrong() {
        let (port, _ca_pem, _dir, upstream) = intercepting_relay(&["a.test"]).await;
        // A DIFFERENT, freshly generated CA — never the one the interceptor actually signs
        // leaves with.
        let wrong_dir = tempfile::tempdir().expect("tempdir");
        let wrong_root = crate::model_relay::ca::ca_dir(wrong_dir.path());
        crate::model_relay::ca::LocalCa::load_or_generate(&wrong_root).expect("a different CA");
        let wrong_ca_pem = crate::model_relay::ca::ca_pem_path(&wrong_root);

        let cfg = crate::config::Config::defaults();
        let err = probe_intercept(
            &cfg,
            port,
            WireFormat::AnthropicMessages,
            &format!("http://127.0.0.1:{upstream}"),
            &wrong_ca_pem,
            "a.test",
            PREFLIGHT_TIMEOUT,
        )
        .await
        .expect_err("a mismatched CA must fail the TLS handshake");
        assert!(err.starts_with(CA_REASON_MARKER), "{err}");
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn probe_intercept_refuses_an_unsupported_route() {
        use crate::core::egress::{EgressConfig, ProxyMode};

        let (port, ca_pem, _dir, upstream) = intercepting_relay(&["a.test"]).await;

        let mut manual_cfg = crate::config::Config::defaults();
        manual_cfg.egress = EgressConfig {
            mode: ProxyMode::Manual,
            url: Some("https://proxy.test:3128".to_string()),
            ..EgressConfig::direct()
        };
        let err = probe_intercept(
            &manual_cfg,
            port,
            WireFormat::AnthropicMessages,
            &format!("http://127.0.0.1:{upstream}"),
            &ca_pem,
            "a.test",
            PREFLIGHT_TIMEOUT,
        )
        .await
        .expect_err("an https:// proxy route cannot carry an opaque tunnel yet");
        assert!(
            !err.starts_with(CA_REASON_MARKER),
            "a route refusal is not a CA failure: {err}"
        );

        let direct_cfg = crate::config::Config::defaults();
        probe_intercept(
            &direct_cfg,
            port,
            WireFormat::AnthropicMessages,
            &format!("http://127.0.0.1:{upstream}"),
            &ca_pem,
            "a.test",
            PREFLIGHT_TIMEOUT,
        )
        .await
        .expect("Direct egress must reach the scheme check and pass — it never reaches it");
    }

    #[test]
    fn prove_store_passes_only_when_the_store_verifies_the_leaf() {
        use crate::model_relay::ca::Interceptor;
        use crate::model_relay::trust_store::test_support::FakeStore;

        let dir = tempfile::tempdir().expect("tempdir");
        let ca_root = crate::model_relay::ca::ca_dir(dir.path());
        let interceptor = Interceptor::new(&ca_root, ["a.test"]).expect("interceptor");

        let store = FakeStore::default();
        store
            .verify
            .store(true, std::sync::atomic::Ordering::SeqCst);
        prove_store(&store, &interceptor, "a.test").expect("verify: true must pass");

        let store = FakeStore::default();
        store
            .verify
            .store(false, std::sync::atomic::Ordering::SeqCst);
        let err = prove_store(&store, &interceptor, "a.test").expect_err("verify: false must fail");
        assert!(err.starts_with(CA_REASON_MARKER), "{err}");
    }
}