[WIP] Isolate stale failed-agent state between sessions (#5110)

* Initial plan

* Changes before error encountered

Agent-Logs-Url: https://github.com/Hmbown/CodeWhale/sessions/3a6b0bb2-7714-4b43-b421-7724e5a5adb9

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>

crates/tui/src/automation_manager.rs

@@ -1473,6 +1473,7 @@ async fn enqueue_run_task(
         allow_shell: Some(automation.task_allow_shell()),
         trust_mode: Some(automation.task_trust_mode()),
         auto_approve: Some(automation.task_auto_approve()),
+        owner_session_id: None,
     };
 
     match task_manager.add_task(new_task).await {

crates/tui/src/task_manager.rs

@@ -228,6 +228,8 @@ pub struct TaskRecord {
     pub thread_id: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub turn_id: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub owner_session_id: Option<String>,
     #[serde(default)]
     pub runtime_event_count: usize,
     /// Monotonic owner-lifecycle sequence used by Work Graph reconciliation.
@@ -272,6 +274,8 @@ pub struct TaskSummary {
     pub thread_id: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub turn_id: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub owner_session_id: Option<String>,
 }
 
 impl From<&TaskRecord> for TaskSummary {
@@ -292,6 +296,7 @@ impl From<&TaskRecord> for TaskSummary {
             error: value.error.clone(),
             thread_id: value.thread_id.clone(),
             turn_id: value.turn_id.clone(),
+            owner_session_id: value.owner_session_id.clone(),
         }
     }
 }
@@ -316,6 +321,7 @@ pub struct NewTaskRequest {
     pub allow_shell: Option<bool>,
     pub trust_mode: Option<bool>,
     pub auto_approve: Option<bool>,
+    pub owner_session_id: Option<String>,
 }
 
 impl NewTaskRequest {
@@ -330,6 +336,7 @@ impl NewTaskRequest {
             allow_shell: None,
             trust_mode: None,
             auto_approve: Some(true),
+            owner_session_id: None,
         }
     }
 }
@@ -936,6 +943,7 @@ impl TaskManager {
             error: None,
             thread_id: None,
             turn_id: None,
+            owner_session_id: req.owner_session_id,
             runtime_event_count: 0,
             lifecycle_seq: 1,
             checklist: TaskChecklistState::default(),
@@ -1993,7 +2001,10 @@ mod tests {
                 .await?;
 
         let task = manager
-            .add_task(NewTaskRequest::from_prompt("test persistence"))
+            .add_task(NewTaskRequest {
+                owner_session_id: Some("session-persist".to_string()),
+                ..NewTaskRequest::from_prompt("test persistence")
+            })
             .await?;
         let finished = wait_for_terminal_state(&manager, &task.id, Duration::from_secs(10)).await?;
         assert_eq!(finished.status, TaskStatus::Completed);
@@ -2013,6 +2024,11 @@ mod tests {
                 .await?;
         let loaded = recovered.get_task(&task.id).await?;
         assert_eq!(loaded.status, TaskStatus::Completed);
+        assert_eq!(
+            loaded.owner_session_id.as_deref(),
+            Some("session-persist"),
+            "session ownership should survive persistence and restart"
+        );
         assert!(!loaded.timeline.is_empty());
         assert_eq!(loaded.checklist.items[0].content, "read fixture");
         Ok(())
@@ -2175,6 +2191,7 @@ mod tests {
             error: None,
             thread_id: Some("thr_stale".to_string()),
             turn_id: Some("turn_stale".to_string()),
+            owner_session_id: Some("session-old".to_string()),
             runtime_event_count: 0,
             lifecycle_seq: 2,
             checklist: TaskChecklistState::default(),
@@ -2402,6 +2419,7 @@ mod tests {
             allow_shell: None,
             trust_mode: None,
             auto_approve: None,
+            owner_session_id: None,
         };
         let task = manager.add_task(req).await?;
 

crates/tui/src/tools/tasks.rs

@@ -456,6 +456,7 @@ impl TasksTool {
             allow_shell: input.get("allow_shell").and_then(Value::as_bool),
             trust_mode: input.get("trust_mode").and_then(Value::as_bool),
             auto_approve: input.get("auto_approve").and_then(Value::as_bool),
+            owner_session_id: Some(context.state_namespace.clone()),
         };
         let task_id = crate::task_manager::TaskManager::new_task_id();
         if let Some(work) = context.runtime.work.as_ref() {

crates/tui/src/tui/subagent_routing.rs

@@ -823,14 +823,25 @@ pub(super) fn format_task_list(tasks: &[TaskSummary]) -> String {
     }
 
     let show_verdict = tasks.iter().any(|task| task.hunt_verdict.is_some());
+    let show_session = tasks.iter().any(|task| task.owner_session_id.is_some());
     let mut lines = vec![format!("Tasks ({})", tasks.len())];
     // Build headers with the same format strings as the rows so the ID
     // column (21-char `task_` ids) can never drift out of alignment again.
-    if show_verdict {
+    if show_verdict && show_session {
+        lines.push(format!(
+            "{:<21}  {:<9}  {:<7}  {:<12}  {:>8}  {}",
+            "ID", "Status", "Verdict", "Session", "Time", "Title"
+        ));
+    } else if show_verdict {
         lines.push(format!(
             "{:<21}  {:<9}  {:<7}  {:>8}  {}",
             "ID", "Status", "Verdict", "Time", "Title"
         ));
+    } else if show_session {
+        lines.push(format!(
+            "{:<21}  {:<9}  {:<12}  {:>8}  {}",
+            "ID", "Status", "Session", "Time", "Title"
+        ));
     } else {
         lines.push(format!(
             "{:<21}  {:<9}  {:>8}  {}",
@@ -843,7 +854,23 @@ pub(super) fn format_task_list(tasks: &[TaskSummary]) -> String {
             .duration_ms
             .map(crate::elapsed::format_elapsed_ms)
             .unwrap_or_else(|| "-".to_string());
-        if show_verdict {
+        let owner_session = task.owner_session_id.as_deref().unwrap_or("-");
+        let owner_session = if owner_session.chars().count() > 12 {
+            format!("{}…", owner_session.chars().take(11).collect::<String>())
+        } else {
+            owner_session.to_string()
+        };
+        if show_verdict && show_session {
+            lines.push(format!(
+                "{:<21}  {:<9}  {:<7}  {:<12}  {:>8}  {}",
+                task.id,
+                task_status_label(task.status),
+                hunt_verdict_glyph(task.hunt_verdict.as_deref()),
+                owner_session,
+                duration,
+                task.prompt_summary
+            ));
+        } else if show_verdict {
             lines.push(format!(
                 "{:<21}  {:<9}  {:<7}  {:>8}  {}",
                 task.id,
@@ -852,6 +879,15 @@ pub(super) fn format_task_list(tasks: &[TaskSummary]) -> String {
                 duration,
                 task.prompt_summary
             ));
+        } else if show_session {
+            lines.push(format!(
+                "{:<21}  {:<9}  {:<12}  {:>8}  {}",
+                task.id,
+                task_status_label(task.status),
+                owner_session,
+                duration,
+                task.prompt_summary
+            ));
         } else {
             lines.push(format!(
                 "{:<21}  {:<9}  {:>8}  {}",
@@ -890,6 +926,9 @@ fn format_task_detail(task: &TaskRecord) -> String {
         "Workspace: {}",
         crate::utils::display_path(&task.workspace)
     ));
+    if let Some(owner_session_id) = task.owner_session_id.as_deref() {
+        lines.push(format!("Owning Session: {owner_session_id}"));
+    }
     if let Some(thread_id) = task.thread_id.as_ref() {
         lines.push(format!("Runtime Thread: {thread_id}"));
     }
@@ -1033,6 +1072,7 @@ mod tests {
             error: None,
             thread_id: None,
             turn_id: None,
+            owner_session_id: None,
         }
     }
 
@@ -1103,6 +1143,20 @@ mod tests {
         assert!(output.contains(&format!("{:<21}  {:<9}  ×", "task_escaped", "failed")));
     }
 
+    #[test]
+    fn task_list_shows_owner_session_when_present() {
+        let mut task = task_summary("task_owned", TaskStatus::Running, None);
+        task.owner_session_id = Some("session-123456".to_string());
+
+        let output = format_task_list(&[task]);
+
+        assert!(output.contains(&format!(
+            "{:<21}  {:<9}  {:<12}  {:>8}  {}",
+            "ID", "Status", "Session", "Time", "Title"
+        )));
+        assert!(output.contains("session-1234…"), "{output}");
+    }
+
     #[test]
     fn mailbox_progress_reports_transcript_change_only_for_visible_card_updates() {
         let mut app = App::new(test_options(), &Config::default());

crates/tui/src/tui/ui.rs

@@ -2542,14 +2542,15 @@ fn build_app_system_prompt_with_goal(
 /// Choose which durable-task summaries should appear in the Work
 /// sidebar's Tasks panel.
 ///
-/// Active tasks (`Queued`/`Running`) are always included. Terminal
-/// tasks (`Completed`/`Failed`/`Canceled`) are session-local receipts: both
-/// their creation and completion must fall within this TUI session. Durable
-/// tasks are stored per user rather than per TUI process, and startup recovery
-/// can stamp an old running record with a fresh `ended_at`. Treating that as a
-/// current receipt makes a new same-workspace instance look failed (#4416).
-/// Shared and older task history remains available explicitly through
-/// `/tasks`; it does not belong on the fresh live-work surface by default.
+/// Tasks stamped with the current session owner stay visible on that session's
+/// live surface. Tasks owned by a different session stay in explicit history
+/// (`/tasks`) instead of appearing as live workspace work. Legacy unowned
+/// records fall back to the v0.9.1 timestamp gate: active tasks remain visible,
+/// while terminal receipts must have both creation and completion times inside
+/// this TUI session. Durable tasks are stored per user rather than per TUI
+/// process, and startup recovery can stamp an old running record with a fresh
+/// `ended_at`. Treating that as a current receipt makes a new same-workspace
+/// instance look failed (#4416).
 ///
 /// A terminal task missing `ended_at` is treated as not current and
 /// dropped: durable tasks always stamp `ended_at` when they reach a
@@ -2558,16 +2559,34 @@ fn build_app_system_prompt_with_goal(
 pub(crate) fn select_work_sidebar_tasks(
     tasks: Vec<TaskSummary>,
     session_started_at: chrono::DateTime<chrono::Utc>,
+    current_session_id: Option<&str>,
 ) -> Vec<TaskSummary> {
     tasks
         .into_iter()
-        .filter(|task| match task.status {
-            TaskStatus::Queued | TaskStatus::Running => true,
-            TaskStatus::Completed | TaskStatus::Failed | TaskStatus::Canceled => {
-                task.created_at >= session_started_at
-                    && task
-                        .ended_at
-                        .is_some_and(|ended_at| ended_at >= session_started_at)
+        .filter(|task| {
+            let owner_matches_current = current_session_id
+                .zip(task.owner_session_id.as_deref())
+                .is_some_and(|(current, owner)| current == owner);
+            let owned_by_other_session = current_session_id.is_some()
+                && task
+                    .owner_session_id
+                    .as_deref()
+                    .is_some_and(|owner| Some(owner) != current_session_id);
+            if owned_by_other_session {
+                return false;
+            }
+            match task.status {
+                TaskStatus::Queued | TaskStatus::Running => {
+                    owner_matches_current || task.owner_session_id.is_none()
+                }
+                TaskStatus::Completed | TaskStatus::Failed | TaskStatus::Canceled => {
+                    owner_matches_current
+                        || (task.owner_session_id.is_none()
+                            && task.created_at >= session_started_at
+                            && task
+                                .ended_at
+                                .is_some_and(|ended_at| ended_at >= session_started_at))
+                }
             }
         })
         .collect()
@@ -2620,10 +2639,11 @@ async fn refresh_active_task_panel(app: &mut App, task_manager: &SharedTaskManag
         tracing::warn!(error = %err, "durable task lifecycle checkpoint remains pending");
     }
     let session_started_at = app.session_started_at;
-    let mut entries: Vec<TaskPanelEntry> = select_work_sidebar_tasks(tasks, session_started_at)
-        .into_iter()
-        .map(task_summary_to_panel_entry)
-        .collect();
+    let mut entries: Vec<TaskPanelEntry> =
+        select_work_sidebar_tasks(tasks, session_started_at, app.current_session_id.as_deref())
+            .into_iter()
+            .map(task_summary_to_panel_entry)
+            .collect();
 
     entries.extend(active_rlm_task_entries(app));
 
@@ -12490,6 +12510,7 @@ async fn apply_command_result(
                     allow_shell: Some(app.allow_shell),
                     trust_mode: Some(app.trust_mode),
                     auto_approve: Some(app_auto_approve_enabled(app)),
+                    owner_session_id: app.current_session_id.clone(),
                 };
                 match task_manager.add_task(request).await {
                     Ok(task) => {

crates/tui/src/tui/ui/tests.rs

@@ -19972,6 +19972,7 @@ mod work_sidebar_projection_tests {
         status: TaskStatus,
         created_at: chrono::DateTime<Utc>,
         ended_at: Option<chrono::DateTime<Utc>>,
+        owner_session_id: Option<&str>,
     ) -> TaskSummary {
         TaskSummary {
             id: id.to_string(),
@@ -19989,6 +19990,7 @@ mod work_sidebar_projection_tests {
             error: None,
             thread_id: None,
             turn_id: None,
+            owner_session_id: owner_session_id.map(str::to_string),
         }
     }
 
@@ -20002,12 +20004,14 @@ mod work_sidebar_projection_tests {
             TaskStatus::Running,
             session_started_at - Duration::minutes(30),
             None,
+            Some("session-current"),
         );
         let active_queued = sample_task(
             "active_q",
             TaskStatus::Queued,
             session_started_at - Duration::minutes(20),
             None,
+            Some("session-current"),
         );
 
         // Created and completed during the current session — must show.
@@ -20016,6 +20020,7 @@ mod work_sidebar_projection_tests {
             TaskStatus::Completed,
             session_started_at + Duration::seconds(5),
             Some(session_started_at + Duration::seconds(30)),
+            Some("session-current"),
         );
 
         // Completed shortly before the session started — shared history, not
@@ -20025,6 +20030,7 @@ mod work_sidebar_projection_tests {
             TaskStatus::Failed,
             session_started_at - Duration::minutes(20),
             Some(session_started_at - Duration::minutes(15)),
+            Some("session-old"),
         );
 
         // Exact restart regression: a prior-session running record is marked
@@ -20035,6 +20041,7 @@ mod work_sidebar_projection_tests {
             TaskStatus::Failed,
             session_started_at - Duration::minutes(30),
             Some(session_started_at + Duration::seconds(1)),
+            Some("session-old"),
         );
 
         // Stale completed from 6 days ago (the exact scenario in #1913) —
@@ -20044,18 +20051,28 @@ mod work_sidebar_projection_tests {
             TaskStatus::Completed,
             session_started_at - Duration::days(6) - Duration::minutes(1),
             Some(session_started_at - Duration::days(6)),
+            Some("session-old"),
         );
         let stale_canceled = sample_task(
             "stale_cancel",
             TaskStatus::Canceled,
             session_started_at - Duration::days(7) - Duration::minutes(1),
             Some(session_started_at - Duration::days(7)),
+            Some("session-old"),
         );
         let stale_failed = sample_task(
             "stale_fail",
             TaskStatus::Failed,
             session_started_at - Duration::days(3) - Duration::minutes(1),
             Some(session_started_at - Duration::days(3)),
+            Some("session-old"),
+        );
+        let active_sibling = sample_task(
+            "sibling_run",
+            TaskStatus::Running,
+            session_started_at - Duration::minutes(2),
+            None,
+            Some("session-sibling"),
         );
 
         // A terminal task without `ended_at` shouldn't sneak through.
@@ -20064,6 +20081,14 @@ mod work_sidebar_projection_tests {
             TaskStatus::Completed,
             session_started_at + Duration::seconds(1),
             None,
+            Some("session-current"),
+        );
+        let legacy_finished = sample_task(
+            "legacy_done",
+            TaskStatus::Completed,
+            session_started_at + Duration::seconds(2),
+            Some(session_started_at + Duration::seconds(3)),
+            None,
         );
 
         let tasks = vec![
@@ -20075,10 +20100,12 @@ mod work_sidebar_projection_tests {
             stale_completed.clone(),
             stale_canceled.clone(),
             stale_failed.clone(),
+            active_sibling.clone(),
             terminal_no_timestamp.clone(),
+            legacy_finished.clone(),
         ];
 
-        let kept = select_work_sidebar_tasks(tasks, session_started_at);
+        let kept = select_work_sidebar_tasks(tasks, session_started_at, Some("session-current"));
         let kept_ids: Vec<&str> = kept.iter().map(|t| t.id.as_str()).collect();
 
         assert!(
@@ -20102,6 +20129,10 @@ mod work_sidebar_projection_tests {
             !kept_ids.contains(&"recovered_old_run"),
             "startup recovery must not turn an old task into a fresh red row: {kept_ids:?}"
         );
+        assert!(
+            !kept_ids.contains(&"sibling_run"),
+            "active sibling-session task must stay off the new session's live work surface: {kept_ids:?}"
+        );
 
         assert!(
             !kept_ids.contains(&"stale_done"),
@@ -20119,6 +20150,10 @@ mod work_sidebar_projection_tests {
             !kept_ids.contains(&"ghost"),
             "terminal task missing ended_at must be hidden: {kept_ids:?}"
         );
+        assert!(
+            kept_ids.contains(&"legacy_done"),
+            "legacy unowned task still uses timestamp fallback during migration: {kept_ids:?}"
+        );
     }
 
     #[test]
@@ -20131,13 +20166,31 @@ mod work_sidebar_projection_tests {
             TaskStatus::Completed,
             session_started_at,
             Some(session_started_at),
+            None,
         );
 
-        let kept = select_work_sidebar_tasks(vec![at_boundary], session_started_at);
+        let kept = select_work_sidebar_tasks(vec![at_boundary], session_started_at, None);
         assert_eq!(kept.len(), 1);
         assert_eq!(kept[0].id, "boundary");
     }
 
+    #[test]
+    fn work_sidebar_keeps_current_session_owned_terminals_after_restore() {
+        let session_started_at = Utc.with_ymd_and_hms(2026, 5, 23, 10, 0, 0).unwrap();
+        let restored = sample_task(
+            "restored_done",
+            TaskStatus::Completed,
+            session_started_at - Duration::days(1),
+            Some(session_started_at - Duration::hours(23)),
+            Some("session-current"),
+        );
+
+        let kept =
+            select_work_sidebar_tasks(vec![restored], session_started_at, Some("session-current"));
+        assert_eq!(kept.len(), 1);
+        assert_eq!(kept[0].id, "restored_done");
+    }
+
     #[test]
     fn receipt_summary_truncation_does_not_panic_on_multibyte_boundary() {
         // Build a summary where byte 57 falls mid-character (em dash is 3 bytes).