openlatch-client 0.5.4

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

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

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

/// 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, &[]))
    }

    /// **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("cursor");
        st.seed("claude-code");
        assert_eq!(st.verdict("cursor"), Verdict::Pending);
        assert_eq!(st.wired_agents().get("cursor"), 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);
    }
}