openlatch-client 0.3.3

//! Model-boundary listener — the reliable forwarder (Model Boundary
//! Enforcement Layer, plan 01; owns D-01…D-07).
//!
//! A **second** axum listener inside the `openlatch` binary that Claude Code
//! points `ANTHROPIC_BASE_URL` at. It streams every request through to
//! `api.anthropic.com` with **zero buffering** and **byte-identical** bodies,
//! and forwards everything it does not capture opaquely. It is deliberately the
//! most dependency-minimal loop in the codebase:
//!
//! > **Prime invariant (D-24):** the forwarder is more reliable than anything
//! > it hosts. It never buffers and never mutates the forwarded bytes.
//! > Everything fallible degrades to pass-through within the process.
//!
//! This plan forwards unmodified and records only that a request passed
//! through. Pricing, tokenization, session logic and transforms are plan 02/03.
//! `proxy::observe_request` is a no-op stub wrapped in `catch_unwind` here.
//!
//! ## Deployment shape
//!
//! The boundary is a listener in the **same process** as the hook daemon
//! (spawned from `run_daemon_foreground`), not its own supervised process. It
//! therefore inherits the daemon's OS supervision: a daemon crash restarts both.
//! The pinned port is bound loopback-only with **no `BIND_ALL` escape hatch**
//! (F-22) and is never re-probed (D-25).
//!
//! ## Supervision / restart cadence (C-19)
//!
//! Because the boundary is in-process, "restart this component" means "restart
//! the daemon", governed by the existing units: launchd `ThrottleInterval` and
//! systemd `RestartSec` (both ~10 s today), and — the honest Windows bound —
//! Task Scheduler's **1-minute** minimum restart interval. A sub-second restart
//! on Windows would require splitting the boundary into its own **Windows
//! Service with SCM failure actions** (`RestartDelay`); that is the future path
//! if/when the boundary becomes a standalone process. The residual dead-port
//! window during a restart is accepted (D-10); `openlatch stop` is the immediate
//! escape hatch, since the agent wiring is removed with the listener. This plan
//! does **not** change any supervisor
//! unit — doing so would alter the hook daemon's restart behaviour, which is out
//! of scope here.

pub mod bench;
pub mod billing;
pub mod capture;
pub mod churn;
pub mod emit;
pub mod mock;
pub mod prefix_shape;
pub mod preflight;
pub mod proxy;
pub mod retention;
pub mod session;
pub mod tokenize;
pub mod transforms;
pub mod wire_format;

use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use std::collections::BTreeMap;

use axum::extract::DefaultBodyLimit;
use axum::routing::get;
use axum::Router;
use tokio::net::TcpListener;
use tokio::sync::mpsc::Sender;
use tokio::sync::Semaphore;

use crate::cloud::CloudEvent;
use crate::core::policy::{PolicyHandle, ResidentBundle};
use crate::error::{OlError, ERR_BOUNDARY_SERVE};
use crate::privacy::PrivacyFilter;

use session::SessionRegistry;
use tokenize::Estimator;
use wire_format::{AuthMode, WireFormat};

/// Default pinned loopback port for the boundary listener (D-05/D-25).
///
/// Chosen outside the hook daemon's 7443–7543 probe range so the two listeners
/// never collide. Pinned: the value is deterministic and written into the agent
/// config at `init`; every start reuses it and occupied-at-startup is a loud
/// failure, never a silent move.
pub const DEFAULT_BOUNDARY_PORT: u16 = 7600;

/// Default cap on concurrently *materialized* requests (D-03). A body over the
/// 32 MB ceiling — or arriving while all permits are taken — takes the opaque
/// stream path and never materializes, so it cannot contribute to OOM.
pub const DEFAULT_INFLIGHT: usize = 16;

/// The 32 MB request-body materialization ceiling (F-17). Bodies above this
/// stream through opaque, with no capture.
pub const MAX_MATERIALIZE_BYTES: usize = 32 * 1024 * 1024;

/// How long the forwarder waits for the upstream to return **response headers**
/// before giving up with a synthetic 502.
///
/// `connect_timeout` only bounds TCP/TLS establishment; once connected, an
/// upstream that accepts but never returns a status line would otherwise wedge
/// the request forever (and, on the materialized path, hold a semaphore permit
/// for the life of the process). Generous by design — a slow first token is
/// normal, but a full minute of silence with no headers is a dead connection.
/// This bounds ONLY the header wait; the response BODY stream that follows is
/// never timed out (long SSE turns are legitimate — D-06).
pub const HEADER_TIMEOUT: Duration = Duration::from_secs(60);

/// The Anthropic first-party base URL the boundary forwards to (D-22).
pub const ANTHROPIC_BASE: &str = "https://api.anthropic.com";

/// Shared state cloned into every boundary handler via `Arc`, mirroring
/// `daemon::AppState` but kept a **separate** struct: the boundary has a
/// different body limit (32 MB vs 1 MB) and a different auth posture — it
/// forwards the caller's own provider credential verbatim and is not
/// Bearer-authed like the hook ingest.
pub struct BoundaryState {
    /// The forward client, behind the daemon's swap handle: connection-pooled, with NO
    /// overall `timeout()` on the streaming path — a stream can outlive any fixed deadline.
    /// See [`build_boundary_client`].
    ///
    /// A **handle** rather than a client because the listener is built once and then serves
    /// for the daemon's lifetime. Two things follow. A self-heal pass reaches it without a
    /// restart ([`crate::egress::EgressClients`]), and `[proxy] allow_direct = false` with
    /// nothing working is representable as "no client" — which is the only shape in which
    /// this layer cannot quietly fall through to a direct connection (D-21).
    pub client: crate::egress::ClientHandle,
    /// Upstream provider base **per wire format**, keyed by
    /// [`WireFormat::as_str`]. Read through [`BoundaryState::upstream_for`],
    /// which is a plain infallible lookup and never re-implements
    /// [`crate::config::BoundaryConfig::upstream_for`]'s precedence — the map
    /// arrives already resolved.
    upstream: BTreeMap<&'static str, reqwest::Url>,
    /// The ChatGPT backend, used for an `openai-responses` request whose own
    /// credential names it ([`AuthMode::ChatGptSubscription`]).
    ///
    /// A **second built-in**, not a second config key. Where a request goes on
    /// a ChatGPT plan is a property of how Codex authenticates, not an operator
    /// preference, exactly as `api.openai.com` is on a platform key — so it is
    /// resolved here rather than asked for in `[boundary.upstream]`, and a host
    /// keeps working across a `codex login` that switches plans.
    ///
    /// `None` once an operator configures `openai-responses` explicitly: that
    /// entry names ONE destination for this host's OpenAI traffic, and
    /// second-guessing it per credential is how a corporate gateway gets
    /// bypassed by the very requests it was put there to see. `None` also for
    /// every `BoundaryState::new` — tests and benches keep the single-base
    /// fan-out they were written against.
    openai_chatgpt: Option<reqwest::Url>,
    /// The single URL the constructor was given, and the lookup's last resort.
    ///
    /// Every constructor fans this across **every** format, so a lookup can only
    /// miss for a key no [`WireFormat`] names — which no in-crate caller can
    /// produce. It exists so the lookup is total, never as a routing rule: an
    /// `Option` return would re-open the synthetic 502 the fan-out exists to
    /// prevent, and an `expect` would panic inside the axum handler.
    fallback: reqwest::Url,
    /// Caps concurrently-materialized `/v1/messages` requests (D-03). Acquired
    /// non-blocking with `try_acquire_owned`; saturation → opaque forward.
    pub inflight: Arc<Semaphore>,
    /// Reused credential/secret scrubber for every diagnostic path (F-22).
    /// Held for parity with the daemon and for plan 02's capture paths; the
    /// forward path itself never logs the body or the credential.
    pub privacy: PrivacyFilter,
    /// Wall-clock start, surfaced by `GET /admin/boundary/status`.
    pub started_at: std::time::Instant,
    /// The pinned port this listener bound (surfaced on the status endpoint).
    pub port: u16,
    /// Maximum time to wait for upstream **response headers** before a
    /// synthetic 502 ([`HEADER_TIMEOUT`] in production). Overridable via
    /// [`BoundaryState::with_header_timeout`] so tests can exercise the
    /// timeout path in milliseconds. Bounds only the header wait — never the
    /// response body stream.
    pub header_timeout: Duration,
    // --- plan 02 measurement (D-08…D-15) ---
    /// Shared active-session registry (D-29). The **same** `Arc` the hook side
    /// (`daemon::AppState`) writes on `SessionStart` / tool-call hooks, read here
    /// to resolve the attribution triple + assurance at request time. A
    /// standalone empty registry when measurement is not wired (tests/benches) —
    /// every resolution then degrades to `unknown`, never panics.
    pub registry: Arc<SessionRegistry>,
    /// Economics event sink (D-13). `try_send` fire-and-forget onto the daemon's
    /// existing cloud rail. `None` disables emission (the forward still forwards,
    /// only measurement stops) — the plan-01 behaviour.
    pub cloud_tx: Option<Sender<CloudEvent>>,
    /// Stateless local tokenizer for the interrupted-stream estimate path (D-08).
    pub tokenizer: Estimator,
    /// Per-session previous-prefix store for churn classification (D-12).
    pub churn: Arc<churn::ChurnTracker>,
    /// Per-session **parsed-structure** view of the request prefix (D-28).
    ///
    /// Deliberately separate from [`BoundaryState::churn`], which stays
    /// byte-level and keeps observing the **original** bytes so prefix findings
    /// go on describing what the *agent* does rather than what we did to its
    /// request. This one supplies the two facts acting needs and churn
    /// structurally cannot give: the measured horizon `H`, and per-block
    /// stability. See [`prefix_shape`] for why they are not merged.
    pub prefix_shape: Arc<prefix_shape::PrefixTracker>,
    /// Whether an **acting** `prefix_reorder` (L-0) rule may rewrite the
    /// forwarded request (`[boundary] transforms_act`, D-28).
    ///
    /// **`false` by default, including in every test and bench that does not
    /// set it.** Off is a zero-cost path, not merely an inert one: the parsed
    /// body is dropped at the end of observation and the mutation step is never
    /// entered, so an instance with this off does exactly the work it did before
    /// D-28 and forwards byte-identically.
    pub transforms_act: bool,
    /// Bench/test fault injection: panic inside the mutation step (D-28).
    ///
    /// A **field**, not a process-wide static, because integration tests run in
    /// parallel threads sharing one process — a global would arm every listener
    /// at once and fail the tests running beside the one that armed it. Same
    /// posture as [`BoundaryState::header_timeout`]: a production-shaped knob
    /// that only tests and benches ever move. Always `false` in production.
    pub inject_mutate_panic: bool,
    /// The daemon's resident policy bundle, read lock-free per request.
    ///
    /// The **same** `Arc<ArcSwap<…>>` the hook verdict path reads
    /// (`daemon::AppState`), so a bundle swap is visible here on the next
    /// request with no coordination. `None` when the policy engine is
    /// disabled or not wired (tests/benches) — evaluation then falls back to
    /// [`transforms::BASELINE_RULES`], which is the plan-01 behaviour.
    ///
    /// Until this landed, `src/boundary/` had no reference to the policy
    /// module at all: the transform engine ran off a hardcoded baseline and
    /// every `select` key on an authored rule was deserialised and never read.
    pub policy: Option<PolicyHandle>,
    /// The wiring gate this listener is judged by (`preflight`).
    ///
    /// Owned by the daemon's wiring supervisor, which probes and then writes or
    /// removes `ANTHROPIC_BASE_URL`; read here only to report it on
    /// `GET /admin/boundary/status`, which is how `init`, `status` and `doctor`
    /// learn why an up listener may nonetheless be unwired. A standalone default
    /// (`pending`, unwired) in tests and benches, where nothing wires anything.
    pub wiring: Arc<preflight::WiringState>,
    /// Where this listener's upstream transport outcomes are reported.
    ///
    /// A non-reporting direct route in tests and benches (`new`), and the
    /// daemon's shared health on the production path. The boundary is the one
    /// consumer whose upstream is a *provider* rather than the platform, which
    /// is exactly why it is worth wiring: on a developer laptop it is often the
    /// only thing crossing the corporate proxy all day.
    pub egress: crate::egress::EgressReporter,
}

impl BoundaryState {
    /// Build boundary state with a freshly-constructed **direct** forward client.
    ///
    /// The twenty callers of this constructor are tests and benches with no config in
    /// scope, and a hermetic fixture that silently inherited the developer's corporate
    /// proxy would be worse than useless. Production goes through
    /// [`BoundaryState::new_with_egress`] instead.
    pub fn new(
        upstream_base: reqwest::Url,
        port: u16,
        inflight: usize,
        extra_patterns: &[String],
    ) -> Self {
        Self::new_with_egress(
            upstream_base,
            port,
            inflight,
            extra_patterns,
            &crate::egress::EgressConfig::direct(),
        )
    }

    /// Build boundary state whose forward client uses the resolved egress route.
    ///
    /// Called from the daemon's supervised task factory, so a restart rebuilds the client
    /// and picks up a route that changed while the previous run was alive -- a pool left
    /// pointing at a dead proxy is exactly what a restart is supposed to clear.
    pub fn new_with_egress(
        upstream_base: reqwest::Url,
        port: u16,
        inflight: usize,
        extra_patterns: &[String],
        egress: &crate::egress::EgressConfig,
    ) -> Self {
        Self {
            client: crate::egress::ClientHandle::new(build_boundary_client_with(egress)),
            // ONE URL populates EVERY format — not a single-entry map. Written
            // here rather than in `new` because production constructs through
            // this function, so an implementer editing only `new` would leave
            // the daemon without the fan-out. A probe or a forward for ANY
            // format through a test/bench state reaches the one base it was
            // given, which is what keeps `tests/boundary_forward.rs` and
            // `preflight::tests::state_for` working untouched.
            upstream: WireFormat::ALL
                .iter()
                .map(|f| (f.as_str(), upstream_base.clone()))
                .collect(),
            // Off unless the daemon installs a resolved map. A test or bench
            // handed one base means every format reaches that base — adding a
            // built-in second origin here would send a fixture's
            // ChatGPT-credentialled request to the real chatgpt.com.
            openai_chatgpt: None,
            fallback: upstream_base,
            inflight: Arc::new(Semaphore::new(inflight.max(1))),
            privacy: PrivacyFilter::new(extra_patterns),
            started_at: std::time::Instant::now(),
            port,
            header_timeout: HEADER_TIMEOUT,
            // Measurement defaults to "off": a standalone empty registry (every
            // resolution → unknown) and no event sink. `with_measurement` wires
            // the shared registry + cloud rail on the production path. Keeping
            // `new`'s signature stable means every plan-01 forwarder test/bench
            // call site is untouched (no forwarder regression).
            registry: Arc::new(SessionRegistry::default()),
            cloud_tx: None,
            tokenizer: Estimator,
            churn: Arc::new(churn::ChurnTracker::default()),
            prefix_shape: Arc::new(prefix_shape::PrefixTracker::default()),
            // D-28: acting is opt-in. `new` is the plan-01 forwarder
            // constructor every test and bench uses, so the default here is
            // what keeps them byte-identical without touching a call site.
            transforms_act: false,
            inject_mutate_panic: false,
            policy: None,
            // Nothing wires anything in a test or a bench, so the standalone
            // default reads `pending` / unwired forever — the honest answer for
            // a listener no agent was ever pointed at.
            wiring: Arc::new(preflight::WiringState::default()),
            egress: crate::egress::EgressReporter::silent(egress),
        }
    }

    /// Share the daemon's forward-client handle instead of this state's own.
    ///
    /// Load-bearing twice over. A supervised restart rebuilds `BoundaryState` from the
    /// route resolved at daemon start, so without this a restart after a self-heal pass
    /// would silently revert the boundary to the dead proxy; and the shared handle is what
    /// lets a swap reach a listener that is already serving.
    pub fn with_client_handle(mut self, client: crate::egress::ClientHandle) -> Self {
        self.client = client;
        self
    }

    /// Report this listener's upstream transport outcomes into the daemon's
    /// shared egress health. Without it the boundary forwards exactly as before
    /// and records nothing.
    pub fn with_egress_reporter(mut self, egress: crate::egress::EgressReporter) -> Self {
        self.egress = egress;
        self
    }

    /// Override the upstream header-wait timeout (tests/benches only). Lets a
    /// test drive the [`HEADER_TIMEOUT`] path in milliseconds instead of the
    /// production minute.
    pub fn with_header_timeout(mut self, timeout: Duration) -> Self {
        self.header_timeout = timeout;
        self
    }

    /// Wire plan-02 measurement: share the daemon's session registry and the
    /// cloud event rail. Used on the production path (`daemon::serve_with_listener`)
    /// and by the emission tests. Without this the boundary forwards exactly as in
    /// plan 01 and emits nothing.
    pub fn with_measurement(
        mut self,
        registry: Arc<SessionRegistry>,
        cloud_tx: Option<Sender<CloudEvent>>,
    ) -> Self {
        self.registry = registry;
        self.cloud_tx = cloud_tx;
        self
    }

    /// Allow an acting `prefix_reorder` (L-0) rule to rewrite the forwarded
    /// request (`[boundary] transforms_act`, D-28).
    ///
    /// The single switch between "measure and forward verbatim" and "measure,
    /// and forward a rewritten body when an authored `enforce` L-0 rule matched
    /// a viable request". It gates **only** L-0; `history_trim` and
    /// `prompt_edit` are coerced to `observe` at bundle load whatever this is.
    pub fn with_transforms_act(mut self, act: bool) -> Self {
        self.transforms_act = act;
        self
    }

    /// Arm the mutation-step panic injection for THIS listener (D-28 bench and
    /// tests only). Proves a panic mid-mutation forwards the original bytes and
    /// keeps the measurement that already succeeded.
    pub fn with_inject_mutate_panic(mut self, inject: bool) -> Self {
        self.inject_mutate_panic = inject;
        self
    }

    /// Share the daemon's resident policy bundle so authored `request` rules —
    /// and their `select` narrowing — reach the transform engine.
    ///
    /// Without this the engine evaluates [`transforms::BASELINE_RULES`] only,
    /// and an authored rule's `select` has no effect whatsoever.
    ///
    /// Direction of dependency is deliberate: `core::policy` is a leaf module
    /// and must not import siblings, so boundary → policy is the only legal
    /// edge. This reads the handle; it never writes it.
    pub fn with_policy(mut self, policy: Option<PolicyHandle>) -> Self {
        self.policy = policy;
        self
    }

    /// Share the daemon's wiring gate so `GET /admin/boundary/status` reports
    /// the same verdict the supervisor acted on.
    ///
    /// The `Arc` outlives any single serve attempt on purpose: a boundary
    /// restart rebuilds `BoundaryState`, and a gate that reset to `pending` on
    /// every restart would tell `init` and `doctor` "no verdict yet" about a
    /// listener the supervisor has already judged.
    pub fn with_wiring(mut self, wiring: Arc<preflight::WiringState>) -> Self {
        self.wiring = wiring;
        self
    }

    /// Install the **resolved** per-format upstream map (D-04/D-05).
    ///
    /// Production's only way to install one; every `tests/` and `bench`
    /// construction stays on [`BoundaryState::new`] and keeps the fan-out above.
    /// The daemon calls `BoundaryConfig::upstream_for(fmt)` for every
    /// [`WireFormat::ALL`] variant and hands the complete map over, so the
    /// lookup can never miss.
    ///
    /// **THE ONE OWNER of the `String` → `Url` parse**, its warning and its
    /// per-format fallback. `BoundaryConfig::upstream_for` returns a `String`
    /// precisely so this stays single-owner: two owners with different fallbacks
    /// is what once routed Codex to a customer's Anthropic gateway.
    pub fn with_upstream_map(mut self, map: BTreeMap<String, String>) -> Self {
        // Walk the VARIANTS, never the input map's keys — a key no variant names
        // cannot be forwarded to anyway, and walking the input would leave a
        // missing variant unresolved.
        for fmt in WireFormat::ALL {
            // A `const` URL parsed at daemon start, not in the handler — the
            // same thing `boundary::default_upstream()` does today.
            //
            // NOT `.expect()`. `.claude/rules/error-handling.md` forbids it in
            // library code, and the reason bites here specifically: this runs
            // while the daemon is coming up, so a panic takes the listener down
            // and with it every agent session on the machine. The input is a
            // built-in constant and cannot fail today; the point is that a
            // future constant with a typo degrades to a working forward instead
            // of to no boundary at all — which is what "everything fails toward
            // the original bytes" means on the startup path.
            // NOT `.expect()`. `.claude/rules/error-handling.md` forbids it in
            // library code, and the reason bites here specifically: this runs
            // while the daemon is coming up, so a panic takes the listener down
            // and with it every agent session on the machine.
            //
            // And NOT a fallback to `self.fallback` either, however tempting a
            // one-liner it looks: this closure is what the present-but-
            // unparseable arm below falls back TO, so pointing it at the
            // gateway would route a Codex request to Anthropic by the exact
            // path the comment there forbids. A built-in constant that does not
            // parse is a defect in OUR constant, and there is no correct
            // destination for that format — so say so loudly and leave the
            // entry out rather than inventing one.
            let default = || reqwest::Url::parse(fmt.default_upstream()).ok();
            let resolved = match map.get(fmt.as_str()) {
                // WARN ONLY ON A PRESENT-BUT-UNPARSEABLE VALUE. An absent key
                // takes the default silently: the daemon supplies every key, so
                // a "not a valid URL" line for a key nobody wrote would be noise.
                None => default(),
                Some(v) => reqwest::Url::parse(v).ok().or_else(|| {
                    // ITS OWN format's default, never another format's entry and
                    // never `self.fallback` — falling back to the gateway here is
                    // exactly how a Codex request would leave for Anthropic.
                    tracing::warn!(
                        format = fmt.as_str(),
                        value = %v,
                        "[boundary.upstream] entry is not a valid URL — forwarding {} to {} instead",
                        fmt.as_str(),
                        fmt.default_upstream()
                    );
                    default()
                }),
            };
            // Only reachable if OUR OWN built-in constant does not parse, which
            // is a defect in this crate rather than in anyone's config. There is
            // no correct destination for the format at that point, so leave the
            // entry out and say so — `upstream_for` then answers with the
            // configured base, which is the same thing it did before this
            // builder existed.
            let Some(url) = resolved else {
                tracing::error!(
                    format = fmt.as_str(),
                    value = fmt.default_upstream(),
                    "[boundary.upstream] built-in default is not a valid URL — leaving this \
                     format unmapped"
                );
                continue;
            };
            if url.path() != "/" {
                // A based upstream is now PRESERVED (`proxy::join_upstream`) —
                // it used to be silently discarded, which is why this line
                // still exists. Say once, at startup, how the prefix will be
                // applied, because the rule decides what the provider receives:
                // the base is the API root, so the inbound's own `/v1` segment
                // is dropped rather than doubled.
                tracing::info!(
                    format = fmt.as_str(),
                    upstream = %url,
                    "[boundary.upstream] entry carries a path — requests are prefixed with {} \
                     and their own leading /v1 is dropped",
                    url.path()
                );
            }
            // The ChatGPT pairing rides on the FIRST-PARTY OpenAI base, and is
            // decided from the RESOLVED VALUE rather than from whether a key is
            // present. The daemon materializes every key before handing the map
            // over — `upstream_for` is called for all of `WireFormat::ALL`, so
            // `openai-responses` is always present and "did the operator
            // configure it?" is not a question the map's shape can answer.
            // Asking it that way pairs on no host at all — written that way
            // first, green on every unit test, and caught only by a live
            // daemon reporting `upstream_chatgpt: null`.
            //
            // Anything else in this entry is a destination the operator chose,
            // and it takes both auth modes with it: someone who put a gateway
            // in front of OpenAI put it there to see the traffic, and a request
            // slipping past it because of a header we read is the failure they
            // would never find.
            if matches!(fmt, WireFormat::OpenAiResponses) {
                let first_party = reqwest::Url::parse(wire_format::OPENAI_BASE)
                    .ok()
                    .is_some_and(|builtin| builtin == url);
                self.openai_chatgpt = first_party
                    .then(|| reqwest::Url::parse(wire_format::CHATGPT_BASE).ok())
                    .flatten();
                if first_party && self.openai_chatgpt.is_none() {
                    // Our own constant, so a defect in this crate rather than
                    // in anyone's config. Leaving it `None` costs a ChatGPT
                    // request its correct origin but keeps the listener up,
                    // which is the trade this whole startup path is written to
                    // make.
                    tracing::error!(
                        value = wire_format::CHATGPT_BASE,
                        "built-in ChatGPT upstream is not a valid URL — a ChatGPT-plan request \
                         will forward to the openai-responses upstream instead"
                    );
                }
            }
            self.upstream.insert(fmt.as_str(), url);
        }
        self
    }

    /// The upstream base for one wire format — a plain, infallible lookup.
    ///
    /// Never `Option`, never `expect`: the forward path answers a synthetic 502
    /// on a `None` upstream, and a panic here would violate *everything fails
    /// toward the original bytes*.
    pub fn upstream_for(&self, fmt: WireFormat) -> &reqwest::Url {
        self.upstream.get(fmt.as_str()).unwrap_or(&self.fallback)
    }

    /// The ChatGPT backend this listener would use, when the built-in pair is
    /// in force. `None` when an operator configured `openai-responses`.
    ///
    /// Exists for the status endpoint. A second real destination that no
    /// diagnostic names is a destination nobody can check, and this one is
    /// reached by exactly the requests the operator is least expecting.
    pub fn chatgpt_upstream(&self) -> Option<&reqwest::Url> {
        self.openai_chatgpt.as_ref()
    }

    /// The upstream base for ONE REQUEST: its format's entry, or the ChatGPT
    /// backend when the request's own credential names it.
    ///
    /// The forward path calls this and never [`Self::upstream_for`] directly —
    /// the format alone cannot answer where an `openai-responses` request goes,
    /// because a ChatGPT-plan token and a platform key ride the same route to
    /// different origins and each is refused by the other's.
    ///
    /// Falls through to the format's entry whenever the pair does not apply: a
    /// non-OpenAI format, an operator-configured `openai-responses`, or a build
    /// whose constant failed to parse. So the worst case is exactly the
    /// behaviour before this existed.
    pub fn upstream_for_request(&self, fmt: WireFormat, auth: AuthMode) -> &reqwest::Url {
        match (fmt, auth) {
            (WireFormat::OpenAiResponses, AuthMode::ChatGptSubscription) => self
                .openai_chatgpt
                .as_ref()
                .unwrap_or_else(|| self.upstream_for(fmt)),
            _ => self.upstream_for(fmt),
        }
    }

    /// The request rules resident right now, or an empty slice.
    ///
    /// Cheap per request: one `ArcSwap` load, no clone of the rule set. Every
    /// rule here is already gate-validated and mode-coerced by
    /// [`crate::core::policy::ResidentBundle::from_bundle`], so this side never
    /// re-validates and never re-checks `mode`.
    pub fn resident_request_rules(&self) -> Option<arc_swap::Guard<Arc<Option<ResidentBundle>>>> {
        self.policy.as_ref().map(|handle| handle.load())
    }
}

/// The production upstream base URL as a parsed `reqwest::Url`.
///
/// # Panics
///
/// Never in practice — [`ANTHROPIC_BASE`] is a compile-time constant valid URL.
pub fn default_upstream() -> reqwest::Url {
    reqwest::Url::parse(ANTHROPIC_BASE).expect("ANTHROPIC_BASE is a valid URL")
}

/// The default boundary port — what an instance binds unless `[boundary] port`
/// says otherwise. **Never re-probed** (D-25).
///
/// This used to be the only answer, and the reason was two-owner divergence:
/// `init` wrote `ANTHROPIC_BASE_URL` while the supervised daemon bound the port,
/// in different environments (a shell variable set at `init` is not inherited by
/// a launchd/systemd start after a reboot), so any ambient override could make
/// the written value and the bound value disagree. The daemon now does both, and
/// writes what it just bound, so they cannot disagree — see
/// [`crate::config::BoundaryConfig::port`]. D-25 is untouched: the port is still
/// never silently re-probed onto a different one when the bind fails; it fails.
///
/// Callers that have a `Config` should read `cfg.boundary.port` instead — this
/// is the default, not the effective value. Tests bind ephemeral ports via
/// [`serve_ephemeral`] rather than pinning this one.
pub fn resolve_boundary_port() -> u16 {
    default_boundary_port()
}

/// The port that counts as "the default" for this process.
///
/// [`DEFAULT_BOUNDARY_PORT`] in production. Redirectable only through
/// `OPENLATCH_BOUNDARY_DEFAULT_PORT`, and only for one reason: the wiring
/// invariant can only be tested on the default port — a non-default one is an
/// isolated instance that deliberately never writes the agent config — so those
/// tests had to contend for the single real 7600 and skipped themselves whenever
/// a developer box already had something on it. A test that skips proves
/// nothing, and the ones that skipped were the ones guarding the bug.
///
/// **D-25 is intact.** The invariant was never that the number is 7600; it was
/// that there is exactly one of it. [`crate::config::BoundaryConfig::default`]
/// and [`crate::config::BoundaryConfig::owns_agent_wiring`] both read THIS
/// function, so the port the daemon binds, the port it writes into the agent
/// config, and the port that decides who owns that config are one value. The
/// port is still never *silently re-probed* onto a different one when a bind
/// fails: it fails.
///
/// An unparsable value falls back to the constant rather than failing: this is
/// read on the daemon's startup path, and a typo in a variable nobody sets in
/// production must not be able to keep the boundary from coming up.
pub fn default_boundary_port() -> u16 {
    match std::env::var("OPENLATCH_BOUNDARY_DEFAULT_PORT") {
        Ok(v) => v.trim().parse::<u16>().unwrap_or(DEFAULT_BOUNDARY_PORT),
        Err(_) => DEFAULT_BOUNDARY_PORT,
    }
}

/// Build the boundary's forward client.
///
/// Mirrors `cloud::worker::build_cloud_client` (connection-pooled, rustls-only,
/// no OpenSSL) with **one deliberate difference**: NO overall `timeout()`. A
/// streamed `/v1/messages` response can run far longer than any fixed deadline
/// (long agent turns, large tool outputs); a `timeout()` would truncate the SSE
/// stream mid-flight. A `connect_timeout` still bounds the reach-upstream phase,
/// so a dead provider surfaces as a synthetic 502 rather than an indefinite hang.
pub fn build_boundary_client() -> Option<reqwest::Client> {
    build_boundary_client_with(&crate::egress::EgressConfig::direct())
}

/// Build the boundary's forward client on a given egress route.
///
/// Split from [`build_boundary_client`] rather than replacing it: `BoundaryState::new` has
/// twenty callers, almost all of them tests that want a hermetic direct client and no
/// config in scope. Keeping the no-argument constructor means none of them change, and the
/// one production caller -- the daemon's supervised task factory, which holds the resolved
/// config and rebuilds on every restart -- opts in explicitly.
pub fn build_boundary_client_with(cfg: &crate::egress::EgressConfig) -> Option<reqwest::Client> {
    match crate::egress::build_client(crate::egress::Consumer::Boundary, cfg) {
        Ok(client) => Some(client),
        // D-21 is the one error that must NOT fail static. `allow_direct = false` says the
        // ladder never ends at DIRECT, and a direct fallback here would be exactly the
        // silent direct connection the setting exists to forbid — on the one consumer whose
        // upstream is a third-party provider. The listener keeps running and answers a
        // synthetic 502 per request, which is loud, local, and reversible the moment a
        // route exists; `doctor` names OL-1227.
        Err(e) if e.code == crate::error::ERR_DIRECT_FORBIDDEN => {
            tracing::error!(
                code = %e.code,
                error = %e.message,
                "boundary has no permitted egress route; forwarding is refused rather than                  falling back to a direct connection"
            );
            None
        }
        Err(e) => {
            // Fail-static, the posture this whole layer is built on: a proxy
            // misconfiguration must not take the boundary down, because a boundary that is
            // down means every agent request on this host fails. The error is loud, the
            // forwarder keeps running, and `doctor` is where the operator reads why.
            tracing::error!(
                code = %e.code,
                error = %e.message,
                "boundary egress client could not be built from the configured proxy                  route; forwarding directly instead"
            );
            crate::egress::client_builder()
                .connect_timeout(Duration::from_secs(10))
                .pool_max_idle_per_host(8)
                .build()
                .ok()
                .or_else(|| Some(crate::egress::client()))
        }
    }
}

/// Build the boundary `Router`.
///
/// **Every** path proxies via the `fallback` (`proxy_any`) — unknown paths
/// forward blind, never rejected (D-08). The only locally-served route is the
/// admin status surface; it is loopback-only by construction (the listener
/// binds `127.0.0.1`) and returns non-sensitive liveness data. The
/// `DefaultBodyLimit` here is belt-and-suspenders: the real ceiling is the
/// manual `Content-Length` check plus the explicit `to_bytes(_, 32 MB)` limit
/// in `proxy_any` (the layer is enforced by body *extractors*, and the opaque
/// path never runs an extractor).
pub fn router(state: Arc<BoundaryState>) -> Router {
    Router::new()
        .route("/admin/boundary/status", get(proxy::boundary_status))
        .fallback(proxy::proxy_any)
        .layer(DefaultBodyLimit::max(MAX_MATERIALIZE_BYTES))
        .with_state(state)
}

/// Serve the boundary router on an ephemeral loopback port; returns the bound
/// port. Test/bench support — spawns the server as a detached task and returns
/// once the port is bound (so callers can connect immediately).
pub async fn serve_ephemeral(state: Arc<BoundaryState>) -> u16 {
    let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
    let port = listener.local_addr().unwrap().port();
    let app = router(state);
    tokio::spawn(async move {
        let _ = axum::serve(listener, app).await;
    });
    port
}

/// Bind the pinned loopback port for the boundary listener.
///
/// Loopback ONLY — there is no `OPENLATCH_BIND_ALL` escape hatch here, unlike
/// the hook daemon (F-22). Occupied → loud [`OlError::port_occupied`] (D-25);
/// the caller must NOT re-probe another port.
///
/// **This is the gate the agent wiring hangs off.** `ANTHROPIC_BASE_URL` is
/// written by the daemon only after this call returns `Ok`, and removed when
/// the listener goes away — so the config never advertises a port nobody holds
/// (see `daemon::serve_with_listener`). Binding first and wiring second is what
/// makes that ordering enforceable rather than a convention.
///
/// > **D-01 loopback transport.** The daemon writes a plain
/// > `http://127.0.0.1:PORT` base URL (the plan's default). If a real Claude
/// > Code session is found to REQUIRE HTTPS on the loopback, the fallback — per
/// > D-01 / the PRD transport spike — is: generate a self-signed cert scoped to
/// > `127.0.0.1`, serve TLS here, and write `NODE_EXTRA_CA_CERTS` **per-agent**
/// > (NEVER the system trust store). That branch is fully specified but
/// > intentionally NOT built: the empirical `http://`-vs-HTTPS loopback spike
/// > requires a live Claude Code session and remains the one outstanding open
/// > item (matches the HANDOFF).
pub async fn bind_pinned(port: u16) -> Result<TcpListener, OlError> {
    let addr = SocketAddr::from(([127, 0, 0, 1], port));
    TcpListener::bind(addr)
        .await
        .map_err(|e| OlError::port_occupied(port, e))
}

/// Serve the boundary router on an ALREADY-BOUND `listener` until `shutdown_rx`
/// flips to `true` (or its sender is dropped), then drain and return.
///
/// Deliberately takes the listener rather than binding one: the daemon binds the
/// pinned port up-front — outside its retry loop — so a first-bind failure is a
/// startup error and the agent config is never written (see [`bind_pinned`]).
/// A bind-and-serve helper would put the bind back inside the retry loop, which
/// is exactly the shape that let the daemon advertise a port it never held.
///
/// The `/shutdown` endpoint only stops the hook server on the daemon's main
/// port; the boundary binds a **separate** pinned port (7600), so without this
/// signal it would keep that port bound after the hook server drains — the
/// process never exits and `openlatch stop` fails with OL-1300 "process still
/// running". Wiring the SAME teardown into both listeners is the fix.
pub async fn serve_bound(
    listener: TcpListener,
    state: Arc<BoundaryState>,
    shutdown_rx: tokio::sync::watch::Receiver<bool>,
) -> Result<(), OlError> {
    tracing::info!(
        port = state.port,
        upstream = %state.upstream_for(WireFormat::AnthropicMessages),
        "boundary serving (loopback only)"
    );
    serve_until_shutdown(listener, state, shutdown_rx).await
}

/// One attempt of the daemon's supervised boundary task.
///
/// Takes the pre-bound listener if it is still there (the first attempt, whose
/// bind already succeeded before the daemon wrote any agent config), and rebinds
/// the pinned port otherwise (every restart after a mid-life serve error, which
/// is also what waits out a Windows TIME_WAIT).
///
/// The daemon calls exactly this, so the retry semantics the tests assert are
/// the retry semantics that ship — the two cannot drift into agreement-by-copy.
pub async fn serve_attempt(
    pre_bound: Arc<tokio::sync::Mutex<Option<TcpListener>>>,
    state: Arc<BoundaryState>,
    shutdown_rx: tokio::sync::watch::Receiver<bool>,
) -> Result<(), OlError> {
    let listener = match pre_bound.lock().await.take() {
        Some(l) => l,
        None => bind_pinned(state.port).await?,
    };
    serve_bound(listener, state, shutdown_rx).await
}

async fn serve_until_shutdown(
    listener: TcpListener,
    state: Arc<BoundaryState>,
    mut shutdown_rx: tokio::sync::watch::Receiver<bool>,
) -> Result<(), OlError> {
    let app = router(state);
    axum::serve(listener, app)
        .with_graceful_shutdown(async move {
            // Resolve when the shared teardown flag flips to `true`. `wait_for`
            // also returns (an `Err`) if the sender is dropped, so a daemon torn
            // down without an explicit signal still stops the listener rather
            // than hanging.
            let _ = shutdown_rx.wait_for(|stop| *stop).await;
        })
        .await
        .map_err(|e| OlError::new(ERR_BOUNDARY_SERVE, format!("boundary serve failed: {e}")))
}

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

    #[test]
    fn default_upstream_parses() {
        let u = default_upstream();
        assert_eq!(u.as_str(), "https://api.anthropic.com/");
    }

    /// A configured upstream carrying a path is now PRESERVED
    /// (`proxy::join_upstream`); it used to be silently dropped, so their
    /// traffic landed on `https://gw.example/v1/messages` with the
    /// `/anthropic` prefix gone.
    ///
    /// Asserting the line FIRES is the point: an "is not doubled" assertion is
    /// green on every build, including one that says nothing at all. The line
    /// has to name the prefix, because how it is applied — the base is the API
    /// root, so the inbound `/v1` is dropped — decides what the provider
    /// receives.
    #[test]
    fn upstream_with_a_path_reports_how_the_prefix_is_applied() {
        let mut map = BTreeMap::new();
        map.insert(
            WireFormat::AnthropicMessages.as_str().to_string(),
            "https://gw.example/anthropic".to_string(),
        );

        let (state, logs) = crate::core::policy::test_support::capture_logs(|| {
            BoundaryState::new(default_upstream(), 0, 8, &[]).with_upstream_map(map)
        });

        assert!(
            logs.contains("carries a path"),
            "the prefix must be reported, got: {logs}"
        );
        assert!(
            logs.contains("/anthropic"),
            "the line must name the prefix requests will carry, got: {logs}"
        );
        // The entry is still installed — the line explains the shape, it does
        // not refuse the configuration.
        assert_eq!(
            state.upstream_for(WireFormat::AnthropicMessages).host_str(),
            Some("gw.example")
        );
        assert_eq!(
            state.upstream_for(WireFormat::AnthropicMessages).path(),
            "/anthropic",
            "the prefix must survive into the installed entry — it is what the \
             forward path prepends"
        );
    }

    /// The ChatGPT backend is a second BUILT-IN for `openai-responses`, picked
    /// by the request's own credential. It arrives with the resolved map, so a
    /// host works on either Codex plan without being re-wired.
    #[test]
    fn a_chatgpt_credential_selects_the_chatgpt_backend() {
        let state =
            BoundaryState::new(default_upstream(), 0, 8, &[]).with_upstream_map(BTreeMap::new());

        assert_eq!(
            state
                .upstream_for_request(WireFormat::OpenAiResponses, AuthMode::ChatGptSubscription)
                .as_str(),
            "https://chatgpt.com/backend-api/codex",
        );
        // A platform key on the same route keeps OpenAI's first-party base.
        assert_eq!(
            state
                .upstream_for_request(WireFormat::OpenAiResponses, AuthMode::Platform)
                .host_str(),
            Some("api.openai.com"),
        );
        // The pair is scoped to the OpenAI format: an Anthropic request is
        // never moved by a header it does not send.
        assert_eq!(
            state
                .upstream_for_request(WireFormat::AnthropicMessages, AuthMode::ChatGptSubscription)
                .host_str(),
            Some("api.anthropic.com"),
        );
    }

    /// **The map the DAEMON actually hands over.** `upstream_for` is called for
    /// every `WireFormat::ALL` variant before the map is built, so every key is
    /// present and carries its built-in default on a host that configured
    /// nothing.
    ///
    /// The trap this exists for: keying the pairing on "is this key absent?" is
    /// green against an empty test map and pairs on no real host at all —
    /// `upstream_chatgpt` came back `null` from a live daemon while every unit
    /// test above passed. An empty map is not the production shape, so it
    /// cannot be the only shape under test.
    #[test]
    fn the_pairing_survives_a_fully_materialized_map() {
        let cfg = crate::config::BoundaryConfig::default();
        let map: BTreeMap<String, String> = WireFormat::ALL
            .iter()
            .map(|f| (f.as_str().to_string(), cfg.upstream_for(*f)))
            .collect();
        assert!(
            map.contains_key(WireFormat::OpenAiResponses.as_str()),
            "the daemon materializes every key — if this ever stops being true \
             the pairing's precondition changed"
        );

        let state = BoundaryState::new(default_upstream(), 0, 8, &[]).with_upstream_map(map);

        assert_eq!(
            state.chatgpt_upstream().map(reqwest::Url::as_str),
            Some("https://chatgpt.com/backend-api/codex"),
            "a default host must pair, or no ChatGPT-plan Codex install works"
        );
    }

    /// An explicit `openai-responses` entry names ONE destination for this
    /// host's OpenAI traffic, and takes BOTH auth modes with it.
    ///
    /// The gateway case: an operator who put a proxy in front of OpenAI put it
    /// there to see the traffic, and a request that slipped past it to
    /// `chatgpt.com` because of a header we read is the failure they would
    /// never find.
    #[test]
    fn a_configured_openai_upstream_wins_over_the_chatgpt_builtin() {
        let mut map = BTreeMap::new();
        map.insert(
            WireFormat::OpenAiResponses.as_str().to_string(),
            "https://gw.example".to_string(),
        );

        let state = BoundaryState::new(default_upstream(), 0, 8, &[]).with_upstream_map(map);

        for auth in [AuthMode::ChatGptSubscription, AuthMode::Platform] {
            assert_eq!(
                state
                    .upstream_for_request(WireFormat::OpenAiResponses, auth)
                    .host_str(),
                Some("gw.example"),
                "a configured gateway must not be bypassed for {auth:?}"
            );
        }
    }

    /// An unparseable entry falls back to ITS OWN format's built-in default —
    /// never to another format's entry and never to the constructor's URL.
    ///
    /// The round-5 misroute guard. "Log and skip" left the lookup missing,
    /// `unwrap_or(&self.fallback)` answered with the gateway `new` was given,
    /// and every Codex request left for the customer's *Anthropic* gateway.
    #[test]
    fn unparseable_entry_falls_back_to_its_own_default() {
        let mut map = BTreeMap::new();
        map.insert(
            WireFormat::AnthropicMessages.as_str().to_string(),
            "https://gw.example".to_string(),
        );
        map.insert(
            WireFormat::OpenAiResponses.as_str().to_string(),
            "bogus".to_string(),
        );

        let (state, logs) = crate::core::policy::test_support::capture_logs(|| {
            BoundaryState::new(
                reqwest::Url::parse("https://gw.example").expect("url"),
                0,
                8,
                &[],
            )
            .with_upstream_map(map)
        });

        assert_eq!(
            state.upstream_for(WireFormat::OpenAiResponses).host_str(),
            Some("api.openai.com"),
            "a bad Responses entry takes OpenAI's default, never the gateway"
        );
        assert_eq!(
            state.upstream_for(WireFormat::AnthropicMessages).host_str(),
            Some("gw.example"),
            "the good entry is untouched"
        );
        assert!(
            logs.contains("not a valid URL"),
            "a present-but-unparseable value must say so, got: {logs}"
        );
        assert!(
            !logs.contains("format=\"unknown\"") && !logs.contains("format: \"unknown\""),
            "an ABSENT key takes its default silently — no spurious warning: {logs}"
        );
    }

    /// One URL populates EVERY format, not a single-entry map.
    ///
    /// This is what keeps `tests/boundary_forward.rs::boundary_at` and
    /// `preflight::tests::state_for` working untouched: a probe or a forward for
    /// ANY format through a test state reaches the one base it was given, rather
    /// than leaving the machine for `api.openai.com`.
    #[test]
    fn one_url_populates_every_format() {
        let base = reqwest::Url::parse("http://127.0.0.1:1").expect("url");
        let state = BoundaryState::new(base.clone(), 0, 8, &[]);
        for fmt in WireFormat::ALL {
            assert_eq!(state.upstream_for(fmt), &base, "{fmt:?}");
        }
    }

    #[test]
    fn resolve_boundary_port_is_deterministic() {
        // The pinned default is returned every call, with no ambient override —
        // the property D-25 leans on (never re-probed, config and bind agree).
        assert_eq!(resolve_boundary_port(), DEFAULT_BOUNDARY_PORT);
        assert_eq!(resolve_boundary_port(), DEFAULT_BOUNDARY_PORT);
    }

    #[tokio::test]
    async fn boundary_listener_terminates_on_shutdown_signal() {
        // OL-1300 regression guard. Serve on an EPHEMERAL loopback port (never
        // the pinned 7600 — a live daemon may hold it) and prove the listener
        // returns promptly once the shared shutdown flag flips, releasing the
        // port. This is exactly the signal `/shutdown` broadcasts to the hook
        // server; wiring it here is what lets `openlatch stop` succeed.
        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let state = Arc::new(BoundaryState::new(default_upstream(), port, 8, &[]));

        let serve = tokio::spawn(serve_bound(listener, state, shutdown_rx));

        // While serving, the port is held — a competing bind fails loudly.
        assert!(
            bind_pinned(port).await.is_err(),
            "port must be held while the boundary is serving"
        );

        // Fire the same teardown the hook server observes on `/shutdown`.
        shutdown_tx.send(true).unwrap();

        // The serve future must resolve well within the daemon's 5 s bounded
        // await, and return Ok from a clean graceful shutdown.
        let joined = tokio::time::timeout(Duration::from_secs(5), serve)
            .await
            .expect("boundary listener did not terminate after shutdown signal")
            .expect("boundary serve task panicked");
        assert!(joined.is_ok(), "graceful shutdown should return Ok");

        // The port is released — re-binding it now succeeds.
        assert!(
            bind_pinned(port).await.is_ok(),
            "port must be free after graceful shutdown"
        );
    }

    /// A bind failure AFTER the first successful one must be **transient**, not
    /// terminal.
    ///
    /// The boundary task used to be spawned one-shot: an occupied port logged a
    /// single ERROR and the task returned forever. With `ANTHROPIC_BASE_URL`
    /// pointing every agent on the machine at this listener, that meant
    /// universal ECONNREFUSED while the hook daemon kept answering `/health`
    /// with `ok`. Under the supervisor the same `Err` is just another retry.
    ///
    /// The FIRST bind no longer takes this path — the daemon binds it up-front
    /// and refuses to start on failure, so the config is never written against a
    /// port we do not hold. Everything after it still does, which is what this
    /// test drives: an empty `pre_bound` slot is precisely the state every
    /// restart sees.
    ///
    /// Squat an ephemeral port, prove the supervised listener keeps retrying it,
    /// then free the port and prove it binds and serves — without ever touching
    /// the pinned 7600 a live daemon may hold. This is defect #3 (permanent bind
    /// failure) and #6 (Windows TIME_WAIT rebind) reproduced directly: both are
    /// "the bind fails now and would succeed later".
    #[tokio::test]
    async fn supervised_boundary_retries_a_failed_bind_and_recovers() {
        use crate::core::supervision::task::{
            spawn_supervised, Backoff, HealthRegistry, RestartPolicy, TaskSpec,
        };

        // Hold the port so the first (and next several) binds fail.
        let squatter = TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
        let port = squatter.local_addr().unwrap().port();

        let registry = Arc::new(HealthRegistry::new());
        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        // Milliseconds instead of the production 1s→60s so the test does not
        // spend a minute proving a property that has nothing to do with wall time.
        let spec = TaskSpec::new("boundary", RestartPolicy::Always).with_backoff(Backoff::new(
            Duration::from_millis(20),
            Duration::from_millis(60),
        ));

        // Empty: the post-first-attempt state, where every run must rebind.
        let pre_bound = Arc::new(tokio::sync::Mutex::new(None));
        let task_shutdown_rx = shutdown_rx.clone();
        let handle = spawn_supervised(&registry, spec, shutdown_rx, move || {
            let state = Arc::new(BoundaryState::new(default_upstream(), port, 4, &[]));
            serve_attempt(pre_bound.clone(), state, task_shutdown_rx.clone())
        });

        // While the squatter holds the port every attempt fails — and keeps
        // being retried rather than giving up after the first.
        //
        // Polled to a deadline rather than slept for a fixed span. The property
        // is a *count* — "more than one attempt" — and nothing about it is
        // wall-clock; a fixed window only ever approximated it. Under
        // tarpaulin's ptrace instrumentation a 250ms window sized for several
        // 20–60ms retries fits exactly one, which failed the coverage job with
        // all 1545 other tests passing.
        //
        // Both halves are read in the same iteration on purpose. `is_degraded()`
        // is `state != Running`, and the supervisor sets `Running` at the top of
        // every attempt, before that attempt's bind fails — so a poll that exits
        // on the restart count alone can land mid-attempt and see a boundary
        // that reads healthy while it is anything but.
        //
        // The entry is looked up *inside* the loop rather than indexed once
        // before it: the supervised task registers asynchronously, so
        // `tasks()[0]` on the first iteration is a panic waiting for a slow
        // scheduler — the one under `tarpaulin` above, for instance.
        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        let mut retried_while_degraded = false;
        let mut health = None;
        while std::time::Instant::now() < deadline {
            if let Some(h) = registry.tasks().first().cloned() {
                retried_while_degraded = h.restarts() >= 2 && registry.is_degraded();
                health = Some(h);
                if retried_while_degraded {
                    break;
                }
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        let health = health.expect("the supervised task never registered with the health registry");
        assert!(
            retried_while_degraded,
            "an occupied port must be retried and read as degraded, \
             saw {} restarts and degraded={}",
            health.restarts(),
            registry.is_degraded()
        );
        // Safe to read after the fact: the supervisor records the error before
        // it increments the restart counter, so a restart implies an error.
        let recorded = health.last_error().unwrap_or_default();
        assert!(
            recorded.contains(&port.to_string()),
            "the bind failure must be recorded on the health entry, got {recorded:?}"
        );

        // Free the port: the next attempt inside the backoff window must bind.
        drop(squatter);

        let client = crate::egress::client_builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let url = format!("http://127.0.0.1:{port}/admin/boundary/status");
        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        let mut served = false;
        while std::time::Instant::now() < deadline {
            if let Ok(r) = client.get(&url).send().await {
                if r.status().is_success() {
                    served = true;
                    break;
                }
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
        assert!(
            served,
            "the boundary must bind and serve once the port is released"
        );

        shutdown_tx.send(true).expect("shutdown send");
        let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
    }

    /// The first attempt serves the listener the daemon already bound — it does
    /// NOT bind again.
    ///
    /// That ordering is what the whole invariant rests on: the daemon binds,
    /// then writes `ANTHROPIC_BASE_URL`, then hands the live listener to the
    /// supervised task. If the task re-bound instead, there would be a window
    /// where the config names a port nothing holds, and a squatter arriving in
    /// that window would win it.
    #[tokio::test]
    async fn first_attempt_serves_the_prebound_listener_without_rebinding() {
        let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let pre_bound = Arc::new(tokio::sync::Mutex::new(Some(listener)));

        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let state = Arc::new(BoundaryState::new(default_upstream(), port, 4, &[]));
        let serve = tokio::spawn(serve_attempt(pre_bound.clone(), state, shutdown_rx));

        // Serving on the handed-down listener: the admin surface answers, and
        // the slot is empty so any restart would have to rebind.
        let client = crate::egress::client_builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let url = format!("http://127.0.0.1:{port}/admin/boundary/status");
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        let mut served = false;
        while std::time::Instant::now() < deadline {
            if let Ok(r) = client.get(&url).send().await {
                if r.status().is_success() {
                    served = true;
                    break;
                }
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
        assert!(served, "the pre-bound listener must be served as-is");
        assert!(
            pre_bound.lock().await.is_none(),
            "the pre-bound listener is consumed by the first attempt only"
        );

        shutdown_tx.send(true).expect("shutdown send");
        let _ = tokio::time::timeout(Duration::from_secs(5), serve).await;
    }

    #[tokio::test]
    async fn bind_pinned_is_loud_when_occupied() {
        // First bind wins; the second bind of the SAME port fails loudly with
        // the D-25 code rather than silently re-probing elsewhere.
        let probe = TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
        let port = probe.local_addr().unwrap().port();
        drop(probe);

        let held = bind_pinned(port).await.expect("first bind succeeds");
        let occupied = bind_pinned(port).await;
        assert!(occupied.is_err(), "second bind of a held port must fail");
        assert_eq!(
            occupied.unwrap_err().code,
            crate::error::ERR_BOUNDARY_PORT_IN_USE
        );
        drop(held);
    }
}