fix: report live Lab cook progress accurately

Author: chubes4

src/commands/runner/status.rs

@@ -918,6 +918,15 @@ pub(super) fn runner_status_operator_hints(report: &RunnerStatusReport) -> Vec<S
             "Active-job status for `{}` is unavailable: {reason}. Treat active_job_count=0 as unknown, not idle.",
             report.runner_id
         ));
+    } else if let Some(error) = report
+        .active_job_error
+        .as_ref()
+        .filter(|error| error.code == "active_job_view_inconsistent")
+    {
+        hints.push(format!(
+            "Active-job view for `{}` is inconsistent: {}",
+            report.runner_id, error.message
+        ));
     }
     if report.stale_runner_job_count > 0 {
         hints.push(format!(

src/core/agent_task_lifecycle/lifecycle_ops.rs

@@ -500,6 +500,22 @@ pub(crate) fn reconcile_runner_job_snapshot(
             let metadata = reconciled.ensure_metadata_object();
             metadata.insert("runner_job_status".to_string(), json!(snapshot.job.status));
             metadata.insert("runner_job_last_seen_at".to_string(), json!(last_seen_at));
+            metadata.insert("runner_job_events".to_string(), json!(snapshot.events));
+            metadata.insert("phase".to_string(), json!("executing"));
+            metadata.insert(
+                "phase_activity".to_string(),
+                json!("provider/executor process is active"),
+            );
+            metadata.insert("provider_state".to_string(), json!("active"));
+            if let Some(provider) = metadata
+                .get("provider_rotation")
+                .and_then(|rotation| rotation.get("entries"))
+                .and_then(Value::as_array)
+                .and_then(|entries| entries.first())
+            {
+                metadata.insert("active_provider".to_string(), provider.clone());
+            }
+            merge_live_provider_handles(&mut reconciled, &snapshot.events);
             store::write_record(&reconciled)?;
         }
         crate::core::api_jobs::JobStatus::Succeeded
@@ -530,6 +546,52 @@ pub(crate) fn reconcile_runner_job_snapshot(
     Ok(())
 }
 
+fn merge_live_provider_handles(
+    record: &mut AgentTaskRunRecord,
+    events: &[crate::core::api_jobs::JobEvent],
+) {
+    for handle in events.iter().filter_map(|event| {
+        event
+            .data
+            .as_ref()
+            .and_then(|data| {
+                data.pointer("/metadata/provider_handle")
+                    .or_else(|| data.get("provider_handle"))
+            })
+            .and_then(provider_handle_from_value)
+    }) {
+        if record
+            .provider_handles
+            .iter()
+            .any(|existing| existing.provider_run_id == handle.run_id)
+        {
+            continue;
+        }
+        record.provider_handles.push(AgentTaskRunProviderHandle {
+            kind: handle.kind,
+            task_id: handle.task_id,
+            backend: handle.backend,
+            provider_run_id: handle.run_id,
+            stream_uri: handle.stream_uri,
+            state: Some(AgentTaskState::Running),
+            metadata: handle.metadata,
+        });
+    }
+    if !record.provider_handles.is_empty() {
+        record.lifecycle.provider_runtime = record
+            .provider_handles
+            .iter()
+            .map(provider_runtime_for_handle)
+            .collect();
+        record.lifecycle.external_runtime_ids = record
+            .lifecycle
+            .provider_runtime
+            .iter()
+            .flat_map(|runtime| runtime.external_runtime_ids.clone())
+            .collect();
+    }
+}
+
 fn validate_runner_job_snapshot(
     record: &AgentTaskRunRecord,
     snapshot: &crate::core::runner::RunnerJobLogSnapshot,
@@ -845,8 +907,9 @@ pub fn record_lab_offload_phase(
         durable_plan,
     )?;
     record.updated_at = Some(now_timestamp());
+    let phase_started_at = record.updated_at.clone().unwrap_or_else(now_timestamp);
     let metadata = record.ensure_metadata_object();
-    metadata.insert("phase".to_string(), json!(phase));
+    record_lab_offload_phase_metadata(metadata, phase, &phase_started_at);
     metadata.insert("provider_state".to_string(), json!("pending"));
     if let Some(remote_workspace) = remote_workspace {
         metadata.insert("remote_workspace".to_string(), json!(remote_workspace));
@@ -875,8 +938,9 @@ pub fn record_lab_offload_phase_executions(
         .filter(|id| !id.trim().is_empty())
         .collect();
     record.updated_at = Some(now_timestamp());
+    let phase_started_at = record.updated_at.clone().unwrap_or_else(now_timestamp);
     let metadata = record.ensure_metadata_object();
-    metadata.insert("phase".to_string(), json!(phase));
+    record_lab_offload_phase_metadata(metadata, phase, &phase_started_at);
     metadata.insert(
         "materialization_execution_ids".to_string(),
         json!(execution_ids),
@@ -889,6 +953,44 @@ pub fn record_lab_offload_phase_executions(
     Ok(record)
 }
 
+fn record_lab_offload_phase_metadata(
+    metadata: &mut serde_json::Map<String, Value>,
+    phase: &str,
+    started_at: &str,
+) {
+    let previous_phase = metadata
+        .get("phase")
+        .and_then(Value::as_str)
+        .map(str::to_string);
+    if previous_phase.as_deref() != Some(phase) {
+        if let Some(previous_phase) = previous_phase {
+            if let Some(entry) = metadata
+                .get_mut("phase_history")
+                .and_then(Value::as_array_mut)
+                .and_then(|entries| {
+                    entries.iter_mut().rev().find(|entry| {
+                        entry.get("phase").and_then(Value::as_str) == Some(previous_phase.as_str())
+                            && entry.get("ended_at").is_none()
+                    })
+                })
+            {
+                entry["ended_at"] = json!(started_at);
+            }
+        }
+        metadata
+            .entry("phase_history".to_string())
+            .or_insert_with(|| json!([]))
+            .as_array_mut()
+            .expect("phase history is an array")
+            .push(json!({ "phase": phase, "started_at": started_at }));
+    }
+    metadata.insert("phase".to_string(), json!(phase));
+    metadata.insert(
+        "phase_activity".to_string(),
+        json!(format!("Homeboy {phase}")),
+    );
+}
+
 pub fn record_detached_lab_run(input: DetachedLabRunRecord<'_>) -> Result<AgentTaskRunRecord> {
     let run_id = sanitize_run_id(input.run_id);
     let plan = detached_lab_plan(&run_id, &input);
@@ -1209,14 +1311,19 @@ pub fn retry(run_id: &str, requested_run_id: Option<&str>) -> Result<AgentTaskRu
 }
 
 pub fn logs(run_id: &str) -> Result<AgentTaskRunLog> {
-    let run_id = resolve_run_id(run_id)?;
-    let record = store::read_record(&run_id)?;
+    // Status reconciliation fetches the live daemon snapshot for a bound Lab
+    // child, making executor progress visible before the child is terminal.
+    let record = status(run_id)?;
+    let run_id = record.run_id.clone();
     let (events, artifact_refs) = match store::read_aggregate(&run_id) {
         Ok(aggregate) => {
             let refs = artifact_refs_for_outcomes(&aggregate.outcomes);
             (aggregate.events, refs)
         }
-        Err(_) => (queued_events(&record.tasks), record.artifact_refs.clone()),
+        Err(_) => (
+            runner_job_progress_events(&record).unwrap_or_else(|| queued_events(&record.tasks)),
+            record.artifact_refs.clone(),
+        ),
     };
     let normalized_events = normalize_progress_events(&run_id, &events, &artifact_refs);
     Ok(AgentTaskRunLog {
@@ -1227,6 +1334,26 @@ pub fn logs(run_id: &str) -> Result<AgentTaskRunLog> {
     })
 }
 
+fn runner_job_progress_events(record: &AgentTaskRunRecord) -> Option<Vec<AgentTaskProgressEvent>> {
+    let events = record.metadata.get("runner_job_events")?.as_array()?;
+    let task_id = record
+        .tasks
+        .first()
+        .map(|task| task.task_id.clone())
+        .unwrap_or_else(|| record.run_id.clone());
+    Some(
+        events
+            .iter()
+            .map(|event| AgentTaskProgressEvent {
+                task_id: task_id.clone(),
+                state: AgentTaskState::Running,
+                attempt: 0,
+                message: serde_json::to_string(event).ok(),
+            })
+            .collect(),
+    )
+}
+
 pub fn artifacts(run_id: &str) -> Result<AgentTaskRunArtifacts> {
     let run_id = resolve_run_id(run_id)?;
     let record = store::read_record(&run_id)?;

src/core/agent_task_lifecycle/tests.rs

@@ -12,6 +12,7 @@ use crate::core::agent_task_scheduler::{
     AgentTaskAggregate, AgentTaskAggregateStatus, AgentTaskAggregateTotals,
     AGENT_TASK_AGGREGATE_SCHEMA,
 };
+use crate::core::api_jobs::{JobEvent, JobEventKind};
 use crate::test_support::with_isolated_home;
 
 #[test]
@@ -444,6 +445,47 @@ fn controller_proxy_records_pre_execution_phase_progress() {
         let loaded = status("agent-task-pre-execution").expect("status resolves during setup");
         assert_eq!(loaded.metadata["phase"], "hydrating");
         assert_eq!(loaded.metadata["provider_state"], "pending");
+        let phases = loaded.metadata["phase_history"]
+            .as_array()
+            .expect("phase history");
+        assert_eq!(phases.len(), 2);
+        assert_eq!(phases[0]["phase"], "materializing");
+        assert!(phases[0].get("started_at").is_some());
+        assert!(phases[0].get("ended_at").is_some());
+        assert_eq!(phases[1]["phase"], "hydrating");
+        assert!(phases[1].get("started_at").is_some());
+    });
+}
+
+#[test]
+fn logs_expose_mirrored_live_runner_events_before_terminal_aggregate() {
+    with_isolated_home(|_| {
+        let command = vec!["homeboy".to_string(), "agent-task".to_string()];
+        let mut record = record_detached_lab_run(DetachedLabRunRecord {
+            run_id: "live-runner-events",
+            runner_id: "homeboy-lab",
+            runner_job_id: "job-live",
+            remote_workspace: "/runner/workspace/homeboy",
+            remote_command: &command,
+        })
+        .expect("running proxy");
+        record.metadata["runner_job_events"] = json!([JobEvent {
+            sequence: 1,
+            job_id: uuid::Uuid::new_v4(),
+            kind: JobEventKind::Progress,
+            timestamp_ms: 42,
+            message: Some("provider started".to_string()),
+            data: Some(json!({"provider": "openai/gpt-5.6-terra"})),
+        }]);
+        store::write_record(&record).expect("persist mirrored event");
+
+        let log = logs("live-runner-events").expect("live logs resolve");
+
+        assert_eq!(log.events.len(), 1);
+        assert!(log.events[0]
+            .message
+            .as_deref()
+            .is_some_and(|message| message.contains("provider started")));
     });
 }
 
@@ -645,6 +687,58 @@ fn reachable_running_child_clears_disconnected_liveness_and_refreshes_heartbeat(
     });
 }
 
+#[test]
+fn running_child_snapshot_persists_provider_handle_and_live_log_progress() {
+    with_isolated_home(|_| {
+        let command = vec!["homeboy".to_string(), "agent-task".to_string()];
+        let mut record = record_detached_lab_run(DetachedLabRunRecord {
+            run_id: "agent-task-live-provider",
+            runner_id: "homeboy-lab",
+            runner_job_id: "00000000-0000-0000-0000-000000000123",
+            remote_workspace: "/runner/workspace/repo",
+            remote_command: &command,
+        })
+        .expect("running proxy");
+        let mut snapshot = terminal_child_snapshot(&succeeded_aggregate(&test_plan()));
+        snapshot.job.status = crate::core::api_jobs::JobStatus::Running;
+        snapshot.events = vec![crate::core::api_jobs::JobEvent {
+            sequence: 1,
+            job_id: snapshot.job.id,
+            kind: crate::core::api_jobs::JobEventKind::Progress,
+            timestamp_ms: 2,
+            message: Some("provider dispatch accepted".to_string()),
+            data: Some(json!({
+                "metadata": {
+                    "provider_handle": AgentTaskExecutionHandle {
+                        kind: crate::core::agent_task::AgentTaskExecutionHandleKind::ProviderRun,
+                        task_id: "task-a".to_string(),
+                        backend: "openai/gpt-5.6-terra".to_string(),
+                        run_id: "provider-run-live".to_string(),
+                        stream_uri: Some("provider://runs/provider-run-live/events".to_string()),
+                        metadata: json!({"progress": "accepted"}),
+                    }
+                }
+            })),
+        }];
+
+        reconcile_runner_job_snapshot(&mut record, &snapshot).expect("live reconciliation");
+
+        assert_eq!(record.metadata["phase"], "executing");
+        assert_eq!(record.metadata["provider_state"], "active");
+        assert_eq!(record.provider_handles.len(), 1);
+        assert_eq!(
+            record.provider_handles[0].provider_run_id,
+            "provider-run-live"
+        );
+        let log = logs(&record.run_id).expect("live logs");
+        assert_eq!(log.events.len(), 1);
+        assert!(log.events[0]
+            .message
+            .as_deref()
+            .is_some_and(|message| message.contains("provider dispatch accepted")));
+    });
+}
+
 #[test]
 fn terminal_child_projection_rejects_mismatched_persisted_run_identity() {
     with_isolated_home(|_| {