fix(lab): poll reverse staging through broker (#10589)

Author: chubes4

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

@@ -2363,11 +2363,28 @@ impl ProductionLabStagingOperations {
                 events: observed.events,
             });
         }
+        let broker_url = Self::status_for_recipe_transport(request)?
+            .session
+            .and_then(|session| session.broker_url)
+            .ok_or_else(|| {
+                Error::validation_invalid_argument(
+                    "runner",
+                    "durable reverse Lab staging requires its accepted broker endpoint",
+                    Some(request.recipe.runner_id.clone()),
+                    None,
+                )
+            })?;
         loop {
             if cancellation.is_some_and(LabStagingCancellationToken::is_cancelled) {
                 return Err(Self::cancellation_error());
             }
-            match crate::runner_job_log_snapshot(&request.recipe.runner_id, runner_job_id) {
+            match crate::connection::reverse_broker_job_snapshot_at(
+                &broker_url,
+                &request.recipe.runner_id,
+                runner_job_id,
+            )
+            .map(|(job, events)| homeboy_core::api_jobs::RunnerJobLogSnapshot { job, events })
+            {
                 Ok(snapshot) if snapshot.job.status.is_terminal() => return Ok(snapshot),
                 Ok(_) => std::thread::sleep(Duration::from_millis(200)),
                 Err(mut error) => {

tests/reverse_cook_queue_acceptance.rs

@@ -329,6 +329,26 @@ fn detached_cook_accepts_reverse_capacity_queue_and_worker_completes_once() {
         terminal_result.get("exit_code").is_some(),
         "broker terminal result preserves the typed payload: {terminal_result}"
     );
+    let broker_events: serde_json::Value =
+        reqwest::blocking::get(format!("{}/jobs/{}/events", broker.url(), completed.id))
+            .expect("fetch broker events over HTTP")
+            .json()
+            .expect("parse broker events response");
+    let broker_terminal_result = broker_events
+        .pointer("/data/body/events")
+        .and_then(serde_json::Value::as_array)
+        .and_then(|events| {
+            events.iter().rev().find_map(|event| {
+                (event["kind"] == serde_json::json!("result")).then(|| event["data"].clone())
+            })
+        })
+        .expect("broker HTTP response retains terminal result");
+    serde_json::from_value::<homeboy::core::api_jobs::RemoteRunnerJobResult>(
+        broker_terminal_result.clone(),
+    )
+    .unwrap_or_else(|error| {
+        panic!("broker HTTP terminal result must retain its typed contract: {error}\nresult={broker_terminal_result}")
+    });
 
     let (_, duplicate_code) =
         homeboy::runner::run_reverse_worker(homeboy::runner::ReverseRunnerWorkerOptions {