fix(runner): bound recovery churn (#11957)

AI assistance: OpenAI GPT-5.6 Sol via OpenCode implemented and tested the bounded recovery convergence. Chris Huber remains responsible for every line.

Author: Chris Huber

crates/homeboy-cli/src/cli_runtime.rs

@@ -775,13 +775,13 @@ fn schedule_runner_exec_recovery() {
     let Ok(Some(schedule)) = crate::runner::schedule_terminal_runner_exec_recovery() else {
         return;
     };
-    eprintln!(
-        "runner-exec recovery accepted: owner_id={} deferred_count={} budget_ms={} inspect=`{}`",
-        schedule.owner_id, schedule.deferred_count, schedule.budget_ms, schedule.inspection_action
-    );
     if !schedule.is_new_owner {
         return;
     }
+    eprintln!(
+        "runner-exec recovery scheduled: owner_id={} deferred_count={} inspect=`{}`",
+        schedule.owner_id, schedule.deferred_count, schedule.inspection_action
+    );
     let executable = match std::env::current_exe() {
         Ok(executable) => executable,
         Err(error) => {

crates/homeboy-cli/src/command_capability.rs

@@ -149,29 +149,15 @@ pub fn classify(args: &[String]) -> CommandCapability {
     }
 }
 
-/// Whether startup may reconcile runner-exec state before this invocation.
+/// Whether this command owns runner-exec recovery.
 ///
-/// This is intentionally narrower than command mutation: a review may
-/// materialize a controller-local artifact, but runner recovery would make its
-/// already-durable aggregate unavailable during a Lab outage.
+/// Recovery may contact a previously selected runner, so only `runner exec`
+/// may start it. Other mutations must not turn routine commands into an
+/// unrelated background-recovery admission path.
 pub fn requires_startup_reconciliation(args: &[String]) -> bool {
-    classify(args) == CommandCapability::Mutation
-        // The lifecycle command that launches these internal processes already
-        // performs startup recovery. Repeating it here delays lease publication
-        // and can make the supervisor kill an otherwise healthy cold start.
-        && !args
-            .get(1..)
-            .unwrap_or_default()
-            .windows(2)
-            .any(|args| {
-                matches!(args, [command, subcommand]
-                    if command == "daemon" && matches!(subcommand.as_str(), "supervise" | "serve"))
-            })
-        && !args
-            .get(1..)
-            .unwrap_or_default()
-            .windows(2)
-            .any(|args| args == ["agent-task", "review"])
+    homeboy_owned_args(args.get(1..).unwrap_or_default())
+        .windows(2)
+        .any(|args| args == ["runner", "exec"])
 }
 
 #[cfg(test)]
@@ -255,7 +241,7 @@ mod tests {
     }
 
     #[test]
-    fn review_keeps_its_local_materialization_capability_but_skips_runner_recovery() {
+    fn only_runner_exec_owns_startup_recovery() {
         let review = args(&[
             "homeboy",
             "agent-task",
@@ -266,17 +252,20 @@ mod tests {
         ]);
         assert_eq!(classify(&review), CommandCapability::Mutation);
         assert!(!requires_startup_reconciliation(&review));
-        assert!(requires_startup_reconciliation(&args(&[
+        assert!(!requires_startup_reconciliation(&args(&[
             "homeboy",
             "agent-task",
             "retry",
             "run-1",
             "--run",
         ])));
+        assert!(requires_startup_reconciliation(&args(&[
+            "homeboy", "runner", "exec", "lab", "--", "true",
+        ])));
     }
 
     #[test]
-    fn internal_daemon_processes_publish_their_lease_without_repeating_recovery() {
+    fn other_mutations_do_not_own_runner_recovery() {
         for command in [
             args(&[
                 "homeboy",
@@ -305,7 +294,7 @@ mod tests {
             args(&["homeboy", "daemon", "start"]),
             args(&["homeboy", "daemon", "ensure-running"]),
         ] {
-            assert!(requires_startup_reconciliation(&command));
+            assert!(!requires_startup_reconciliation(&command));
         }
     }
 

crates/homeboy-core/src/observation/store/runs.rs

@@ -322,6 +322,25 @@ impl ObservationStore {
         Ok(rows == 1)
     }
 
+    /// Persist a retryable recovery blocker only while this child still owns the
+    /// source lease. The replacement metadata deliberately releases the lease
+    /// so a later, bounded retry can make progress.
+    pub fn defer_running_runner_exec_recovery_source(
+        &self,
+        run_id: &str,
+        child_token: &str,
+        metadata_json: serde_json::Value,
+    ) -> Result<bool> {
+        let metadata_json = serialize_metadata(&metadata_json)?;
+        let rows = execute_with_retry("defer claimed runner recovery source", || {
+            self.connection.execute(
+                "UPDATE runs SET metadata_json = ?1 WHERE id = ?2 AND status = 'running' AND json_extract(metadata_json, '$.runner_exec_source_lease.source_token') = ?3",
+                params![metadata_json, run_id, child_token],
+            )
+        })?;
+        Ok(rows == 1)
+    }
+
     pub fn start_run_with_context(
         &self,
         run: NewRunRecord,

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

@@ -19,6 +19,7 @@ const RECOVERY_OWNER_ID: &str = "runner-exec-recovery";
 const RECOVERY_CHILD_KIND: &str = "runner_exec_recovery_child";
 const RECOVERY_OWNER_LEASE: Duration = Duration::from_secs(30);
 const RECOVERY_LEASE_HEARTBEAT: Duration = Duration::from_secs(1);
+const MAX_SOURCE_RECOVERY_DEFERRALS: u64 = 3;
 #[derive(Clone)]
 struct RecoveryWorker {
     id: String,
@@ -223,7 +224,9 @@ fn reconcile_terminal_runner_exec_runs_with_owner(
                 if error.details.get("http_status").and_then(Value::as_u64) != Some(404) {
                     unavailable_endpoints.insert(endpoint);
                 }
-                deferred += 1;
+                if defer_or_fail_recovery_source(&store, run, &worker.token, job_id, &error)? {
+                    deferred += 1;
+                }
                 continue;
             }
         };
@@ -337,6 +340,54 @@ fn reconcile_terminal_runner_exec_runs_with_owner(
     Ok((reconciled, deferred))
 }
 
+/// A runner that remains unavailable cannot be retried by every unrelated
+/// command forever. Keep a small, inspectable retry budget, then terminalize
+/// the source with the exact blocker retained in its durable metadata.
+fn defer_or_fail_recovery_source(
+    store: &ObservationStore,
+    run: &homeboy_core::observation::RunRecord,
+    child_token: &str,
+    runner_job_id: &str,
+    error: &Error,
+) -> Result<bool> {
+    let attempts = run
+        .metadata_json
+        .pointer("/runner_exec_recovery/deferral_count")
+        .and_then(Value::as_u64)
+        .unwrap_or(0)
+        + 1;
+    let mut metadata = run.metadata_json.clone();
+    metadata.as_object_mut().map(|metadata| {
+        metadata.remove("runner_exec_source_lease");
+    });
+    metadata["runner_exec_recovery"] = json!({
+        "schema": "homeboy/runner-exec-recovery/v1",
+        "phase": if attempts >= MAX_SOURCE_RECOVERY_DEFERRALS { "blocked" } else { "deferred" },
+        "deferral_count": attempts,
+        "max_deferrals": MAX_SOURCE_RECOVERY_DEFERRALS,
+        "reason": error.message,
+        "details": error.details,
+        "inspection_action": format!("homeboy runs show {}", run.id),
+    });
+    if attempts >= MAX_SOURCE_RECOVERY_DEFERRALS {
+        metadata["runner_terminal_projection"] = json!({
+            "state": "recovery_blocked",
+            "classification": "runner_unavailable_after_bounded_recovery",
+            "runner_id": run.metadata_json["runner_id"],
+            "runner_job_id": runner_job_id,
+        });
+        store.fail_running_runner_exec_recovery_source(
+            &run.id,
+            child_token,
+            runner_job_id,
+            metadata,
+        )?;
+        return Ok(false);
+    }
+    store.defer_running_runner_exec_recovery_source(&run.id, child_token, metadata)?;
+    Ok(true)
+}
+
 fn endpoint_identity(session: &crate::RunnerSession) -> String {
     session
         .remote_daemon_address
@@ -1012,6 +1063,102 @@ mod tests {
         });
     }
 
+    #[test]
+    fn repeated_deferred_recovery_converges_without_stale_children() {
+        with_isolated_home(|_| {
+            let run_id = "unavailable-source";
+            let runner_job_id = "unavailable-job";
+            homeboy_agents::agent_task_lifecycle::record_runner_exec_job_identity(
+                run_id,
+                "unavailable-runner",
+                runner_job_id,
+                "/workspace",
+                &[],
+            )
+            .expect("source");
+            let store = ObservationStore::open_initialized().expect("store");
+            let error = Error::validation_invalid_argument(
+                "runner",
+                "runner has no persisted daemon session for recovery",
+                Some("unavailable-runner".to_string()),
+                None,
+            );
+
+            let mut child_ids = BTreeSet::new();
+            for attempt in 1..=MAX_SOURCE_RECOVERY_DEFERRALS {
+                let owner = schedule_terminal_runner_exec_recovery()
+                    .expect("schedule")
+                    .expect("owner");
+                let work = run_scheduled_terminal_runner_exec_recovery(
+                    &owner.owner_id,
+                    &owner.owner_token,
+                )
+                .expect("schedule child")
+                .expect("owner work");
+                assert_eq!(work.children.len(), 1);
+                let child = &work.children[0];
+                child_ids.insert(child.child_id.clone());
+                let source = store.get_run(run_id).expect("read").expect("source");
+                assert_eq!(
+                    defer_or_fail_recovery_source(
+                        &store,
+                        &source,
+                        &child.child_token,
+                        runner_job_id,
+                        &error,
+                    )
+                    .expect("record bounded deferral"),
+                    attempt < MAX_SOURCE_RECOVERY_DEFERRALS,
+                );
+                let child_record = store
+                    .get_run(&child.child_id)
+                    .expect("read")
+                    .expect("child");
+                store
+                    .finish_running_run_with_owner_token(
+                        &child.child_id,
+                        &child.child_token,
+                        RunStatus::Pass,
+                        child_record.metadata_json,
+                    )
+                    .expect("finish child");
+                finish_scheduled_terminal_runner_exec_recovery(
+                    &owner.owner_id,
+                    &owner.owner_token,
+                    1,
+                    0,
+                    usize::from(attempt < MAX_SOURCE_RECOVERY_DEFERRALS),
+                )
+                .expect("finish owner");
+            }
+
+            let source = store.get_run(run_id).expect("read").expect("source");
+            assert_eq!(source.status, RunStatus::Fail.as_str());
+            assert_eq!(
+                source.metadata_json["runner_exec_recovery"]["deferral_count"],
+                MAX_SOURCE_RECOVERY_DEFERRALS
+            );
+            assert_eq!(
+                source.metadata_json["runner_terminal_projection"]["classification"],
+                "runner_unavailable_after_bounded_recovery"
+            );
+            let reader = ObservationStore::open_scheduler_reader().expect("reader");
+            assert!(recovery_candidates(&reader).expect("candidates").is_empty());
+            let children = store
+                .list_runs(RunListFilter {
+                    kind: Some(RECOVERY_CHILD_KIND.to_string()),
+                    ..RunListFilter::default()
+                })
+                .expect("list children");
+            assert_eq!(child_ids.len(), 1, "retries reuse one child identity");
+            assert_eq!(children.len(), 1, "retries do not accumulate children");
+            assert_ne!(children[0].status, RunStatus::Running.as_str());
+            assert!(schedule_terminal_runner_exec_recovery()
+                .expect("final schedule")
+                .is_none());
+        });
+    }
+
     #[test]
     fn owner_schedules_one_durable_child_per_source_within_its_budget() {
         with_isolated_home(|_| {

tests/cook_lab_handoff_test.rs

@@ -2,7 +2,7 @@ use std::path::Path;
 use std::process::Output;
 use std::time::{Duration, Instant};
 
-use homeboy_core::observation::{NewRunRecord, ObservationStore, RunListFilter, RunStatus};
+use homeboy_core::observation::{NewRunRecord, ObservationStore, RunListFilter};
 use homeboy_core::test_support::{bounded_output, HermeticTestContext, TestBinary};
 
 /// Run the fixture binary through the shared hermetic harness.
@@ -292,11 +292,11 @@ fn non_tty_local_wait_stays_foreground() {
     );
 }
 
-/// Historical runner recovery is a distinct owner, never an admission gate for
-/// an accepted Cook. This invokes the actual detached Cook command path rather
-/// than a scheduler helper so its output and durable handoff remain observable.
+/// Historical runner recovery belongs to `runner exec`, never an unrelated
+/// Cook admission. This invokes the actual detached Cook command path so the
+/// ownership boundary remains observable.
 #[test]
-fn detached_cook_admission_is_bounded_with_a_hundred_unavailable_recovery_records() {
+fn detached_cook_admission_does_not_schedule_unrelated_recovery_records() {
     let context = HermeticTestContext::new();
     let database = context.data_dir().join("homeboy.sqlite");
     let store = ObservationStore::open_initialized_at(&database).expect("open fixture store");
@@ -354,43 +354,21 @@ fn detached_cook_admission_is_bounded_with_a_hundred_unavailable_recovery_record
     assert_eq!(handoff["detached"], true, "{stdout}");
     assert!(handoff["cook_id"].as_str().is_some_and(|id| !id.is_empty()));
 
-    let observation_deadline = Instant::now() + Duration::from_secs(5);
-    let (owners, children) = loop {
-        let store = ObservationStore::open_initialized_at(&database).expect("reopen fixture store");
-        let owners = store
-            .list_runs(RunListFilter {
-                kind: Some("runner_exec_recovery".to_string()),
-                ..RunListFilter::default()
-            })
-            .expect("list recovery owners");
-        let children = store
-            .list_runs(RunListFilter {
-                kind: Some("runner_exec_recovery_child".to_string()),
-                ..RunListFilter::default()
-            })
-            .expect("list recovery children");
-        if owners.len() == 1
-            && owners[0].status != RunStatus::Running.as_str()
-            && children.len() == 100
-            && children
-                .iter()
-                .all(|child| child.status != RunStatus::Running.as_str())
-        {
-            break (owners, children);
-        }
-        assert!(
-            Instant::now() < observation_deadline,
-            "recovery state did not settle"
-        );
-        std::thread::sleep(Duration::from_millis(25));
-    };
-    assert_eq!(owners.len(), 1, "recovery has its own durable owner");
-    assert_eq!(owners[0].status, RunStatus::Pass.as_str());
-    assert_eq!(owners[0].metadata_json["scheduled_count"], 100);
-    assert_eq!(children.len(), 100, "each source has a durable child");
-    assert!(children
-        .iter()
-        .all(|child| child.status == RunStatus::Pass.as_str()));
+    let store = ObservationStore::open_initialized_at(&database).expect("reopen fixture store");
+    assert!(store
+        .list_runs(RunListFilter {
+            kind: Some("runner_exec_recovery".to_string()),
+            ..RunListFilter::default()
+        })
+        .expect("list recovery owners")
+        .is_empty());
+    assert!(store
+        .list_runs(RunListFilter {
+            kind: Some("runner_exec_recovery_child".to_string()),
+            ..RunListFilter::default()
+        })
+        .expect("list recovery children")
+        .is_empty());
 
     #[cfg(unix)]
     if let Some(pid) = handoff["pid"].as_u64() {