fix: preserve reverse broker handoff transport (#10573)

Author: chubes4

crates/homeboy-core/src/test_support.rs

@@ -1115,6 +1115,18 @@ fn handle_reverse_broker_request(
             .expect("claim broker job");
         return ok(json!({ "claim": claim }));
     }
+    if request.method == "POST" && request.path == "/runner/jobs/reconcile" {
+        let reconciled = store
+            .reconcile_expired_remote_runner_claims_for_runner(
+                chrono::Utc::now().timestamp_millis().max(0) as u64,
+                Some(runner_id),
+            )
+            .expect("reconcile broker jobs");
+        return ok(json!({
+            "reconciled_count": reconciled.len(),
+            "reconciled": reconciled,
+        }));
+    }
     // Reverse-runner file transfer (`RunnerFileChannel::BrokerHttp`) posts
     // `/files/{mkdir,upload,download}` for the same host the fixture runs on, so
     // the fixture performs the real filesystem operation. Without these the

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

@@ -1828,14 +1828,14 @@ pub(crate) fn reverse_broker_job_snapshot_at(
     let job_data = broker_http::get_json(
         &client,
         broker_url,
-        &format!("/runner/jobs/{job_id}"),
+        &format!("/jobs/{job_id}"),
         "fetch reverse runner broker job",
         token.as_deref(),
     )?;
     let events_data = broker_http::get_json(
         &client,
         broker_url,
-        &format!("/runner/jobs/{job_id}/events"),
+        &format!("/jobs/{job_id}/events"),
         "fetch reverse runner broker job events",
         token.as_deref(),
     )?;

crates/homeboy-lab-runner/src/evidence/mirror.rs

@@ -254,11 +254,13 @@ pub fn mirror_reverse_broker_evidence(
         let terminal_result: homeboy_core::api_jobs::RemoteRunnerJobResult =
             serde_json::from_value(result.clone()).map_err(|error| {
                 Error::validation_invalid_argument(
-                "result.observation_run_details",
-                "reverse runner terminal result is not a valid typed observation detail contract",
-                Some(error.to_string()),
-                None,
-            )
+                    "result.observation_run_details",
+                    format!(
+                        "reverse runner terminal result is not a valid typed observation detail contract: {error}"
+                    ),
+                    None,
+                    None,
+                )
             })?;
         let runs;
         let local_artifacts = if job.status == JobStatus::Failed {
@@ -512,6 +514,40 @@ pub fn mirror_daemon_job_progress(
     .map(|run| run.run)
 }
 
+pub fn mirror_reverse_broker_job_progress(
+    runner: &Runner,
+    broker_url: &str,
+    cwd: &str,
+    command: &[String],
+    job: &Job,
+    run_id: Option<&str>,
+) -> Result<RunRecord> {
+    let store = ObservationStore::open_initialized()?;
+    let run = mirror_job_run(
+        &store,
+        runner,
+        cwd,
+        command,
+        job,
+        &[],
+        &json!({}),
+        run_id,
+        None,
+    )?
+    .run;
+    record_reverse_broker_metadata(
+        &store,
+        run,
+        runner,
+        broker_url,
+        job,
+        &[],
+        &json!({}),
+        Vec::new(),
+        None,
+    )
+}
+
 /// Records that the controller can no longer observe an accepted runner job.
 /// The remote job may still exist, but the controller-side lifecycle is terminal
 /// and includes the polling diagnostic instead of leaving a stale running mirror.

crates/homeboy-lab-runner/src/evidence/mod.rs

@@ -22,7 +22,8 @@ pub use homeboy_core::api_jobs::RunnerJobLogSnapshot;
 pub use mirror::runner_job_log_snapshot_for_session;
 pub use mirror::{
     controller_artifact_metadata, mirror_connected_runner_run, mirror_daemon_evidence,
-    mirror_daemon_job_progress, mirror_reverse_broker_evidence, mirrored_runner_job_identity,
-    refresh_mirrored_daemon_evidence, runner_job_log_snapshot, terminalize_mirrored_daemon_job,
+    mirror_daemon_job_progress, mirror_reverse_broker_evidence, mirror_reverse_broker_job_progress,
+    mirrored_runner_job_identity, refresh_mirrored_daemon_evidence, runner_job_log_snapshot,
+    terminalize_mirrored_daemon_job,
 };
 pub(crate) use util::{local_job_run_id, runner_exec_run_label};

crates/homeboy-lab-runner/src/evidence/tests/mod.rs

@@ -1566,8 +1566,8 @@ fn reverse_broker_refresh_uses_persisted_transport_after_store_reopen() {
                 broker.join().expect("broker requests")
             },
             vec![
-                format!("/runner/jobs/{}", job.id),
-                format!("/runner/jobs/{}/events", job.id),
+                format!("/jobs/{}", job.id),
+                format!("/jobs/{}/events", job.id),
                 "/runner/jobs/reconcile".to_string(),
             ]
         );

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

@@ -194,7 +194,16 @@ pub(super) fn exec_via_reverse_broker(
         }
     }
     let persisted_run_id = mirror_evidence
-        .then(|| persist_lab_offload_handoff_run(runner, &cwd, &command, &job, run_id.as_deref()))
+        .then(|| {
+            persist_lab_offload_handoff_run(
+                runner,
+                &cwd,
+                &command,
+                &job,
+                run_id.as_deref(),
+                Some(broker_url),
+            )
+        })
         .flatten();
     validate_generic_exec_mirror_run_id(
         run_id_owns_generic_exec,

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

@@ -221,7 +221,9 @@ pub(super) fn exec_via_daemon(
         }
     }
     let persisted_run_id = mirror_evidence
-        .then(|| persist_lab_offload_handoff_run(runner, &cwd, &command, &job, run_id.as_deref()))
+        .then(|| {
+            persist_lab_offload_handoff_run(runner, &cwd, &command, &job, run_id.as_deref(), None)
+        })
         .flatten();
     validate_generic_exec_mirror_run_id(
         run_id_owns_generic_exec,

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

@@ -10,7 +10,7 @@ use homeboy_core::error::{Error, Result};
 use crate::daemon_http_get::parse_daemon_response_json;
 
 use super::super::broker_http;
-use super::super::evidence::mirror_daemon_job_progress;
+use super::super::evidence::{mirror_daemon_job_progress, mirror_reverse_broker_job_progress};
 use super::super::{load, status, Runner, RunnerTunnelMode};
 
 #[allow(unused_imports)]
@@ -393,8 +393,15 @@ pub(super) fn persist_lab_offload_handoff_run(
     command: &[String],
     job: &Job,
     run_id: Option<&str>,
+    reverse_broker_url: Option<&str>,
 ) -> Option<String> {
-    match mirror_daemon_job_progress(runner, cwd, command, job, &[], run_id) {
+    let mirrored = match reverse_broker_url {
+        Some(broker_url) => {
+            mirror_reverse_broker_job_progress(runner, broker_url, cwd, command, job, run_id)
+        }
+        None => mirror_daemon_job_progress(runner, cwd, command, job, &[], run_id),
+    };
+    match mirrored {
         Ok(run) => Some(run.id),
         Err(err) => {
             eprintln!(

crates/homeboy-lab-runner/src/execution/tests/handoff.rs

@@ -381,6 +381,7 @@ fn lab_offload_handoff_persists_run_when_job_is_accepted() {
             &["homeboy".to_string(), "trace".to_string()],
             &job,
             None,
+            None,
         )
         .expect("persist handoff run");
 
@@ -399,6 +400,38 @@ fn lab_offload_handoff_persists_run_when_job_is_accepted() {
     });
 }
 
+#[test]
+fn reverse_handoff_persists_broker_transport_for_post_disconnect_refresh() {
+    homeboy_core::test_support::with_isolated_home(|_| {
+        let runner = ssh_runner();
+        let job = running_job();
+        let command = vec!["homeboy".to_string(), "trace".to_string()];
+        let run_id = persist_lab_offload_handoff_run(
+            &runner,
+            "/srv/homeboy/project",
+            &command,
+            &job,
+            None,
+            Some("http://127.0.0.1:4321"),
+        )
+        .expect("persist reverse handoff run");
+
+        let store = homeboy_core::observation::ObservationStore::open_initialized().expect("store");
+        let run = store
+            .get_run(&run_id)
+            .expect("get run")
+            .expect("persisted reverse handoff run");
+        assert_eq!(
+            run.metadata_json["lab"]["reverse_broker"]["broker_url"],
+            "http://127.0.0.1:4321"
+        );
+        assert_eq!(
+            run.metadata_json["lab"]["reverse_broker"]["job_id"],
+            job.id.to_string()
+        );
+    });
+}
+
 #[test]
 fn accepted_job_that_disappears_persists_a_terminal_controller_failure() {
     homeboy_core::test_support::with_isolated_home(|_| {
@@ -410,9 +443,15 @@ fn accepted_job_that_disappears_persists_a_terminal_controller_failure() {
             "runtime".to_string(),
             "refresh".to_string(),
         ];
-        let run_id =
-            persist_lab_offload_handoff_run(&runner, "/srv/homeboy/project", &command, &job, None)
-                .expect("accepted handoff mirror");
+        let run_id = persist_lab_offload_handoff_run(
+            &runner,
+            "/srv/homeboy/project",
+            &command,
+            &job,
+            None,
+            None,
+        )
+        .expect("accepted handoff mirror");
 
         let err = terminal_runner_poll_failure(
             &runner,
@@ -480,9 +519,15 @@ fn transient_daemon_transport_drop_keeps_the_durable_job_recoverable() {
             "runtime".to_string(),
             "refresh".to_string(),
         ];
-        let run_id =
-            persist_lab_offload_handoff_run(&runner, "/srv/homeboy/project", &command, &job, None)
-                .expect("accepted handoff mirror");
+        let run_id = persist_lab_offload_handoff_run(
+            &runner,
+            "/srv/homeboy/project",
+            &command,
+            &job,
+            None,
+            None,
+        )
+        .expect("accepted handoff mirror");
 
         // A transport-layer drop: the daemon endpoint became unreachable while
         // polling. This is the shape `runner_daemon_health_failure` classifies

tests/reverse_cook_queue_acceptance.rs

@@ -317,6 +317,18 @@ fn detached_cook_accepts_reverse_capacity_queue_and_worker_completes_once() {
             .count(),
         1,
     );
+    let terminal_result = broker
+        .store
+        .events(completed.id)
+        .expect("broker events")
+        .into_iter()
+        .find(|event| event.kind == JobEventKind::Result)
+        .and_then(|event| event.data)
+        .expect("broker terminal result event");
+    assert!(
+        terminal_result.get("exit_code").is_some(),
+        "broker terminal result preserves the typed payload: {terminal_result}"
+    );
 
     let (_, duplicate_code) =
         homeboy::runner::run_reverse_worker(homeboy::runner::ReverseRunnerWorkerOptions {