openlatch-client 0.1.18

//! 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 preflight;
pub mod proxy;
pub mod retention;
pub mod session;
pub mod tokenize;
pub mod transforms;

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

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;

/// 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 {
    /// Built ONCE, connection-pooled. NO overall `timeout()` on the streaming
    /// path — a stream can outlive any fixed deadline. See [`build_boundary_client`].
    pub client: reqwest::Client,
    /// Upstream provider base (`https://api.anthropic.com` in production; a
    /// plain-`http://` mock in tests/benches).
    pub upstream_base: 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-09). 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>,
    /// 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>,
}

impl BoundaryState {
    /// Build boundary state with a freshly-constructed forward client.
    pub fn new(
        upstream_base: reqwest::Url,
        port: u16,
        inflight: usize,
        extra_patterns: &[String],
    ) -> Self {
        Self {
            client: build_boundary_client(),
            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()),
            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()),
        }
    }

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

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

    /// 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 unparseable 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() -> reqwest::Client {
    reqwest::Client::builder()
        .connect_timeout(Duration::from_secs(10))
        .pool_max_idle_per_host(8)
        .use_rustls_tls()
        .build()
        .expect("failed to build boundary reqwest 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_base, "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/");
    }

    #[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.
        tokio::time::sleep(Duration::from_millis(250)).await;
        let health = registry.tasks()[0].clone();
        assert!(
            health.restarts() >= 2,
            "an occupied port must be retried, saw {} restarts",
            health.restarts()
        );
        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:?}"
        );
        assert!(
            registry.is_degraded(),
            "a boundary that cannot bind must read as degraded, not ok"
        );

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

        let client = reqwest::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 = reqwest::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);
    }
}