openlatch-client 0.3.3

//! The `egress-monitor` supervised task.
//!
//! Two jobs. It keeps the published [`EgressStatus`] honest and says out loud when it
//! moves, and — when a resolver is wired — it runs the **self-heal** pass that gets a
//! roaming host back onto a working route (D-18). Outcomes are recorded in place by the
//! consumers themselves, so the observer half never touches the network; the self-heal
//! half is the only thing here that does, and only while egress is failed.
//!
//! # Why it is a task at all
//!
//! Three reasons a plain function would not do:
//!
//! - A transition needs to be logged **once**, with the code and the masked route. Doing
//!   that inside `record_failure` would emit a line per failed request across every
//!   consumer.
//! - `/health` reports a `RestartPolicy::Always` subsystem that is not running as degraded,
//!   so an operator can see the egress observer is alive — including on a host that has
//!   never authenticated, where no cloud worker exists at all.
//! - Self-heal is a backing-off loop that must survive a panic and must stop on shutdown.
//!
//! # Where the probe is
//!
//! Not here — deliberately, and this is the plan's flagged deviation. The cloud worker
//! already runs a health probe against `{api_url}/api/v1/health` on a 60 s cadence
//! (`CloudState::next_health_delay`). A second loop would double-probe the platform from
//! one process. What this plan adds instead is the missing half: that probe is now
//! **suppressed while traffic is flowing** ([`EgressState::is_idle`]) and **retried once
//! immediately after a failure**, and its outcomes feed this state. The probe therefore has
//! exactly one owner, and it is the task that already had it.
//!
//! The self-heal pass issues its own request, which is a different question: not "is the
//! current route alive" but "does *this candidate* work", and the answer has to come from
//! the candidate's own client before anything is swapped.

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

use tokio::sync::watch;

use crate::core::error::{ERR_DIRECT_FORBIDDEN, ERR_PROXY_UNREACHABLE};
use crate::core::supervision::task::Backoff;

use super::clients::EgressClients;
use super::config::{EgressConfig, ProxySource};
use super::state::{mask_text, EgressSnapshot, EgressState, EgressStatus};
use super::ProxyResolver;

/// How often the monitor re-derives the published status without being woken.
///
/// A backstop, not the main path: [`EgressState::record_failure`] wakes the loop
/// immediately. The `cfg(test)` value is what lets the unit tests below assert the loop's
/// behaviour without a wall-clock wait, exactly as the cloud worker's `health_delay` does.
#[cfg(not(test))]
const TICK: Duration = Duration::from_secs(30);
#[cfg(test)]
const TICK: Duration = Duration::from_millis(20);

/// The floor of the self-heal backoff — the shortest gap between two discovery passes.
pub const HEAL_BACKOFF_INITIAL: Duration = Duration::from_secs(60);

/// The ceiling of the self-heal backoff.
///
/// **DEVIATION, deliberate.** `.claude/rules/subsystem-supervision.md` caps backoff at
/// 60 s, and that sentence is about *restarting a task*: a task that died wants to be back
/// promptly, and retrying it costs nothing outside this process. Re-running **discovery**
/// is a different act. Off-network it re-walks the OS proxy rungs, which on Windows and
/// macOS includes WPAD auto-detection, and WPAD is a documented attack vector (D-25): a
/// laptop on an untrusted network that re-asks for a proxy configuration every minute,
/// forever, is repeatedly offering itself to whoever answers first. Fifteen minutes is
/// D-18's own ceiling, and it is what this uses.
pub const HEAL_BACKOFF_MAX: Duration = Duration::from_secs(900);

/// The backoff a self-heal streak runs under.
///
/// The `cfg(test)` variant compresses the same curve into milliseconds; the shipped
/// constants are asserted directly, on `Backoff::base()`, in the tests below.
#[cfg(not(test))]
fn heal_backoff() -> Backoff {
    Backoff::new(HEAL_BACKOFF_INITIAL, HEAL_BACKOFF_MAX)
}
#[cfg(test)]
fn heal_backoff() -> Backoff {
    Backoff::new(Duration::from_millis(5), Duration::from_millis(40))
}

/// Everything a self-heal pass needs.
///
/// Cloned per supervised restart, so every field is cheap to clone and none of it carries
/// per-run state — the backoff belongs to the run, which is the right behaviour: a task
/// that just panicked should not inherit a fifteen-minute wait.
#[derive(Clone)]
pub struct SelfHeal {
    /// The discovery ladder, in the daemon's own (service) context.
    ///
    /// **I-2 gated.** `None` on this branch and in every test that does not inject a fake:
    /// [`ProxyResolver`] is a seam with no implementation yet, and I-2 supplies the per-OS
    /// rungs. When it lands, the daemon fills this field in one line — `resolver:
    /// Some(Arc::new(<I-2's service-context resolver>))` — and nothing in this file
    /// changes. Until then a failed route logs that there is nothing to try, and the loop
    /// keeps backing off, which is exactly what a host with no ladder should do.
    pub resolver: Option<Arc<dyn ProxyResolver>>,
    /// The handle set every consumer reads. **The swap IS the rebuild** — see
    /// [`EgressClients`].
    pub clients: Arc<EgressClients>,
    /// The route resolved at daemon start. Candidates are folded onto it, so the
    /// operator's bypass list, trust bundle and `allow_direct` survive a heal.
    pub base: Arc<EgressConfig>,
    /// The platform base URL. Candidates are validated against `{api_url}/api/v1/health`,
    /// the same target the cloud worker probes.
    pub api_url: String,
    /// `~/.openlatch` — where the `proxy_configured` shape memo lives.
    pub openlatch_dir: std::path::PathBuf,
}

impl std::fmt::Debug for SelfHeal {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SelfHeal")
            .field("resolver_wired", &self.resolver.is_some())
            .field("api_url", &self.api_url)
            .finish_non_exhaustive()
    }
}

impl SelfHeal {
    /// The health endpoint a candidate is proved against.
    fn probe_url(&self) -> String {
        format!("{}/api/v1/health", self.api_url.trim_end_matches('/'))
    }
}

/// Run the monitor until shutdown.
///
/// Returns on an explicit shutdown signal **and** on the sender being dropped — both mean
/// the daemon is going away, and a loop that kept spinning on a closed channel would burn
/// a core through teardown.
///
/// `heal` is `None` for a daemon with nothing to heal onto (and in the observer-only unit
/// tests); `Some` arms the self-heal pass described on [`SelfHeal`].
pub async fn run_egress_monitor(
    state: EgressState,
    mut shutdown: watch::Receiver<bool>,
    heal: Option<SelfHeal>,
) {
    let mut published = state.status();
    log_transition(&state, None, published);

    let mut backoff = heal_backoff();
    // `Some` = a self-heal pass is armed for that instant. HYSTERESIS lives here: it is
    // set only while the status is failed and cleared the moment egress recovers, so a
    // healed route sticks until it fails again. Nothing re-discovers on a timer.
    let mut heal_at: Option<tokio::time::Instant> = None;
    // One line per failed streak, not one per tick, when the source is manual.
    let mut declined_manual = false;

    loop {
        // A placeholder instant for the disarmed case: the `if` guard below keeps the
        // branch from being polled, and the expression only has to build.
        let heal_deadline = heal_at.unwrap_or_else(|| tokio::time::Instant::now() + TICK);
        tokio::select! {
            _ = shutdown.changed() => return,
            _ = state.failure_notify().notified() => {}
            _ = tokio::time::sleep(TICK) => {}
            _ = tokio::time::sleep_until(heal_deadline), if heal_at.is_some() => {
                heal_at = None;
                if let Some(heal) = heal.as_ref() {
                    attempt_self_heal(&state, heal).await;
                }
            }
        }

        // Defensive: every recording call already publishes on a transition, so this
        // normally finds nothing to do. It costs one atomic load and it is what makes the
        // published snapshot self-correcting if a future caller ever mutates the counters
        // without going through `record_*`.
        state.publish_status();

        let current = state.status();
        if current != published {
            log_transition(&state, Some(published), current);
            published = current;
        }

        if current == EgressStatus::Failed {
            if heal_at.is_none() {
                heal_at = arm_self_heal(&state, heal.as_ref(), &mut backoff, &mut declined_manual);
            }
        } else {
            // Recovered, or never failed: disarm and forget the streak, so the next
            // failure starts at the floor instead of inheriting a fifteen-minute wait.
            heal_at = None;
            declined_manual = false;
            backoff.reset();
        }
    }
}

/// Decide when — or whether — the next self-heal pass runs.
///
/// `None` means "not for this streak": no resolver is wired, or the route is manual.
fn arm_self_heal(
    state: &EgressState,
    heal: Option<&SelfHeal>,
    backoff: &mut Backoff,
    declined_manual: &mut bool,
) -> Option<tokio::time::Instant> {
    heal?;

    // D-7 and D-18, the same rule twice: a human set this route, so neither discovery nor
    // self-heal may touch it. The remedy belongs to the operator, and `doctor` names it.
    if state.snapshot().source == Some(ProxySource::Manual) {
        if !*declined_manual {
            *declined_manual = true;
            tracing::info!(
                target: "egress",
                "egress is failing on a manually configured proxy; self-heal will not change it \
                 -- run `openlatch proxy discover` to look for a working route"
            );
        }
        return None;
    }

    Some(tokio::time::Instant::now() + backoff.next_delay())
}

/// One self-heal pass: walk the candidates, prove one, swap it in.
///
/// Returns whether a candidate was applied.
async fn attempt_self_heal(state: &EgressState, heal: &SelfHeal) -> bool {
    let _probing = state.probe_guard();

    let Some(resolver) = heal.resolver.as_ref() else {
        // **I-2 gated:** see [`SelfHeal::resolver`]. There is nothing to try, and nothing
        // is wrong with the host — the ladder simply does not exist in this build yet, so
        // this is deliberately not recorded as a transport failure.
        tracing::debug!(
            target: "egress",
            "self-heal has no resolver wired; leaving the route as it is"
        );
        return false;
    };

    let candidates = resolver.candidates(&heal.api_url);
    let attempted = candidates.len() as u32;
    if candidates.is_empty() {
        tracing::info!(target: "egress", "self-heal found no proxy candidates");
        return no_candidate_worked(state, heal, 0);
    }

    let probe_url = heal.probe_url();
    for candidate in &candidates {
        let cfg = heal.base.with_candidate(candidate);
        let masked = EgressSnapshot::from_config(&cfg).proxy_url_masked;
        let route = masked.as_deref().unwrap_or("direct");

        // Build every consumer's client on the candidate, and install none of them yet:
        // the request below has to happen through the candidate before anything moves.
        let staged = match heal.clients.stage(&cfg) {
            Ok(staged) => staged,
            Err(e) => {
                tracing::info!(
                    target: "egress",
                    source = candidate.source.as_str(),
                    route,
                    code = %e.code,
                    detail = %e.message,
                    "self-heal candidate rejected"
                );
                continue;
            }
        };

        // THE real request — not a reachability heuristic, and not the old client. The
        // candidate's own cloud client against the endpoint the worker probes. Any
        // response at all proves the route; even a 401 says the platform is there.
        match staged.cloud().get(&probe_url).send().await {
            Ok(response) => {
                // Applied, then published, then recorded — in that order. `record_ok` is
                // the only thing that turns the state `ok`, and it fires strictly after a
                // request has already succeeded through the new route (D-18).
                heal.clients.install(staged);
                state.publish_snapshot(EgressSnapshot::from_config(&cfg));
                state.record_ok();

                tracing::info!(
                    target: "egress",
                    source = candidate.source.as_str(),
                    route,
                    status = response.status().as_u16(),
                    attempts = attempted,
                    "self-heal switched egress onto a working route"
                );

                // **I-2 gated:** the winning route is not written to `config.toml` yet.
                // `persist_proxy_config` — url + source, or source only for a PAC/WPAD
                // win, with `source = "manual"` never overwritten — is I-2's, and lands in
                // `src/core/config/mod.rs` beside `persist_api_url`. Its call belongs
                // HERE, after the install above, so the ordering the acceptance criterion
                // checks (a real request through the new route precedes the config write)
                // holds by construction rather than by review.

                super::shape_memo::emit_if_changed(
                    &heal.openlatch_dir,
                    &state.snapshot(),
                    attempted,
                );
                return true;
            }
            Err(e) => {
                tracing::info!(
                    target: "egress",
                    source = candidate.source.as_str(),
                    route,
                    detail = %mask_text(&e.to_string()),
                    "self-heal candidate did not reach the platform"
                );
            }
        }
    }

    no_candidate_worked(state, heal, attempted)
}

/// Record the outcome of a pass in which nothing worked. Always returns `false`.
///
/// The important half is D-21: with `allow_direct = false` the ladder must **not** end at
/// DIRECT. There is no terminal direct rung here to skip — the loop only ever installs a
/// candidate's *proxied* clients — so the whole job is to say why, with the code the
/// operator can act on. The dead clients stay installed, and that is the point: they fail,
/// and failing is precisely what the policy asks for.
fn no_candidate_worked(state: &EgressState, heal: &SelfHeal, attempted: u32) -> bool {
    if heal.base.allow_direct {
        state.record_failure(
            ERR_PROXY_UNREACHABLE,
            format!("no proxy candidate worked after {attempted} attempt(s)"),
        );
    } else {
        state.record_failure(
            ERR_DIRECT_FORBIDDEN,
            format!(
                "no proxy candidate worked after {attempted} attempt(s), and \
                 [proxy] allow_direct = false forbids going direct"
            ),
        );
    }
    false
}

/// Emit the one line an operator greps for when egress moves.
///
/// Carries the masked route rather than the raw one for the same reason the endpoint does:
/// `daemon.log` is collected, shipped and attached to tickets.
fn log_transition(state: &EgressState, from: Option<EgressStatus>, to: EgressStatus) {
    let snapshot = state.snapshot();
    let route = snapshot.proxy_url_masked.as_deref().unwrap_or("direct");
    let error = state.last_error();
    let code = error.as_ref().map(|e| e.code).unwrap_or("");
    match to {
        EgressStatus::Failed => tracing::warn!(
            target: "egress",
            code,
            route,
            consecutive_failures = state.consecutive_failures(),
            detail = error.as_ref().map(|e| e.message.as_str()).unwrap_or(""),
            "egress failed"
        ),
        EgressStatus::Degraded => tracing::info!(
            target: "egress",
            route,
            warnings = state.warnings().join("; "),
            "egress degraded"
        ),
        EgressStatus::Ok if from.is_some() => tracing::info!(
            target: "egress",
            route,
            "egress recovered"
        ),
        // The boot line. `debug!` because a healthy direct host has nothing to say and
        // every daemon start would otherwise carry a noise line.
        EgressStatus::Ok | EgressStatus::Unknown => tracing::debug!(
            target: "egress",
            route,
            proxy_in_use = snapshot.proxy_in_use,
            "egress monitor started"
        ),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::egress::config::{EnvSource, ProxyToml};
    use crate::core::egress::factory::Timeouts;
    use crate::core::egress::ProxyCandidate;
    use std::sync::atomic::{AtomicU32, Ordering};

    struct NoEnv;
    impl EnvSource for NoEnv {
        fn var(&self, _key: &str) -> Option<String> {
            None
        }
    }

    fn resolve(toml: ProxyToml) -> EgressConfig {
        EgressConfig::resolve(Some(&toml), &NoEnv, 7443, 7444).expect("resolve")
    }

    /// Drive the monitor until `check` holds, or give up. Beats a fixed sleep: the loop is
    /// woken by a notify, so the settle time is scheduler-dependent.
    async fn eventually(mut check: impl FnMut() -> bool) -> bool {
        for _ in 0..400 {
            if check() {
                return true;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        false
    }

    // -----------------------------------------------------------------------
    // The I-2 seam: a fake resolver
    // -----------------------------------------------------------------------

    /// A [`ProxyResolver`] that answers a fixed list and counts how often it was asked.
    ///
    /// This is what makes the self-heal loop testable **without I-2**: the loop talks to
    /// the trait and never to a discovery implementation, so backoff, hysteresis,
    /// ordering, the manual gate and the `allow_direct` posture are all provable here, and
    /// the real ladder drops in behind the same method.
    struct FakeResolver {
        candidates: Vec<ProxyCandidate>,
        asked: Arc<AtomicU32>,
    }

    impl FakeResolver {
        /// Returns the resolver behind the trait object the loop actually holds, plus the
        /// call counter — which is why this is not a `new` returning `Self`.
        fn wired(candidates: Vec<ProxyCandidate>) -> (Arc<dyn ProxyResolver>, Arc<AtomicU32>) {
            let asked = Arc::new(AtomicU32::new(0));
            (
                Arc::new(Self {
                    candidates,
                    asked: asked.clone(),
                }),
                asked,
            )
        }
    }

    impl ProxyResolver for FakeResolver {
        fn candidates(&self, _target: &str) -> Vec<ProxyCandidate> {
            self.asked.fetch_add(1, Ordering::Relaxed);
            self.candidates.clone()
        }
    }

    /// A minimal HTTP forward proxy: reads the absolute-URI request line reqwest sends for
    /// a proxied `http://` target, records that it was used, and answers `200`.
    ///
    /// Real enough for the only claim the tests make — that the probe travelled through
    /// *this* route — and twenty-five lines instead of a fixture dependency.
    async fn spawn_mock_forward_proxy() -> (u16, Arc<AtomicU32>) {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
            .await
            .expect("bind");
        let port = listener.local_addr().expect("addr").port();
        let hits = Arc::new(AtomicU32::new(0));
        let counter = hits.clone();
        tokio::spawn(async move {
            loop {
                let Ok((mut stream, _)) = listener.accept().await else {
                    return;
                };
                let counter = counter.clone();
                tokio::spawn(async move {
                    let mut scratch = [0u8; 2048];
                    if stream.read(&mut scratch).await.is_ok() {
                        counter.fetch_add(1, Ordering::Relaxed);
                        let _ = stream
                            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
                            .await;
                        let _ = stream.flush().await;
                    }
                });
            }
        });
        (port, hits)
    }

    /// A loopback port that refuses connections: **probed, never bound**.
    ///
    /// Binding port 0 and reading the number back is the obvious way to write this and it is
    /// racy — the kernel returns that number to the very next `bind(0)`, and this module's own
    /// fixtures call it. So draw from a range below every platform's ephemeral floor (32768 on
    /// Linux, 49152 on macOS and Windows), which no `bind(0)` can ever hand out, and return a
    /// port only once a connect to it has been refused. `tests/support/mod.rs` carries the same
    /// helper for the integration suites; an integration test cannot see a `#[cfg(test)]` item,
    /// which is the only reason there are two.
    fn dead_port() -> u16 {
        use std::net::{SocketAddr, TcpStream};
        use std::sync::atomic::{AtomicU16, Ordering as AtomicOrdering};

        static NEXT: AtomicU16 = AtomicU16::new(20000);

        loop {
            let port = NEXT.fetch_add(1, AtomicOrdering::Relaxed);
            assert!(port < 21000, "exhausted 20000..21000 dead ports");
            let addr = SocketAddr::from(([127, 0, 0, 1], port));
            if TcpStream::connect_timeout(&addr, Duration::from_millis(50)).is_err() {
                return port;
            }
        }
    }

    fn heal_for(
        resolver: Option<Arc<dyn ProxyResolver>>,
        base: EgressConfig,
        dir: &std::path::Path,
    ) -> (SelfHeal, Arc<EgressClients>) {
        let clients = Arc::new(EgressClients::new(Timeouts::default(), Timeouts::default()));
        (
            SelfHeal {
                resolver,
                clients: clients.clone(),
                base: Arc::new(base),
                // Never resolved locally: every candidate below is an http proxy, so
                // reqwest sends the absolute URI to the proxy and never looks this up.
                api_url: "http://platform.test.invalid".into(),
                openlatch_dir: dir.to_path_buf(),
            },
            clients,
        )
    }

    // -----------------------------------------------------------------------
    // Observer half
    // -----------------------------------------------------------------------

    #[tokio::test]
    async fn the_monitor_survives_a_failure_streak_and_publishes_it() {
        let state = EgressState::new(&EgressConfig::direct());
        let (_tx, rx) = watch::channel(false);
        let task = tokio::spawn(run_egress_monitor(state.clone(), rx, None));

        state.record_failure(ERR_PROXY_UNREACHABLE, "connection refused");
        state.record_failure(ERR_PROXY_UNREACHABLE, "connection refused");

        assert!(
            eventually(|| state.snapshot().status == EgressStatus::Failed).await,
            "two consecutive failures must publish failed"
        );
        task.abort();
    }

    #[tokio::test]
    async fn the_monitor_publishes_recovery() {
        let state = EgressState::new(&EgressConfig::direct());
        let (_tx, rx) = watch::channel(false);
        let task = tokio::spawn(run_egress_monitor(state.clone(), rx, None));

        state.record_failure(ERR_PROXY_UNREACHABLE, "boom");
        state.record_failure(ERR_PROXY_UNREACHABLE, "boom");
        assert!(eventually(|| state.snapshot().status == EgressStatus::Failed).await);

        state.record_ok();
        assert!(
            eventually(|| state.snapshot().status == EgressStatus::Ok).await,
            "one success must clear the streak"
        );
        task.abort();
    }

    #[tokio::test]
    async fn the_monitor_returns_on_shutdown() {
        let state = EgressState::new(&EgressConfig::direct());
        let (tx, rx) = watch::channel(false);
        let task = tokio::spawn(run_egress_monitor(state, rx, None));
        tx.send(true).expect("send shutdown");
        tokio::time::timeout(Duration::from_secs(2), task)
            .await
            .expect("monitor must return on shutdown")
            .expect("monitor task");
    }

    #[tokio::test]
    async fn the_monitor_returns_when_the_sender_is_dropped() {
        // A closed channel resolves `changed()` immediately and forever; without the
        // explicit return that would spin the select at full speed.
        let state = EgressState::new(&EgressConfig::direct());
        let (tx, rx) = watch::channel(false);
        let task = tokio::spawn(run_egress_monitor(state, rx, None));
        drop(tx);
        tokio::time::timeout(Duration::from_secs(2), task)
            .await
            .expect("monitor must return when the daemon drops the signal")
            .expect("monitor task");
    }

    #[tokio::test]
    async fn a_monitor_with_no_credentials_still_publishes_state() {
        // The credential-less host: no cloud worker exists, so nothing probes, but the
        // state is live and the surfaces have something to serve from the moment the
        // daemon starts.
        let state = EgressState::new(&EgressConfig::direct());
        let (_tx, rx) = watch::channel(false);
        let task = tokio::spawn(run_egress_monitor(state.clone(), rx, None));
        assert!(eventually(|| state.snapshot().status == EgressStatus::Ok).await);
        assert!(state.is_idle(), "no outcome has ever been recorded");
        task.abort();
    }

    // -----------------------------------------------------------------------
    // Backoff (G-06, and the flagged deviation)
    // -----------------------------------------------------------------------

    #[test]
    fn the_shipped_heal_backoff_is_a_minute_to_fifteen() {
        // Asserted on `base()` rather than on a sampled delay: `next_delay` jitters ±20%,
        // which would make a delay assertion flaky for no gain. The house sentence caps
        // task restarts at 60 s; this deliberately ceilings at D-18's 900 s, because
        // re-running WPAD discovery in a tight loop off-network is a rogue-PAC exposure.
        assert_eq!(HEAL_BACKOFF_INITIAL, Duration::from_secs(60));
        assert_eq!(HEAL_BACKOFF_MAX, Duration::from_secs(900));

        let mut backoff = Backoff::new(HEAL_BACKOFF_INITIAL, HEAL_BACKOFF_MAX);
        assert_eq!(
            backoff.base(),
            Duration::from_secs(60),
            "the first pass waits a minute"
        );
        for _ in 0..20 {
            backoff.next_delay();
        }
        assert_eq!(
            backoff.base(),
            HEAL_BACKOFF_MAX,
            "the curve must settle at the ceiling and never past it"
        );
        backoff.reset();
        assert_eq!(
            backoff.base(),
            HEAL_BACKOFF_INITIAL,
            "a recovered route starts the next streak at the floor"
        );
    }

    // -----------------------------------------------------------------------
    // Self-heal
    // -----------------------------------------------------------------------

    #[tokio::test]
    async fn without_a_resolver_self_heal_does_nothing_and_says_so() {
        // The state of this branch: the `ProxyResolver` seam has no implementation, so a
        // failing route stays as it is and no client is touched. I-2 fills the seam.
        let dir = tempfile::tempdir().expect("tempdir");
        let state = EgressState::new(&EgressConfig::direct());
        let (heal, clients) = heal_for(None, EgressConfig::direct(), dir.path());
        clients.apply(&EgressConfig::direct()).expect("seed");

        assert!(!attempt_self_heal(&state, &heal).await);
        assert_eq!(state.status(), EgressStatus::Ok);
        assert!(
            state.last_error().is_none(),
            "a missing resolver is not a transport failure and must not be recorded as one"
        );
        assert!(
            clients.cloud.current().is_some(),
            "the installed client must survive a pass with nothing to try"
        );
    }

    #[tokio::test]
    async fn a_working_candidate_is_probed_first_then_installed() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (proxy_port, proxy_hits) = spawn_mock_forward_proxy().await;
        let (resolver, asked) = FakeResolver::wired(vec![
            // Nothing is listening here: the first rung must fail and the pass must move
            // on rather than give up.
            ProxyCandidate {
                url: format!("http://127.0.0.1:{}", dead_port()),
                source: ProxySource::Env,
            },
            ProxyCandidate {
                url: format!("http://127.0.0.1:{proxy_port}"),
                source: ProxySource::Gnome,
            },
        ]);

        let state = EgressState::new(&EgressConfig::direct());
        let (heal, clients) = heal_for(Some(resolver), EgressConfig::direct(), dir.path());
        clients.apply(&EgressConfig::direct()).expect("seed");

        assert!(
            attempt_self_heal(&state, &heal).await,
            "the second candidate works"
        );
        assert_eq!(asked.load(Ordering::Relaxed), 1, "one ladder walk per pass");
        assert!(
            proxy_hits.load(Ordering::Relaxed) >= 1,
            "the probe must have travelled through the candidate's own route"
        );
        assert_eq!(
            state.status(),
            EgressStatus::Ok,
            "record_ok fires only from the real request"
        );

        let snapshot = state.snapshot();
        assert_eq!(snapshot.source, Some(ProxySource::Gnome));
        assert!(snapshot.proxy_in_use);
        assert!(
            snapshot
                .proxy_url_masked
                .as_deref()
                .is_some_and(|u| u.contains(&proxy_port.to_string())),
            "the published route must be the one that was proved"
        );
        assert!(
            clients.boundary.current().is_some(),
            "every consumer handle is swapped, not just the one that probed"
        );
    }

    #[tokio::test]
    async fn a_candidate_that_fails_its_probe_is_never_installed() {
        // The apply-before-persist ordering, asserted from the other side: nothing moves
        // until a real request has already succeeded.
        let dir = tempfile::tempdir().expect("tempdir");
        let (resolver, _) = FakeResolver::wired(vec![ProxyCandidate {
            url: format!("http://127.0.0.1:{}", dead_port()),
            source: ProxySource::Env,
        }]);

        let state = EgressState::new(&EgressConfig::direct());
        let (heal, clients) = heal_for(Some(resolver), EgressConfig::direct(), dir.path());
        clients.apply(&EgressConfig::direct()).expect("seed");

        assert!(!attempt_self_heal(&state, &heal).await);
        let snapshot = state.snapshot();
        assert!(
            !snapshot.proxy_in_use && snapshot.source.is_none(),
            "a failed candidate must not be published as the route"
        );
        assert_eq!(
            state.last_error().map(|e| e.code),
            Some(ERR_PROXY_UNREACHABLE)
        );
    }

    #[tokio::test]
    async fn allow_direct_false_ends_the_ladder_at_ol_1227_not_at_direct() {
        // D-21. The pass only ever installs a *proxied* candidate, so there is no direct
        // rung to skip — what has to be true is that the failure is named, with the code
        // the operator can act on, instead of the host quietly connecting anyway.
        let dir = tempfile::tempdir().expect("tempdir");
        let base = resolve(ProxyToml {
            mode: Some("manual".into()),
            url: Some("http://proxy.corp:8080".into()),
            allow_direct: Some(false),
            ..Default::default()
        });
        let (resolver, _) = FakeResolver::wired(vec![ProxyCandidate {
            url: format!("http://127.0.0.1:{}", dead_port()),
            source: ProxySource::Env,
        }]);

        let state = EgressState::new(&base);
        let (heal, _clients) = heal_for(Some(resolver), base, dir.path());

        assert!(!attempt_self_heal(&state, &heal).await);
        assert_eq!(
            state.last_error().map(|e| e.code),
            Some(ERR_DIRECT_FORBIDDEN),
            "allow_direct = false must surface OL-1227 when nothing works"
        );
    }

    #[tokio::test]
    async fn no_candidates_at_all_still_records_a_failure() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (resolver, _) = FakeResolver::wired(vec![]);
        let state = EgressState::new(&EgressConfig::direct());
        let (heal, _clients) = heal_for(Some(resolver), EgressConfig::direct(), dir.path());

        assert!(!attempt_self_heal(&state, &heal).await);
        assert_eq!(
            state.last_error().map(|e| e.code),
            Some(ERR_PROXY_UNREACHABLE)
        );
    }

    #[tokio::test]
    async fn a_manual_source_is_never_healed() {
        // D-7/D-18: a human set this route. The loop declines to even ask the resolver.
        let manual = resolve(ProxyToml {
            mode: Some("manual".into()),
            url: Some("http://proxy.corp:8080".into()),
            source: Some("manual".into()),
            ..Default::default()
        });
        let (resolver, asked) = FakeResolver::wired(vec![ProxyCandidate {
            url: "http://127.0.0.1:1".into(),
            source: ProxySource::Env,
        }]);
        let dir = tempfile::tempdir().expect("tempdir");
        let state = EgressState::new(&manual);
        let (heal, _clients) = heal_for(Some(resolver), manual, dir.path());

        let (_tx, rx) = watch::channel(false);
        let task = tokio::spawn(run_egress_monitor(state.clone(), rx, Some(heal)));
        state.record_failure(ERR_PROXY_UNREACHABLE, "dead");
        state.record_failure(ERR_PROXY_UNREACHABLE, "dead");
        assert!(eventually(|| state.snapshot().status == EgressStatus::Failed).await);

        // Long enough for several compressed backoff windows to have elapsed.
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert_eq!(
            asked.load(Ordering::Relaxed),
            0,
            "a manual route must never reach the resolver"
        );
        task.abort();
    }

    #[tokio::test]
    async fn the_loop_heals_a_failed_route_and_then_stops_discovering() {
        // Hysteresis: discovery runs while failed and stops the moment egress recovers.
        // Nothing re-discovers on a timer.
        let dir = tempfile::tempdir().expect("tempdir");
        let (proxy_port, _) = spawn_mock_forward_proxy().await;
        let (resolver, asked) = FakeResolver::wired(vec![ProxyCandidate {
            url: format!("http://127.0.0.1:{proxy_port}"),
            source: ProxySource::Windows,
        }]);

        let state = EgressState::new(&EgressConfig::direct());
        let (heal, clients) = heal_for(Some(resolver), EgressConfig::direct(), dir.path());
        clients.apply(&EgressConfig::direct()).expect("seed");

        let (_tx, rx) = watch::channel(false);
        let task = tokio::spawn(run_egress_monitor(state.clone(), rx, Some(heal)));

        state.record_failure(ERR_PROXY_UNREACHABLE, "dead");
        state.record_failure(ERR_PROXY_UNREACHABLE, "dead");

        assert!(
            eventually(|| state.snapshot().source == Some(ProxySource::Windows)).await,
            "the loop must heal onto the working candidate"
        );
        assert!(eventually(|| state.snapshot().status == EgressStatus::Ok).await);

        let after_heal = asked.load(Ordering::Relaxed);
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert_eq!(
            asked.load(Ordering::Relaxed),
            after_heal,
            "a healed route sticks: no further discovery until it fails again"
        );
        task.abort();
    }

    #[tokio::test]
    async fn a_failing_streak_keeps_retrying_under_backoff() {
        // "Never gives up; the ceiling holds" — with nothing working, the pass runs again
        // and again rather than latching off after the first miss.
        let dir = tempfile::tempdir().expect("tempdir");
        let (resolver, asked) = FakeResolver::wired(vec![ProxyCandidate {
            url: format!("http://127.0.0.1:{}", dead_port()),
            source: ProxySource::Env,
        }]);
        let state = EgressState::new(&EgressConfig::direct());
        let (heal, _clients) = heal_for(Some(resolver), EgressConfig::direct(), dir.path());

        let (_tx, rx) = watch::channel(false);
        let task = tokio::spawn(run_egress_monitor(state.clone(), rx, Some(heal)));
        state.record_failure(ERR_PROXY_UNREACHABLE, "dead");
        state.record_failure(ERR_PROXY_UNREACHABLE, "dead");

        assert!(
            eventually(|| asked.load(Ordering::Relaxed) >= 2).await,
            "the pass must repeat while the route is failed"
        );
        task.abort();
    }
}