fix(runner): retire rejected state-loss replacements (#11908)

Fixes #11901.

AI assistance: OpenAI gpt-5.6-sol and gpt-5.6-terra via OpenCode implemented, tested, reviewed, and prepared this change for landing. Chris Huber remains responsible for every line.

Author: chubes4

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

@@ -64,6 +64,7 @@ struct CliEnvelope {
     success: bool,
     data: Option<Value>,
     error: Option<Value>,
+    diagnostics: Option<Value>,
 }
 
 struct RemoteDaemonConnectOptions<'a> {
@@ -674,6 +675,11 @@ fn connect_with_orphan_adoption_and_live_lease(
                 REMOTE_LEASELESS_RECOVERY_TIMEOUT,
             );
             if !output.success {
+                if kind == "state-loss" && is_terminal_state_loss_refusal(&output) {
+                    super::generation_store::retire_rejected_state_loss_replacement(
+                        runner_id, &output,
+                    )?;
+                }
                 return Ok(failed_connect(
                     runner_id,
                     session_path,
@@ -691,6 +697,11 @@ fn connect_with_orphan_adoption_and_live_lease(
                 )
             })?;
             if !envelope.success {
+                if kind == "state-loss" && is_terminal_state_loss_refusal(&output) {
+                    super::generation_store::retire_rejected_state_loss_replacement(
+                        runner_id, &output,
+                    )?;
+                }
                 return Ok(failed_connect(
                     runner_id,
                     session_path,
@@ -800,6 +811,11 @@ fn connect_with_orphan_adoption_and_live_lease(
                 REMOTE_LEASELESS_RECOVERY_TIMEOUT,
             );
             if !recovery.success {
+                if is_terminal_state_loss_refusal(&recovery) {
+                    super::generation_store::retire_rejected_state_loss_replacement(
+                        runner_id, &recovery,
+                    )?;
+                }
                 return Ok(failed_connect(
                     runner_id,
                     session_path,
@@ -814,6 +830,11 @@ fn connect_with_orphan_adoption_and_live_lease(
                 )
             })?;
             if !envelope.success {
+                if is_terminal_state_loss_refusal(&recovery) {
+                    super::generation_store::retire_rejected_state_loss_replacement(
+                        runner_id, &recovery,
+                    )?;
+                }
                 return Ok(failed_connect(
                     runner_id,
                     session_path,
@@ -1489,6 +1510,25 @@ fn state_loss_recovery_failure_message(output: &homeboy_core::server::CommandOut
     }
 }
 
+fn is_terminal_state_loss_refusal(output: &homeboy_core::server::CommandOutput) -> bool {
+    if output.timed_out {
+        return false;
+    }
+    let Ok(envelope) = parse_envelope(&output.stdout) else {
+        return false;
+    };
+    if envelope.success {
+        return false;
+    }
+    let code = envelope
+        .diagnostics
+        .as_ref()
+        .and_then(|diagnostics| diagnostics.get("code"))
+        .or_else(|| envelope.error.as_ref().and_then(|error| error.get("code")))
+        .and_then(Value::as_str);
+    matches!(code, Some(code) if code.starts_with("validation.") || code.starts_with("policy."))
+}
+
 fn remote_state_loss_recovery_command(
     homeboy: &str,
     lease_id: &str,

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

@@ -1345,6 +1345,170 @@ esac
     });
 }
 
+#[cfg(unix)]
+fn rejected_state_loss_refusal_is_retired_before_plain_connect_retries(
+    legacy_replay: bool,
+    transport_success: bool,
+) {
+    test_support::with_isolated_home(|home| {
+        let daemon = home.path().join("remote-homeboy");
+        let generation_count = home.path().join("daemon-generations");
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("listener");
+        let address = listener.local_addr().expect("address");
+        let endpoint = std::thread::spawn(move || {
+            for _ in 0..16 {
+                let (mut stream, _) = listener.accept().expect("endpoint request");
+                let mut request = [0; 4096];
+                let length = stream.read(&mut request).expect("read endpoint request");
+                let request = String::from_utf8_lossy(&request[..length]);
+                let body = if request.starts_with("GET /health ") {
+                    r#"{"freshness":{"fresh":true,"restartable":true,"lease_id":"lease-b","pid":4242,"active_jobs":0},"pid":4242}"#
+                } else {
+                    r#"{"version":"0.284.0","build_identity":{"display":"homeboy 0.284.0+test"},"lease":{"lease_id":"lease-b"}}"#
+                };
+                stream
+                    .write_all(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes())
+                    .expect("endpoint response");
+            }
+        });
+        std::fs::write(
+            &daemon,
+            format!(
+                r#"#!/bin/sh
+case "$1 $2" in
+  "self identity")
+    printf '%s\n' '{{"success":true,"data":{{"version":"0.284.0","display":"homeboy 0.284.0+test"}}}}'
+    ;;
+  "daemon status")
+    printf '%s\n' '{{"success":true,"data":{{"running":false,"fresh":false,"reachable":false,"freshness":{{"active_jobs":0}}}}}}'
+    ;;
+  "daemon recover-missing-lease-state")
+    if [ "$3" = "--help" ]; then
+      printf '%s\n' 'OPTIONS:' '    --replacement-operation-id <ID>'
+    else
+      printf '%s\n' '{{"success":false,"diagnostics":{{"code":"validation.invalid_argument"}}}}'
+{rejection_exit}
+    fi
+    ;;
+  "daemon ensure-running")
+    if [ "$3" = "--help" ]; then
+      printf '%s\n' 'OPTIONS:' '    --replacement-operation-id <ID>'
+    else
+      printf '%s' '1' > "{generation_count}"
+      printf '%s\n' '{{"success":true,"data":{{"pid":4242,"address":"{address}","state_path":"/tmp/state-b.json","lease_id":"lease-b"}}}}'
+    fi
+    ;;
+esac
+"#,
+                address = address,
+                generation_count = generation_count.display(),
+                rejection_exit = if transport_success { "" } else { "      exit 2" },
+            ),
+        )
+        .expect("write remote Homeboy shim");
+        let mut permissions = std::fs::metadata(&daemon).expect("metadata").permissions();
+        permissions.set_mode(0o755);
+        std::fs::set_permissions(&daemon, permissions).expect("make shim executable");
+        server::create(
+            &serde_json::json!({ "id": "local-runner", "host": "localhost", "user": "test" })
+                .to_string(),
+            false,
+        )
+        .expect("create local server");
+        crate::create(
+            &serde_json::json!({ "id": "local-runner", "kind": "ssh", "homeboy_path": daemon })
+                .to_string(),
+            false,
+        )
+        .expect("enable local runner");
+
+        if legacy_replay {
+            let operation_id = crate::generation_store::replacement_operation("local-runner")
+                .expect("create legacy operation");
+            crate::generation_store::record_replacement_operation_replay(
+                "local-runner",
+                "state-loss",
+                &remote_state_loss_recovery_command(
+                    daemon.to_str().expect("daemon path"),
+                    "lease-lost",
+                    41,
+                    "127.0.0.1:7419",
+                    &operation_id,
+                ),
+            )
+            .expect("write legacy state-loss replay");
+        }
+
+        let (_rejected, exit_code) = connect_with_orphan_adoption(
+            "local-runner",
+            None,
+            &[],
+            false,
+            (!legacy_replay).then_some("lease-lost"),
+            (!legacy_replay).then_some(41),
+            (!legacy_replay).then_some("127.0.0.1:7419"),
+        )
+        .expect("rejected state-loss result");
+        assert_eq!(exit_code, 20);
+        let journal_dir = homeboy_core::paths::runner_sessions_dir()
+            .expect("runner sessions")
+            .join("local-runner");
+        assert!(!journal_dir.join("pending-replacement.json").exists());
+        let operation: serde_json::Value = serde_json::from_str(
+            &std::fs::read_to_string(journal_dir.join("replacement-operation.json"))
+                .expect("replacement operation"),
+        )
+        .expect("replacement operation JSON");
+        assert!(operation["kind"].is_null());
+        assert!(operation["replay_command"].is_null());
+        let evidence = std::fs::read_dir(journal_dir.join("rejected-replacements"))
+            .expect("rejection evidence")
+            .next()
+            .expect("one rejection evidence")
+            .expect("evidence entry");
+        let evidence: serde_json::Value = serde_json::from_str(
+            &std::fs::read_to_string(evidence.path()).expect("read rejection evidence"),
+        )
+        .expect("rejection evidence JSON");
+        assert_eq!(evidence["kind"], "state-loss");
+        assert_eq!(evidence["exit_code"], if transport_success { 0 } else { 2 });
+
+        let (connected, exit_code) =
+            connect_with_orphan_adoption("local-runner", None, &[], false, None, None, None)
+                .expect("plain connect result");
+        assert_eq!(
+            exit_code, 0,
+            "plain connect: {:?}",
+            connected.failure_message
+        );
+        assert!(connected.connected);
+        assert_eq!(connected.remote_daemon_pid, Some(4242));
+        assert_eq!(
+            std::fs::read_to_string(generation_count).expect("start count"),
+            "1"
+        );
+        drop(endpoint);
+    });
+}
+
+/// #11901: a legacy replay journal can contain an already-refused exact
+/// recovery. Its evidence is retained, but its pending authority must not block
+/// a retry of plain connect.
+#[cfg(unix)]
+#[test]
+fn rejected_legacy_state_loss_replay_is_retired_before_plain_connect_retries() {
+    rejected_state_loss_refusal_is_retired_before_plain_connect_retries(true, false);
+}
+
+/// #11901: the fresh exact state-loss command can receive a typed failure
+/// envelope over a successful transport. That terminal refusal must retire its
+/// replay authority before the next plain connect.
+#[cfg(unix)]
+#[test]
+fn rejected_fresh_state_loss_envelope_is_retired_before_plain_connect_retries() {
+    rejected_state_loss_refusal_is_retired_before_plain_connect_retries(false, true);
+}
+
 /// #10430: once recovery creates B, losing the controller's bounded health
 /// requests must leave enough durable evidence for the next invocation to
 /// authenticate B. It must never create an unjournaled C.

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

@@ -67,6 +67,13 @@ fn replacement_operation_path(runner_id: &str) -> Result<PathBuf> {
         .join("replacement-operation.json"))
 }
 
+fn rejected_replacement_path(runner_id: &str) -> Result<PathBuf> {
+    Ok(paths::runner_sessions_dir()?
+        .join(runner_id)
+        .join("rejected-replacements")
+        .join(format!("{}.json", uuid::Uuid::new_v4())))
+}
+
 fn admission_reservation_path(runner_id: &str) -> Result<PathBuf> {
     Ok(paths::runner_sessions_dir()?
         .join(runner_id)
@@ -94,6 +101,20 @@ struct ReplacementOperation {
     kind: Option<String>,
 }
 
+#[derive(serde::Serialize)]
+struct RejectedReplacementEvidence<'a> {
+    schema: &'static str,
+    runner_id: &'a str,
+    operation_id: &'a str,
+    kind: &'a str,
+    replay_command: Option<&'a str>,
+    rejected_at: String,
+    exit_code: i32,
+    timed_out: bool,
+    stdout: &'a str,
+    stderr: &'a str,
+}
+
 pub(crate) fn write_durable_json<T: serde::Serialize>(
     path: &std::path::Path,
     value: &T,
@@ -794,6 +815,68 @@ pub(crate) fn retire_pending_replacement(runner_id: &str) -> Result<String> {
     })
 }
 
+/// Retire a remote policy or validation refusal after preserving the exact
+/// terminal response. Unlike an interrupted mutation, this operation cannot
+/// become publishable by replaying the same authority.
+pub(crate) fn retire_rejected_state_loss_replacement(
+    runner_id: &str,
+    output: &homeboy_core::server::CommandOutput,
+) -> Result<()> {
+    with_registry_lock(runner_id, || {
+        let operation_path = replacement_operation_path(runner_id)?;
+        let operation: ReplacementOperation =
+            serde_json::from_slice(&std::fs::read(&operation_path).map_err(|error| {
+                Error::internal_io(
+                    error.to_string(),
+                    Some(format!("read {}", operation_path.display())),
+                )
+            })?)
+            .map_err(|error| {
+                Error::config_invalid_json(operation_path.display().to_string(), error)
+            })?;
+        if operation.runner_id != runner_id || operation.kind.as_deref() != Some("state-loss") {
+            return Err(Error::internal_unexpected(
+                "refusing to retire a replacement operation that is not the rejected state-loss recovery",
+            ));
+        }
+        let evidence_path = rejected_replacement_path(runner_id)?;
+        write_durable_json(
+            &evidence_path,
+            &RejectedReplacementEvidence {
+                schema: "homeboy/runner-rejected-replacement/v1",
+                runner_id,
+                operation_id: &operation.operation_id,
+                kind: "state-loss",
+                replay_command: operation.replay_command.as_deref(),
+                rejected_at: Utc::now().to_rfc3339(),
+                exit_code: output.exit_code,
+                timed_out: output.timed_out,
+                stdout: &output.stdout,
+                stderr: &output.stderr,
+            },
+        )?;
+        write_durable_json(
+            &operation_path,
+            &ReplacementOperation {
+                runner_id: runner_id.to_string(),
+                operation_id: uuid::Uuid::new_v4().to_string(),
+                replay_command: None,
+                kind: None,
+            },
+        )?;
+        let pending_path = pending_replacement_path(runner_id)?;
+        if pending_path.exists() {
+            std::fs::remove_file(&pending_path).map_err(|error| {
+                Error::internal_io(
+                    error.to_string(),
+                    Some(format!("remove {}", pending_path.display())),
+                )
+            })?;
+        }
+        Ok(())
+    })
+}
+
 /// Promote the direct session that status has just health-checked. The
 /// controller session is the admission authority for new work; older entries
 /// remain draining so their owned jobs keep their original endpoints.