fix(agent-task): isolate fanout liveness attribution (#11939)

AI assistance: OpenAI gpt-5.6-sol via OpenCode inspected the concurrent lifecycle and process activity paths, implemented the per-provider PID attribution, and added deterministic regression tests. Chris Huber remains responsible for this change.

Co-authored-by: Chris Huber <chris@chubes.net>

Author: chubes4

crates/homeboy-agents/src/agent_task_lifecycle/lifecycle_ops.rs

@@ -1661,6 +1661,52 @@ pub fn reserve_provider_execution(
     Ok(reservation)
 }
 
+/// Bind a reserved provider execution to the subprocess that actually runs it.
+///
+/// Fanout workers are threads and therefore share the coordinator PID. The
+/// provider subprocess is the first process identity unique to one child run;
+/// using it keeps liveness and activity evidence from crossing child records.
+pub fn record_provider_execution_process(
+    run_id: &str,
+    task_id: &str,
+    attempt: u32,
+    pid: u32,
+) -> Result<AgentTaskRunRecord> {
+    let run_id = sanitize_run_id(run_id);
+    let key = format!("{task_id}:{attempt}");
+    let record = store::mutate_record(&run_id, |record| {
+        let Some(execution) = record.metadata["provider_executions"]
+            .as_array_mut()
+            .and_then(|executions| {
+                executions
+                    .iter_mut()
+                    .find(|execution| execution["key"] == key)
+            })
+        else {
+            return false;
+        };
+        execution["owner_pid"] = json!(pid);
+        execution["owner_linux_starttime_ticks"] =
+            json!(homeboy_core::process::linux_process_starttime_ticks(pid)
+                .ok()
+                .flatten());
+        true
+    })?;
+    record.ok_or_else(|| {
+        Error::validation_invalid_argument(
+            "provider_execution",
+            "cannot bind a process to an unreserved provider execution",
+            Some(key),
+            None,
+        )
+    })
+}
+
+/// Return the unambiguous running provider PID for activity sampling.
+pub fn running_owner_pid(run_id: &str) -> Result<Option<u32>> {
+    Ok(store::read_record(&sanitize_run_id(run_id))?.owner_pid())
+}
+
 /// Persist the controller-owned Cook phase independently of provider output.
 /// This gives foreground observers a restart-safe liveness source without
 /// treating an arbitrary provider transcript line as durable state.

crates/homeboy-agents/src/agent_task_lifecycle/records.rs

@@ -623,6 +623,22 @@ impl AgentTaskRunRecord {
     }
 
     pub(crate) fn owner_pid(&self) -> Option<u32> {
+        let provider_pids = self
+            .metadata
+            .get("provider_executions")
+            .and_then(Value::as_array)
+            .into_iter()
+            .flatten()
+            .filter(|execution| execution["state"] == "running")
+            .filter_map(|execution| {
+                execution["owner_pid"]
+                    .as_u64()
+                    .and_then(|pid| u32::try_from(pid).ok())
+            })
+            .collect::<Vec<_>>();
+        if provider_pids.len() == 1 {
+            return provider_pids.into_iter().next();
+        }
         self.metadata
             .get("runner_pid")
             .and_then(Value::as_u64)

crates/homeboy-agents/src/agent_task_provider/command_runner.rs

@@ -53,16 +53,18 @@ pub(super) fn run_materialized_provider_command(
     request: &AgentTaskExecutorRequest,
     provider: &AgentTaskExecutorProvider,
     run_id: Option<&str>,
+    execution_attempt: u32,
 ) -> AgentTaskOutcome {
-    let mut attempt = 1;
+    let mut retry_attempt = 1;
     loop {
-        let mut outcome = run_materialized_provider_command_once(request, provider);
+        let mut outcome =
+            run_materialized_provider_command_once(request, provider, run_id, execution_attempt);
         classify_transient_provider_outcome(&mut outcome);
 
         let retryable = outcome_is_transient(&outcome);
-        if !retryable || attempt >= PROVIDER_TRANSIENT_MAX_ATTEMPTS {
-            if attempt > 1 {
-                annotate_transient_retry(&mut outcome, attempt, retryable);
+        if !retryable || retry_attempt >= PROVIDER_TRANSIENT_MAX_ATTEMPTS {
+            if retry_attempt > 1 {
+                annotate_transient_retry(&mut outcome, retry_attempt, retryable);
             }
             attach_runtime_tool_provenance(request, &mut outcome);
             // Preserve and link the latest raw executor input/result as
@@ -71,11 +73,12 @@ pub(super) fn run_materialized_provider_command(
             return outcome;
         }
 
-        let backoff_ms = PROVIDER_TRANSIENT_BASE_BACKOFF_MS.saturating_mul(1u64 << (attempt - 1));
+        let backoff_ms =
+            PROVIDER_TRANSIENT_BASE_BACKOFF_MS.saturating_mul(1u64 << (retry_attempt - 1));
         if backoff_ms > 0 {
             std::thread::sleep(Duration::from_millis(backoff_ms));
         }
-        attempt += 1;
+        retry_attempt += 1;
     }
 }
 
@@ -315,12 +318,16 @@ impl ProviderContainmentReport {
 pub(super) fn run_materialized_provider_command_once(
     request: &AgentTaskExecutorRequest,
     provider: &AgentTaskExecutorProvider,
+    run_id: Option<&str>,
+    attempt: u32,
 ) -> AgentTaskOutcome {
     let mut containment_report = ProviderContainmentReport::default();
     let mut outcome = run_materialized_provider_command_once_contained(
         request,
         provider,
         &mut containment_report,
+        run_id,
+        attempt,
     );
     containment_report.annotate(&mut outcome);
     if let Err(error) = retain_failed_workspace_artifacts(&mut outcome, request) {
@@ -339,6 +346,8 @@ fn run_materialized_provider_command_once_contained(
     request: &AgentTaskExecutorRequest,
     provider: &AgentTaskExecutorProvider,
     containment_report: &mut ProviderContainmentReport,
+    run_id: Option<&str>,
+    attempt: u32,
 ) -> AgentTaskOutcome {
     let command = render_provider_command_display(provider);
     let deadline_remaining_ms = crate::agent_task_timeout::remaining_execution_deadline_ms(
@@ -554,6 +563,17 @@ fn run_materialized_provider_command_once_contained(
         }
     };
 
+    if let Some(run_id) = run_id {
+        // Failure to record this diagnostic identity must not interrupt provider
+        // execution. The reservation remains the execution authority.
+        let _ = crate::agent_task_lifecycle::record_provider_execution_process(
+            run_id,
+            &request.request.task_id,
+            attempt,
+            child.id(),
+        );
+    }
+
     if let Err(error) = containment.attach(&child) {
         let attach_error = error.to_string();
         let cleanup = containment.terminate_live(&mut child);
@@ -1084,7 +1104,7 @@ pub(super) fn run_provider_command(
     run_id: Option<&str>,
 ) -> AgentTaskOutcome {
     let materialized = test_executor_request(request);
-    run_materialized_provider_command(&materialized, provider, run_id)
+    run_materialized_provider_command(&materialized, provider, run_id, 1)
 }
 
 #[cfg(test)]
@@ -1093,7 +1113,7 @@ pub(super) fn run_provider_command_once(
     provider: &AgentTaskExecutorProvider,
 ) -> AgentTaskOutcome {
     let materialized = test_executor_request(request);
-    run_materialized_provider_command_once(&materialized, provider)
+    run_materialized_provider_command_once(&materialized, provider, None, 1)
 }
 
 #[cfg(test)]

crates/homeboy-agents/src/agent_task_provider/executor.rs

@@ -211,7 +211,12 @@ impl AgentTaskExecutorAdapter for ExtensionProviderAgentTaskExecutor {
                 &ready_tools,
             ));
 
-        run_materialized_provider_command(&request, &provider, context.run_id.as_deref())
+        run_materialized_provider_command(
+            &request,
+            &provider,
+            context.run_id.as_deref(),
+            context.attempt,
+        )
     }
 }
 

crates/homeboy-agents/src/agent_task_service/cook.rs

@@ -3705,7 +3705,13 @@ where
                             while let Err(mpsc::RecvTimeoutError::Timeout) =
                                 heartbeat_wait.recv_timeout(COOK_HEARTBEAT_INTERVAL)
                             {
-                                let activity = heartbeat_activity.sample(heartbeat_owner_pid);
+                                let activity_owner_pid = agent_task_lifecycle::running_owner_pid(
+                                    &heartbeat_run_id,
+                                )
+                                .ok()
+                                .flatten()
+                                .unwrap_or(heartbeat_owner_pid);
+                                let activity = heartbeat_activity.sample(activity_owner_pid);
                                 let tick = supervisor.observe(&activity);
                                 let detail = tick.detail_line();
                                 let _ = report_cook_progress_with_activity(

crates/homeboy-agents/src/agent_task_service/discovery.rs

@@ -552,6 +552,13 @@ fn discovery_run(
     let last_update_age_minutes = age_minutes(last_update.as_deref(), now);
     let source = run_source(&record);
     let liveness = classify.then(|| classify_liveness(&record, last_update_age_minutes, now));
+    let stale_reason = metadata_string(
+        &record.metadata,
+        agent_task_lifecycle::METADATA_KEY_STALE_RUNNING_REASON,
+    )
+    .or_else(|| {
+        (liveness == Some(AgentTaskLiveness::Stale)).then(|| stale_reason_for_record(&record))
+    });
 
     // Runner metadata describes execution placement, not necessarily lifecycle
     // record ownership. Controller handoff projections retain their durable
@@ -584,11 +591,12 @@ fn discovery_run(
         runner_id,
         runner_job_id,
         remote_run_id: metadata_string(&record.metadata, "remote_run_id"),
-        stale: metadata_bool(&record.metadata, agent_task_lifecycle::METADATA_KEY_STALE_RUNNING),
-        stale_reason: metadata_string(
-            &record.metadata,
-            agent_task_lifecycle::METADATA_KEY_STALE_RUNNING_REASON,
-        ),
+        stale: (liveness == Some(AgentTaskLiveness::Stale))
+            .then_some(true)
+            .or_else(|| {
+                metadata_bool(&record.metadata, agent_task_lifecycle::METADATA_KEY_STALE_RUNNING)
+            }),
+        stale_reason,
         retryable: metadata_bool(&record.metadata, agent_task_lifecycle::METADATA_KEY_RETRYABLE),
         liveness,
         liveness_reconcilable: liveness.map(AgentTaskLiveness::is_reconcilable),
@@ -613,6 +621,21 @@ fn discovery_run(
     }
 }
 
+/// Explain every stale read projection, including a pure discovery read that
+/// deliberately leaves the durable record untouched.
+fn stale_reason_for_record(record: &AgentTaskRunRecord) -> String {
+    if record.owner_pid().is_some() && !record.owner_process_is_running() {
+        return "owner_process_not_running".to_string();
+    }
+    if record.runner_job_id().is_some() {
+        return "runner_job_unverified_after_daemon_restart".to_string();
+    }
+    if record.owner_pid().is_none() {
+        return "missing_runner_pid".to_string();
+    }
+    "stale_running_marked".to_string()
+}
+
 fn discovery_counts(tasks: &[agent_task_lifecycle::AgentTaskRunTask]) -> AgentTaskDiscoveryCounts {
     let mut counts = AgentTaskDiscoveryCounts::default();
     for task in tasks {

crates/homeboy-agents/src/agent_task_service/process_activity.rs

@@ -631,6 +631,34 @@ mod tests {
         assert_eq!(activity.rss_kib, Some(262_144));
     }
 
+    #[test]
+    fn concurrent_provider_roots_do_not_cross_attribute_activity_or_resources() {
+        // Both provider children share a coordinator parent. Sampling from that
+        // parent would blend their commands and resource trees; each child PID
+        // must be used as the root instead.
+        let rows = parse_process_activity_rows(concat!(
+            "1 0 S 1024 10:00 homeboy agent-task fanout cook\n",
+            "101 1 S 2048 05:00 opencode run child-a\n",
+            "102 101 S 4096 04:00 cargo test child-a\n",
+            "201 1 S 8192 05:00 opencode run child-b\n",
+            "202 201 S 16384 04:00 cargo test child-b\n",
+        ));
+
+        let first = select_provider_activity(&rows, 101, &[]);
+        let second = select_provider_activity(&rows, 201, &[]);
+
+        assert_eq!(
+            first.activity.expect("first activity").command,
+            "cargo test child-a"
+        );
+        assert_eq!(
+            second.activity.expect("second activity").command,
+            "cargo test child-b"
+        );
+        assert_eq!(first.tree_rss_kib, Some(4_096));
+        assert_eq!(second.tree_rss_kib, Some(16_384));
+    }
+
     #[test]
     fn an_unmeasured_tree_reports_no_memory_rather_than_zero() {
         // A budget that fired on a fabricated zero would be worse than no

crates/homeboy-agents/src/agent_task_service/tests.rs

@@ -1541,6 +1541,81 @@ fn dead_owner_process_run_is_classified_stale_and_reconciled() {
     });
 }
 
+#[test]
+fn concurrent_provider_children_keep_liveness_attributed_to_their_own_processes() {
+    with_isolated_home(|_| {
+        let mut child_a = std::process::Command::new("sleep")
+            .arg("60")
+            .spawn()
+            .expect("start first provider child");
+        let mut child_b = std::process::Command::new("sleep")
+            .arg("60")
+            .spawn()
+            .expect("start second provider child");
+
+        let result = (|| {
+            for (run_id, pid) in [("run-child-a", child_a.id()), ("run-child-b", child_b.id())] {
+                let plan = discovery_plan();
+                agent_task_lifecycle::submit_plan(&plan, Some(run_id)).expect("submit child");
+                agent_task_lifecycle::mark_running(run_id).expect("mark child running");
+                agent_task_lifecycle::reserve_provider_execution(run_id, &plan.tasks[0], 1)
+                    .expect("reserve provider execution");
+                agent_task_lifecycle::record_provider_execution_process(
+                    run_id,
+                    &plan.tasks[0].task_id,
+                    1,
+                    pid,
+                )
+                .expect("bind provider process");
+            }
+
+            let active =
+                discover_runs(AgentTaskDiscoveryFilter::Active).expect("discover children");
+            assert!(active
+                .runs
+                .iter()
+                .all(|run| run.liveness == Some(AgentTaskLiveness::Active)));
+
+            child_a.kill().expect("kill first provider child");
+            child_a.wait().expect("reap first provider child");
+            let observed =
+                discover_runs(AgentTaskDiscoveryFilter::Active).expect("rediscover children");
+            assert_eq!(
+                observed
+                    .runs
+                    .iter()
+                    .find(|run| run.run_id == "run-child-a")
+                    .expect("first child")
+                    .liveness,
+                Some(AgentTaskLiveness::Stale)
+            );
+            assert_eq!(
+                observed
+                    .runs
+                    .iter()
+                    .find(|run| run.run_id == "run-child-a")
+                    .expect("first child")
+                    .stale_reason
+                    .as_deref(),
+                Some("owner_process_not_running")
+            );
+            assert_eq!(
+                observed
+                    .runs
+                    .iter()
+                    .find(|run| run.run_id == "run-child-b")
+                    .expect("second child")
+                    .liveness,
+                Some(AgentTaskLiveness::Active)
+            );
+        })();
+
+        let _ = child_b.kill();
+        let _ = child_b.wait();
+        result
+    });
+}
+
 #[test]
 fn discovery_latest_returns_only_newest_run() {
     with_isolated_home(|_| {