Fix detached reverse broker terminal projection

crates/homeboy-agents/src/agent_task_service/cook_job.rs

@@ -347,6 +347,29 @@ impl CookJobDriver {
                 }
             }
 
+            // Runner ownership starts when the durable attempt records its job,
+            // not when the launcher child exits. A detached Cook can retain a
+            // provably-live launcher while its reverse worker has already
+            // published a terminal broker result. Keep process identity as the
+            // child-exit safety guard below, but let the daemon project the
+            // runner authority throughout supervision.
+            if let Some(run_id) = job.run_id.as_deref() {
+                let lifecycle_store =
+                    agent_task_lifecycle::AgentTaskLifecycleStore::from_current_environment()?;
+                let mut record = lifecycle_store.read_record(run_id)?;
+                if record.runner_id().is_some() && record.runner_job_id().is_some() {
+                    agent_task_lifecycle::reconcile_runner_job_state_in_store(
+                        &lifecycle_store,
+                        &mut record,
+                    )?;
+                    handle.checkpoint(job.to_checkpoint()?)?;
+                    handle.progress(job.progress_projection())?;
+                    if record.state.is_terminal() {
+                        return job.observe_terminal(Some(record.run_id));
+                    }
+                }
+            }
+
             if !child_is_live(&job.request) {
                 return job.observe_terminal(job.run_id.clone());
             }

crates/homeboy-cli/src/commands/daemon.rs

@@ -791,6 +791,7 @@ fn artifact_get(args: DaemonArtifactGetArgs) -> CmdResult<DaemonOutput> {
 }
 
 fn serve(addr: &str) -> CmdResult<DaemonOutput> {
+    register_daemon_controller_job_providers();
     let parsed = daemon::parse_bind_addr(addr)?;
     let state = daemon::serve_with_analysis_runner(parsed, CommandAnalysisJobRunner)?;
     Ok((
@@ -804,6 +805,14 @@ fn serve(addr: &str) -> CmdResult<DaemonOutput> {
     ))
 }
 
+fn register_daemon_controller_job_providers() {
+    // A daemon executes controller jobs after the submitting CLI has exited.
+    // Register the runner continuation here, at the process boundary that owns
+    // those jobs, rather than relying on a later status command to populate the
+    // process-global projection provider.
+    crate::runner::register_runner_continuation_provider();
+}
+
 #[derive(Debug, Clone, Copy)]
 struct CommandAnalysisJobRunner;
 
@@ -844,6 +853,13 @@ mod tests {
     use super::*;
     use crate::cli_surface::{Cli, Commands};
 
+    #[test]
+    fn daemon_controller_context_registers_runner_continuation_projection() {
+        register_daemon_controller_job_providers();
+
+        assert!(homeboy::agents::agent_task_lifecycle::runner_authority("local").is_configured());
+    }
+
     #[test]
     fn legacy_child_recovery_parser_requires_exact_evidence() {
         assert!(

crates/homeboy-lab-runner/src/connection.rs

@@ -3504,6 +3504,19 @@ pub(crate) fn reverse_broker_job_snapshot_at(
     Ok((job, events))
 }
 
+/// Return the current controller's persisted reverse-broker endpoint without a
+/// liveness/status probe. A controller job may outlive the reverse worker, but
+/// the broker remains the durable authority for its accepted job.
+pub(crate) fn recorded_reverse_broker_url(runner_id: &str) -> Result<Option<String>> {
+    let Some(session) = read_session(runner_id)? else {
+        return Ok(None);
+    };
+    if session.mode != RunnerTunnelMode::Reverse || session.local_url.is_some() {
+        return Ok(None);
+    }
+    Ok(session.broker_url.filter(|url| !url.trim().is_empty()))
+}
+
 /// Reconcile terminal runner jobs through the session's authoritative transport.
 /// The returned body is transport-neutral so callers retain one command contract.
 pub fn reconcile_terminal_jobs(runner_id: &str) -> Result<Value> {

crates/homeboy-lab-runner/src/continuation_provider.rs

@@ -128,6 +128,15 @@ impl RunnerContinuationProvider for RunnerContinuation {
         runner_id: &str,
         job_id: &str,
     ) -> Result<RunnerJobLogSnapshot> {
+        // Controller-job supervision outlives the detached reverse worker. Its
+        // broker result remains authoritative after the worker heartbeat
+        // expires, while the general status path intentionally performs
+        // liveness probes for interactive callers.
+        if let Some(broker_url) = super::connection::recorded_reverse_broker_url(runner_id)? {
+            let (job, events) =
+                super::connection::reverse_broker_job_snapshot_at(&broker_url, runner_id, job_id)?;
+            return Ok(RunnerJobLogSnapshot { job, events });
+        }
         super::evidence::runner_job_log_snapshot(runner_id, job_id)
     }
 

tests/reverse_cook_queue_acceptance.rs

@@ -85,6 +85,19 @@ impl Drop for ReverseSessionHeartbeat {
     }
 }
 
+/// The prepared-source cache is intentionally immutable while a worker uses it.
+/// Restore owner write access before the hermetic fixture removes its checkout.
+struct WritableTreeOnDrop(PathBuf);
+
+impl Drop for WritableTreeOnDrop {
+    fn drop(&mut self) {
+        let _ = Command::new("chmod")
+            .args(["-R", "u+w"])
+            .arg(&self.0)
+            .output();
+    }
+}
+
 /// Wall-clock ledger for the acceptance run.
 ///
 /// This test is the slowest binary in the suite and its deadlines are wall
@@ -202,6 +215,7 @@ fn pinned_runner_route_persists_the_verified_lab_outcome_through_detached_cook_l
     let broker = ReverseBrokerFixture::start("lab");
     let (_checkout_guard, checkout) =
         homeboy_core::test_support::shared_committed_git_repo_fixture("cook-source");
+    let _checkout_permissions = WritableTreeOnDrop(checkout.clone());
     std::fs::write(checkout.join(".gitignore"), "_lab_workspaces/\n")
         .expect("ignore runner workspace materialization");
     homeboy_core::test_support::run_git_fixture_command(&checkout, &["add", ".gitignore"]);
@@ -578,45 +592,29 @@ fn pinned_runner_route_persists_the_verified_lab_outcome_through_detached_cook_l
     // 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.
+    // Read the record directly: `agent-task status` deliberately uses the
+    // caller_opted_out probe policy, so invoking it here could not prove the
+    // daemon's own controller-job provider registration.
     let run_id = accepted["run_id"].as_str().expect("accepted run id");
-    // Bound this on observations as well as wall clock. Each poll is a whole
-    // `homeboy` subprocess, so on a loaded machine a single observation can
-    // outlast a bare wall-clock deadline and the controller is declared stalled
-    // having been asked exactly once. The deadline then measures how long the
-    // probe took, not whether the controller made progress. Requiring a minimum
-    // number of observations keeps the assertion about the controller while
-    // leaving the failure bounded.
+    // Bound this on observations as well as wall clock. Requiring a minimum
+    // number of durable reads keeps the assertion about controller progress,
+    // rather than a single slow observation, while leaving failure bounded.
     const MINIMUM_TERMINAL_OBSERVATIONS: u32 = 8;
     let deadline = Instant::now() + Duration::from_secs(10);
     let mut observations = 0u32;
     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");
         observations += 1;
-        let parsed: serde_json::Value =
-            serde_json::from_slice(&status.stdout).expect("parse terminal parent status");
-        if matches!(
-            parsed
-                .pointer("/data/state")
-                .and_then(serde_json::Value::as_str),
-            Some("succeeded" | "failed" | "cancelled")
-        ) {
-            break parsed;
+        let record = homeboy::agents::agent_task_lifecycle::AgentTaskLifecycleStore::from_current_environment()
+            .expect("open controller lifecycle store")
+            .read_record(run_id)
+            .expect("read controller parent record");
+        if record.state.is_terminal() {
+            break record;
         }
         if observations >= MINIMUM_TERMINAL_OBSERVATIONS && Instant::now() >= deadline {
             panic!(
-                "controller did not project terminal broker result after {observations} observations\n{}\nstatus={}\ndaemon stderr={}",
+                "controller did not project terminal broker result after {observations} observations\n{}\nrecord={record:#?}\ndaemon stderr={}",
                 ledger.render(),
-                parsed,
                 std::fs::read_to_string(&daemon_stderr_path)
                     .unwrap_or_else(|error| format!("<unavailable: {error}>")),
             );
@@ -625,11 +623,9 @@ fn pinned_runner_route_persists_the_verified_lab_outcome_through_detached_cook_l
     };
     ledger.mark("controller_terminal_projection");
     assert_eq!(
-        terminal
-            .pointer("/data/state")
-            .and_then(serde_json::Value::as_str),
-        Some("succeeded"),
-        "controller terminal projection: {terminal}\n{}",
+        terminal.state,
+        homeboy::agents::agent_task_lifecycle::AgentTaskRunState::Succeeded,
+        "controller terminal projection: {terminal:#?}\n{}",
         ledger.render(),
     );
     let durable_record = homeboy::agents::agent_task_lifecycle::status(run_id)