openlatch-client 0.1.18

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

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Mutex;
use std::time::Duration;

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

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

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

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

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

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

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

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

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

/// Process-wide view of the wiring gate, shared by the three parties that must
/// agree on it: the supervisor that opens and closes it, the admin status
/// endpoint that reports it, and — through that endpoint — `init` and `doctor`.
///
/// Deliberately NOT a field on the per-attempt `BoundaryState`: a boundary
/// restart rebuilds that struct, and the verdict has to survive one.
#[derive(Debug, Default)]
pub struct WiringState {
    wired: AtomicBool,
    verdict: Mutex<Verdict>,
}

impl WiringState {
    /// Whether the agent config currently points at this listener.
    pub fn is_wired(&self) -> bool {
        self.wired.load(Ordering::Relaxed)
    }

    /// Record the wiring state after a successful write / removal.
    pub fn set_wired(&self, wired: bool) {
        self.wired.store(wired, Ordering::Relaxed);
    }

    /// The most recent probe verdict.
    ///
    /// A poisoned lock degrades to the value it was holding rather than
    /// panicking: this is read from the admin handler on the boundary's own
    /// runtime, and the gate's observability must never be able to take the
    /// listener down.
    pub fn verdict(&self) -> Verdict {
        match self.verdict.lock() {
            Ok(v) => v.clone(),
            Err(poisoned) => poisoned.into_inner().clone(),
        }
    }

    /// Store a fresh probe verdict.
    pub fn set_verdict(&self, verdict: Verdict) {
        match self.verdict.lock() {
            Ok(mut v) => *v = verdict,
            Err(poisoned) => *poisoned.into_inner() = verdict,
        }
    }
}

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

    let url = format!("http://127.0.0.1:{port}/v1/messages");
    let sent = client
        .post(&url)
        .header("content-type", "application/json")
        .header("anthropic-version", "2023-06-01")
        .header("x-api-key", PREFLIGHT_API_KEY)
        .header(PREFLIGHT_HEADER, "1")
        .body(PREFLIGHT_BODY)
        .send()
        .await;

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

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

    Ok(())
}

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

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

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

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

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

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

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

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

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

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

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

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