fix(lab): project a terminal reverse result after the worker exits (#10659) (#10688)

* fix(lab): project a terminal reverse result after the worker exits

A detached Cook's reverse worker exits the moment it publishes its
terminal result to the broker, so its controller-session heartbeat
necessarily ages out past REVERSE_RUNNER_HEARTBEAT_TTL. daemon_api_request
gated every reverse read on that liveness, so runner_job_log_snapshot
returned "runner is not connected to a daemon", agent-task reconciliation
annotated the run runner_disconnected, and the durable parent could never
reach a terminal state -- the exact assertion
detached_cook_accepts_reverse_capacity_queue_and_worker_completes_once
makes.

The broker is a controller-reachable durable rendezvous, not the runner
machine. Serve reverse broker reads from a recorded session that names its
broker endpoint. Mutations still assert an intent a live worker has to
honor, and every direct-daemon transport reads the runner machine itself,
so both keep the liveness gate.

The acceptance fixture hid this by pinning last_seen_at five minutes in
the future, which reverse_controller_session_is_live accepts because a
negative age fails Duration::try_from and falls open. That gave the test a
fixed ~390s liveness window while the test itself runs 430-590s, so the
outcome flipped on where that wall-clock cliff landed. Beat the session
honestly while the fixture worker is connected, then record the expired
session a departed worker actually leaves behind before asserting terminal
projection.

Refs #10659

* test(lab): keep reverse status reads on the fixture SSH shim

An expired reverse session makes `runner status` fall through to its SSH
recovery probe. The acceptance fixture already ships an `ssh` shim for
exactly that reason, but the `agent-task status` reads inherited the
ambient PATH, so a real `ssh` to reverse-fixture.invalid could escape the
hermetic context and stall the read.

Refs #10659

---------

Co-authored-by: chubes-bot <266378653+homeboy-ci[bot]@users.noreply.github.com>

Author: chubes4

crates/homeboy-lab-runner/src/execution/daemon_api.rs

@@ -130,10 +130,36 @@ pub fn daemon_api_post(runner_id: &str, path: &str) -> Result<Value> {
     daemon_api_request(runner_id, path, "POST")
 }
 
+/// A reverse runner's broker is a controller-reachable durable rendezvous, not
+/// the runner machine. A detached Cook's worker exits as soon as it publishes
+/// its terminal result, so its controller-session heartbeat necessarily ages
+/// out past `REVERSE_RUNNER_HEARTBEAT_TTL` — but the result it already wrote to
+/// the broker stays readable and must stay projectable
+/// (Extra-Chill/homeboy#10659). Observation of durable broker state therefore
+/// requires only a recorded reverse session that names its broker endpoint.
+/// Mutations still assert an intent that a live worker has to honor, so they
+/// keep the liveness gate, as does every direct-daemon transport.
+fn reverse_broker_read_without_live_worker(session: Option<&RunnerSession>, method: &str) -> bool {
+    method == "GET"
+        && session.is_some_and(|session| {
+            session.mode == RunnerTunnelMode::Reverse
+                && session.local_url.is_none()
+                && session
+                    .broker_url
+                    .as_deref()
+                    .is_some_and(|url| !url.trim().is_empty())
+        })
+}
+
 pub(super) fn daemon_api_request(runner_id: &str, path: &str, method: &str) -> Result<Value> {
     let runner = load(runner_id)?;
     let connected = status(runner_id)?;
-    let Some(legacy_session) = connected.session.filter(|_| connected.connected) else {
+    let recorded_broker_read =
+        reverse_broker_read_without_live_worker(connected.session.as_ref(), method);
+    let Some(legacy_session) = connected
+        .session
+        .filter(|_| connected.connected || recorded_broker_read)
+    else {
         return Err(Error::validation_invalid_argument(
             "runner",
             "runner is not connected to a daemon; run `homeboy runner connect <runner-id>` first",
@@ -143,7 +169,14 @@ pub(super) fn daemon_api_request(runner_id: &str, path: &str, method: &str) -> R
             ]),
         ));
     };
-    let session = daemon_api_session_for_path(runner_id, path, legacy_session)?;
+    // Generation routing resolves which *live* direct-daemon endpoint owns a
+    // job. A recorded-only reverse session has no generation to reconcile, so
+    // read it through the broker endpoint it recorded.
+    let session = if connected.connected {
+        daemon_api_session_for_path(runner_id, path, legacy_session)?
+    } else {
+        legacy_session
+    };
     let client = Client::builder()
         .no_proxy()
         .timeout(Duration::from_secs(10))
@@ -289,6 +322,67 @@ mod tests {
         let body = canonical_daemon_body(&data, "reverse broker job").expect("canonical body");
         assert_eq!(body["job"]["id"], "job-1");
     }
+
+    fn reverse_session() -> RunnerSession {
+        let mut session = session("lease-reverse", "unused");
+        session.mode = RunnerTunnelMode::Reverse;
+        session.broker_url = Some("http://127.0.0.1:9000".to_string());
+        session.local_url = None;
+        session.local_port = None;
+        session.remote_daemon_address = None;
+        session
+    }
+
+    #[test]
+    fn recorded_reverse_session_still_serves_broker_reads_after_its_worker_exits() {
+        // A detached Cook's worker exits the moment it publishes its terminal
+        // result, so its session heartbeat is always stale by the time the
+        // controller projects. The broker is durable and controller-reachable,
+        // so the read must not depend on worker liveness
+        // (Extra-Chill/homeboy#10659).
+        assert!(reverse_broker_read_without_live_worker(
+            Some(&reverse_session()),
+            "GET"
+        ));
+    }
+
+    #[test]
+    fn recorded_reverse_session_still_gates_mutations_and_direct_transports() {
+        // A mutation asserts an intent a live worker has to honor.
+        assert!(!reverse_broker_read_without_live_worker(
+            Some(&reverse_session()),
+            "POST"
+        ));
+        // Direct-SSH transport reads the runner machine itself, so liveness
+        // remains the correct gate.
+        assert!(!reverse_broker_read_without_live_worker(
+            Some(&session("lease-a", "daemon-a")),
+            "GET"
+        ));
+        // A reverse session that never recorded a broker endpoint has nothing
+        // durable to read.
+        let mut without_broker = reverse_session();
+        without_broker.broker_url = None;
+        assert!(!reverse_broker_read_without_live_worker(
+            Some(&without_broker),
+            "GET"
+        ));
+        let mut blank_broker = reverse_session();
+        blank_broker.broker_url = Some("   ".to_string());
+        assert!(!reverse_broker_read_without_live_worker(
+            Some(&blank_broker),
+            "GET"
+        ));
+        // A reverse session pinned to a local daemon endpoint would route
+        // direct, and that endpoint really is dead once the worker exits.
+        let mut with_local = reverse_session();
+        with_local.local_url = Some("http://127.0.0.1:4100".to_string());
+        assert!(!reverse_broker_read_without_live_worker(
+            Some(&with_local),
+            "GET"
+        ));
+        assert!(!reverse_broker_read_without_live_worker(None, "GET"));
+    }
 }
 
 pub(super) fn daemon_post(client: &Client, local_url: &str, path: &str) -> Result<Value> {

tests/reverse_cook_queue_acceptance.rs

@@ -1,9 +1,90 @@
+use std::path::{Path, PathBuf};
 use std::process::{Command, Stdio};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use homeboy::core::api_jobs::{JobEventKind, JobStatus};
 use homeboy_core::test_support::{HermeticTestContext, ReverseBrokerFixture, TestBinary};
 
+/// A live reverse worker republishes its controller session heartbeat while it
+/// is connected; the recorded `last_seen_at` is only as old as its last beat.
+/// The fixture used to fake that with a single timestamp five minutes in the
+/// future, which `reverse_controller_session_is_live` accepts because a
+/// negative age fails `Duration::try_from` and falls open. That gave the test a
+/// fixed ~390 s liveness window (300 s of future skew plus the 90 s heartbeat
+/// TTL) while the test itself runs 430–590 s, so whether it passed depended on
+/// where that wall-clock cliff landed. Beat the session honestly instead.
+struct ReverseSessionHeartbeat {
+    path: PathBuf,
+    session: serde_json::Value,
+    stop: Arc<AtomicBool>,
+    handle: Option<std::thread::JoinHandle<()>>,
+}
+
+impl ReverseSessionHeartbeat {
+    fn start(path: &Path, mut session: serde_json::Value) -> Self {
+        let stop = Arc::new(AtomicBool::new(false));
+        session["last_seen_at"] = serde_json::json!(chrono::Utc::now().to_rfc3339());
+        Self::write(path, &session);
+        let handle = {
+            let path = path.to_path_buf();
+            let session = session.clone();
+            let stop = stop.clone();
+            std::thread::spawn(move || {
+                while !stop.load(Ordering::SeqCst) {
+                    std::thread::sleep(Duration::from_millis(500));
+                    if stop.load(Ordering::SeqCst) {
+                        break;
+                    }
+                    let mut beat = session.clone();
+                    beat["last_seen_at"] = serde_json::json!(chrono::Utc::now().to_rfc3339());
+                    Self::write(&path, &beat);
+                }
+            })
+        };
+        Self {
+            path: path.to_path_buf(),
+            session,
+            stop,
+            handle: Some(handle),
+        }
+    }
+
+    /// Stop beating and record the session a worker that has already exited
+    /// leaves behind: a real `last_seen_at` older than the reverse heartbeat
+    /// TTL. The controller must still project the terminal result the worker
+    /// published to the broker before it exited.
+    fn expire(&mut self) {
+        self.stop.store(true, Ordering::SeqCst);
+        if let Some(handle) = self.handle.take() {
+            handle.join().expect("reverse session heartbeat thread");
+        }
+        let mut expired = self.session.clone();
+        expired["last_seen_at"] =
+            serde_json::json!((chrono::Utc::now() - chrono::Duration::minutes(5)).to_rfc3339());
+        Self::write(&self.path, &expired);
+    }
+
+    /// Publish through a same-directory rename. A truncating rewrite would let
+    /// a controller read observe an empty session file mid-beat and report the
+    /// runner disconnected for reasons that have nothing to do with liveness.
+    fn write(path: &Path, session: &serde_json::Value) {
+        let staged = path.with_extension("beat");
+        std::fs::write(&staged, session.to_string()).expect("stage reverse controller session");
+        std::fs::rename(&staged, path).expect("publish reverse controller session");
+    }
+}
+
+impl Drop for ReverseSessionHeartbeat {
+    fn drop(&mut self) {
+        self.stop.store(true, Ordering::SeqCst);
+        if let Some(handle) = self.handle.take() {
+            let _ = handle.join();
+        }
+    }
+}
+
 fn output(command: &mut Command) -> std::process::Output {
     let output = command.output().expect("run homeboy fixture command");
     assert!(
@@ -163,8 +244,10 @@ fn detached_cook_accepts_reverse_capacity_queue_and_worker_completes_once() {
         .join("runner-sessions/lab/fixture-controller.json");
     std::fs::create_dir_all(session_path.parent().expect("session parent"))
         .expect("create session directory");
-    std::fs::write(
-        session_path,
+    // Beat the session for as long as the fixture worker is "connected", the
+    // way `homeboy runner work` does, instead of pinning one future timestamp.
+    let mut session_heartbeat = ReverseSessionHeartbeat::start(
+        &session_path,
         serde_json::json!({
             "runner_id": "lab",
             "mode": "reverse",
@@ -179,15 +262,12 @@ fn detached_cook_accepts_reverse_capacity_queue_and_worker_completes_once() {
             "connected_at": "2026-01-01T00:00:00Z",
             "worker_identity": "fixture-worker",
             "worker_pid": 1,
-            "last_seen_at": (chrono::Utc::now() + chrono::Duration::minutes(5)).to_rfc3339()
-        })
-        .to_string(),
-    )
-    .expect("write reverse controller session");
+        }),
+    );
 
     let mut cook_command = context.command(TestBinary::HomeboyFixture);
     cook_command
-        .env("PATH", path)
+        .env("PATH", &path)
         .env("HOMEBOY_CONTROLLER_ID", "fixture-controller")
         .args([
             "--runner",
@@ -249,6 +329,7 @@ fn detached_cook_accepts_reverse_capacity_queue_and_worker_completes_once() {
             let run_id = accepted["latest_run_id"].as_str().unwrap_or("unknown");
             let status = context
                 .command(TestBinary::HomeboyFixture)
+                .env("PATH", &path)
                 .args(["agent-task", "status", run_id])
                 .output()
                 .expect("inspect stalled controller parent");
@@ -366,6 +447,12 @@ fn detached_cook_accepts_reverse_capacity_queue_and_worker_completes_once() {
         })
         .expect("duplicate worker wake");
     assert_eq!(duplicate_code, 0);
+    // Both worker waves have returned, so the reverse worker is gone and its
+    // controller-session heartbeat stops. That is the steady state of a
+    // detached Cook, not an edge case: the worker exits the moment it publishes
+    // its terminal result. Record it explicitly so terminal projection is
+    // proven against a genuinely expired session instead of racing one.
+    session_heartbeat.expire();
     // The controller must project the broker result after the worker exits.
     // `daemon serve` is intentionally un-tokenized, so terminate the test-owned
     // foreground child only after that durable parent lifecycle is terminal.
@@ -374,6 +461,10 @@ fn detached_cook_accepts_reverse_capacity_queue_and_worker_completes_once() {
     let terminal = loop {
         let status = context
             .command(TestBinary::HomeboyFixture)
+            // A recorded-only reverse session makes `runner status` reach for
+            // its SSH recovery probe. Keep that on the fixture shim rather than
+            // letting a real `ssh` escape the hermetic context.
+            .env("PATH", &path)
             .args(["agent-task", "status", run_id])
             .output()
             .expect("read terminal parent status");