fix(agents): retain session roster and guide safe checkpoint continuation

The Agents panel now merges retained session receipts with live workers so
completed agents do not disappear when their 45-second display cache expires.
Keep newer live rows authoritative and scope retained Agents/Price/command
rosters to the exact conversation; parked and waiting remain distinct.

Park notifications now recommend duplicate-guarded followup continuation,
identify the successor ID and preserve predecessor lineage. Explicit
start+resume_from remains a separate transcript fork.

Validation: 37 targeted regressions passed, 0 failed (12,575 filtered).
Includes cache expiry with another active worker, owner isolation, parked vs
waiting, repeated public followup, saved authority and manifest lineage.
Strict workspace/all-target/all-feature Clippy and formatting/diff passed.
A prior Clippy attempt caught concurrent goal edits; their owner restored
them before the passing rerun. GUI acceptance remains paused.

crates/tui/src/tools/subagent/mod.rs

@@ -6097,6 +6097,7 @@ impl SubAgentManager {
                 .unwrap_or(false),
             claim_pre_namespaced: claim.is_some(),
             preserve_runtime_profile: preserved_profile,
+            resume_from_agent_id: Some(agent_id.clone()),
             ..Default::default()
         };
         let resumed = self.spawn_background_with_assignment_options(
@@ -8611,7 +8612,7 @@ impl ToolSpec for AgentTool {
                 "action": {
                     "type": "string",
                     "enum": ["start", "roster", "status", "peek", "message", "followup", "interrupt", "wait", "claim", "release", "cancel"],
-                    "description": "start launches a turn-owned worker and returns immediately. roster lists roles with their resolved routes and capability/cost evidence. status/peek inspect running or retained workers. message queues a note without waking a running child. followup delivers queued notes and wakes a running child for its next user-provenance model turn. interrupt stops the current turn while preserving the child checkpoint. wait only observes; see until. claim widens your own enforced write scope (see write_roots). release clears write claims whose owner is no longer running — the remediation a write-scope contention refusal names; pass agent_id to clear one, omit it to sweep. cancel permanently cancels a running child."
+                    "description": "start launches a turn-owned worker and returns immediately. roster lists roles with their resolved routes and capability/cost evidence. status/peek inspect running or retained workers. message queues a note without waking a running child. followup delivers notes to a running child or continues an interrupted child from its checkpoint; use the returned agent_id for subsequent waits/messages. Retrying followup on the original interrupted id reuses its successor in this runtime. interrupt stops the current turn while preserving the child checkpoint. wait only observes; see until. claim widens your own enforced write scope (see write_roots). release clears write claims whose owner is no longer running — the remediation a write-scope contention refusal names; pass agent_id to clear one, omit it to sweep. cancel permanently cancels a running child."
                 },
                 "until": {
                     "type": "string",
@@ -8624,7 +8625,7 @@ impl ToolSpec for AgentTool {
                 },
                 "message": {
                     "type": "string",
-                    "description": "Parent note for action=message or action=followup. message queues only; followup also wakes a running child."
+                    "description": "Parent note for action=message or action=followup. message queues only; followup wakes a running child or continues a parked child from its checkpoint."
                 },
                 "name": {
                     "type": "string",
@@ -8672,7 +8673,7 @@ impl ToolSpec for AgentTool {
                 },
                 "resume_from": {
                     "type": "string",
-                    "description": "Settled child agent_id or session name to continue. The source must not be running. Its full transcript is loaded and prepended as the new child's context (fork_context=true), continuing the transcript lineage under a new role or profile (e.g. explore → implementer → verifier). Mutually exclusive with fork_context=false. Cross-workspace or missing sources are rejected with a clear error."
+                    "description": "Settled child agent_id or session name to fork into a separate new worker. Repeating start with resume_from creates another independent worker; use action=followup to continue parked work without an accidental duplicate. The source must not be running. Its full transcript is loaded and prepended as the new child's context (fork_context=true), continuing the transcript lineage under a new role or profile (e.g. explore → implementer → verifier). Mutually exclusive with fork_context=false. Cross-workspace or missing sources are rejected with a clear error."
                 }
             },
             "dependentSchemas": {
@@ -11416,7 +11417,7 @@ fn subagent_cancellation_projection(
         turn_end_parking.is_some_and(|signal| signal.load(std::sync::atomic::Ordering::Acquire));
     if parking_requested {
         let reason = format!(
-            "Parent turn ended before this turn-owned child settled. Work was parked instead of discarded; resume with agent(action=\"start\", prompt=\"Continue the parked assignment.\", resume_from=\"{agent_id}\")."
+            "Parent turn ended before this turn-owned child settled. Work was parked instead of discarded; continue with agent(action=\"followup\", agent_id=\"{agent_id}\", message=\"Continue the parked assignment.\"). Use the returned agent_id for subsequent waits and messages."
         );
         let mut checkpoint = build_subagent_checkpoint(agent_id, &reason, messages, steps, true);
         // Parked, not asking: nothing will answer this child, and its write
@@ -11424,7 +11425,7 @@ fn subagent_cancellation_projection(
         checkpoint.parked_at_turn_end = true;
         let needs_input = SubAgentNeedsInput {
             question: format!(
-                "Resume this parked child with agent(action=\"start\", prompt=\"Continue the parked assignment.\", resume_from=\"{agent_id}\")."
+                "Continue this parked child with agent(action=\"followup\", agent_id=\"{agent_id}\", message=\"Continue the parked assignment.\"). This returns a successor agent_id; the original parked receipt remains intact."
             ),
         };
         return (

crates/tui/src/tools/subagent/tests.rs

@@ -20628,6 +20628,98 @@ async fn resume_from_checkpoint_spawns_seeded_agent_with_checkpoint_context() {
     assert!(matches!(prior.status, SubAgentStatus::Interrupted(_)));
 }
 
+#[tokio::test]
+async fn parked_followup_reuses_successor_and_preserves_route_authority_and_lineage() {
+    let tmp = tempdir().unwrap();
+    let manager = new_shared_subagent_manager(tmp.path().to_path_buf(), 4);
+    let saved_profile = WorkerRuntimeProfile::for_role(FleetRole::Scout);
+    let saved_route = ChildRouteReceipt {
+        requested_type: "explore".into(),
+        requested_profile: None,
+        resolved_profile_id: None,
+        profile_origin: None,
+        canonical_role: "explore".into(),
+        provider_id: "deepseek".into(),
+        model_id: "deepseek-v4-flash".into(),
+        route_source: "role.pin".into(),
+        requested_reasoning: "inherit".into(),
+        effective_reasoning: None,
+        runtime_version: "fixture".into(),
+        runtime_build_sha: "fixture".into(),
+    };
+    let agent_id = {
+        let mut guard = manager.write().await;
+        let id = guard.insert_test_running_agent("parked-inventory", tmp.path());
+        let agent = guard.agents.get_mut(&id).unwrap();
+        agent.agent_type = FleetRole::Scout;
+        agent.model = saved_route.model_id.clone();
+        agent.allowed_tools = Some(Vec::new());
+        let spec = &mut guard.worker_records.get_mut(&id).unwrap().spec;
+        spec.agent_type = FleetRole::Scout;
+        spec.model = saved_route.model_id.clone();
+        spec.runtime_profile = saved_profile.clone();
+        spec.child_route = Some(saved_route.clone());
+        park_child_at_turn_end(&mut guard, &id);
+        id
+    };
+    let prior = manager.read().await.get_result(&agent_id).unwrap();
+    let instruction = prior.needs_input.as_ref().unwrap().question.as_str();
+    assert!(instruction.contains("action=\"followup\""), "{instruction}");
+    assert!(instruction.contains(&format!("agent_id=\"{agent_id}\"")));
+
+    // Reuse the existing loopback client so a scheduled child cannot contact
+    // a provider. The public tool must preserve the child's read-only profile
+    // even though the caller permits writes and has a different workspace.
+    let (client, _calls, _bodies) = delayed_chat_client(Duration::from_secs(30), "done").await;
+    let mut runtime = stub_runtime();
+    runtime.client = client;
+    runtime.manager = Arc::clone(&manager);
+    assert_ne!(runtime.context.workspace, tmp.path());
+    let tool = AgentTool::new(Arc::clone(&manager), runtime);
+    let context = ToolContext::new(tmp.path());
+    let mut successors = Vec::new();
+    for message in [
+        "Continue the parked assignment.",
+        "Include the remaining checks.",
+    ] {
+        let result = tool
+            .execute(
+                json!({"action": "followup", "agent_id": agent_id, "message": message}),
+                &context,
+            )
+            .await
+            .expect("the emitted recovery action must execute");
+        let receipt: Value = serde_json::from_str(&result.content).unwrap();
+        assert_eq!(receipt["continued_from_checkpoint"], true);
+        successors.push(receipt["agent_id"].as_str().unwrap().to_string());
+    }
+    assert_eq!(successors[0], successors[1]);
+    assert_ne!(successors[0], agent_id);
+
+    let mut guard = manager.write().await;
+    assert_eq!(guard.agents.len(), 2, "one predecessor and one successor");
+    let spec = &guard.worker_records.get(&successors[0]).unwrap().spec;
+    assert_eq!(spec.child_route.as_ref(), Some(&saved_route));
+    assert_eq!(spec.runtime_profile.role, saved_profile.role);
+    assert_eq!(spec.runtime_profile.permissions, saved_profile.permissions);
+    assert_eq!(spec.runtime_profile.shell, saved_profile.shell);
+    assert_eq!(spec.runtime_profile.tools, saved_profile.tools);
+    let manifest = spec.launch_manifest.as_ref().unwrap();
+    assert_eq!(
+        manifest.resume_from_agent_id.as_deref(),
+        Some(agent_id.as_str())
+    );
+    assert_eq!(manifest.cwd.as_deref(), tmp.path().to_str());
+    let after = guard.get_result(&agent_id).unwrap();
+    assert_eq!(after.status, prior.status);
+    assert_eq!(after.checkpoint, prior.checkpoint);
+    assert!(
+        guard.child_was_woken(&successors[0])
+            || guard.queued_mail_depth(&successors[0]).unwrap_or(0) >= 1
+    );
+    let _ = guard.cancel_agent(&successors[0]);
+}
+
 #[tokio::test]
 async fn resume_from_checkpoint_is_idempotent_across_repeated_followups() {
     let tmp = tempdir().unwrap();

crates/tui/src/tui/app.rs

@@ -1910,9 +1910,12 @@ pub struct App {
     /// boundary (`agent_id` → count), from the latest `AgentList` refresh.
     pub agent_queued_follow_ups: HashMap<String, usize>,
     /// Receipts-only roster of every agent that ran this session (#5479).
-    /// Refreshed wholesale on each `AgentList` event; rendered by `/agents`
-    /// and, in a later slice, by the agents rail.
+    /// Refreshed wholesale on each `AgentList` event; shared by `/agents`,
+    /// the Agents register and the Price view.
     pub agent_roster: Vec<crate::tui::agent_roster::AgentRosterRow>,
+    /// Original conversation owner of the retained snapshot. A process boot
+    /// marker or a worker's parent run is not a conversation identity.
+    pub agent_roster_session_id: Option<String>,
     /// `/agents list` asked for a one-shot transcript listing. Cleared by the
     /// `AgentList` handler that prints it.
     pub agent_roster_print_requested: bool,
@@ -2556,6 +2559,22 @@ fn push_enabled_provider_model(
 }
 
 impl App {
+    /// A retained roster remains readable only in its owning conversation.
+    pub(crate) fn current_agent_roster(&self) -> &[crate::tui::agent_roster::AgentRosterRow] {
+        if self
+            .current_session_id
+            .as_deref()
+            .is_some_and(|session_id| {
+                !session_id.is_empty()
+                    && self.agent_roster_session_id.as_deref() == Some(session_id)
+            })
+        {
+            &self.agent_roster
+        } else {
+            &[]
+        }
+    }
+
     /// Who owns the keyboard right now.
     ///
     /// The single derivation of [`Focus`], mirroring the order in which the

crates/tui/src/tui/app/init.rs

@@ -1028,6 +1028,7 @@ impl App {
             active_cell_revision: 0,
             active_tool_details: HashMap::new(),
             agent_roster: Vec::new(),
+            agent_roster_session_id: None,
             agent_roster_print_requested: false,
             active_tool_entry_completed_at: HashMap::new(),
             exploring_cell: None,

crates/tui/src/tui/ui/event_loop.rs

@@ -3557,9 +3557,10 @@ pub(crate) async fn run_event_loop(
                     {
                         app.agent_queued_follow_ups = queued_follow_ups;
                         app.agent_roster = roster;
+                        app.agent_roster_session_id = Some(owner_session_id);
                         if std::mem::take(&mut app.agent_roster_print_requested) {
                             let content = crate::tui::agent_roster::render_agent_roster(
-                                &app.agent_roster,
+                                app.current_agent_roster(),
                                 "main",
                             );
                             app.add_message(crate::tui::history::HistoryCell::System { content });

crates/tui/src/tui/work_surface/model.rs

@@ -699,10 +699,63 @@ fn goal_row_label(app: &App) -> Option<String> {
 /// `▾ Subagents N` group door, oldest-first as the runtime reports them.
 fn agents_view_rows(app: &mut App) -> Vec<WorkRow> {
     let rows = project(app);
-    let agents: Vec<WorkRow> = rows
+    let mut agents: Vec<WorkRow> = rows
         .into_iter()
         .filter(|row| row.id.0.starts_with("worker:"))
         .collect();
+    // The compact cache expires settled cards after 45 seconds. The explicit
+    // register is session history: retain its receipt rows, preserving fresh
+    // cache/progress projections by worker ID when both sources know a worker.
+    let mut retained = Vec::new();
+    let mut seen = HashSet::new();
+    for receipt in app.current_agent_roster() {
+        let id = WorkRowId(format!("worker:{}", receipt.worker_id));
+        if !seen.insert(id.clone()) {
+            continue;
+        }
+        if let Some(index) = agents.iter().position(|row| row.id == id) {
+            retained.push(agents.remove(index));
+            continue;
+        }
+        let parked = receipt.state == crate::tui::agent_roster::RosterState::Parked;
+        let status = if parked {
+            current_activity_status_label(AgentCurrentActivityStatus::Parked, app.ui_locale)
+        } else {
+            std::borrow::Cow::Borrowed(worker_status_label(receipt.status))
+        };
+        let activity = receipt.activity.clone().unwrap_or_default();
+        retained.push(WorkRow {
+            id,
+            mark: receipt.state.glyph(),
+            label: receipt.display_name.clone(),
+            detail: if activity.is_empty() {
+                status.to_string()
+            } else {
+                format!("{status} · {activity}")
+            },
+            tone: bucket_tone(if parked {
+                WorkBucket::Ready
+            } else {
+                worker_status_bucket(receipt.status)
+            }),
+            selectable: true,
+            primary_action: Some(SidebarRowAction::OpenAgentTranscript {
+                agent_id: receipt.worker_id.clone(),
+            }),
+            agent: Some(AgentRowFacts {
+                role_label: receipt.display_name.clone(),
+                status: status.to_string(),
+                objective: activity,
+                elapsed_secs: receipt.millis.map(|millis| millis / 1_000),
+                model: (!receipt.model.is_empty() && receipt.model != "unknown")
+                    .then(|| receipt.model.clone()),
+                tokens: receipt.output_tokens,
+                todos_remaining: None,
+            }),
+        });
+    }
+    retained.extend(agents);
+    let agents = retained;
     let mut out = Vec::with_capacity(agents.len() + 1);
     if !agents.is_empty() {
         out.push(agents_section_heading(&format!(
@@ -2868,6 +2921,154 @@ mod tests {
         }
     }
 
+    fn retained_agent_receipt(
+        id: &str,
+        status: AgentWorkerStatus,
+        state: crate::tui::agent_roster::RosterState,
+    ) -> crate::tui::agent_roster::AgentRosterRow {
+        crate::tui::agent_roster::AgentRosterRow {
+            worker_id: id.to_string(),
+            display_name: format!("retained {id}"),
+            model: "test-model".to_string(),
+            state,
+            status,
+            activity: None,
+            millis: Some(3_000),
+            input_tokens: None,
+            output_tokens: None,
+            cost_microusd: None,
+            steps_taken: 3,
+            parent_run_id: None,
+            run_id: id.to_string(),
+        }
+    }
+
+    #[test]
+    fn agents_register_retains_expired_receipts_alongside_fresh_live_rows() {
+        use crate::tui::agent_roster::RosterState;
+        use crate::tui::subagent_routing::reconcile_subagent_activity_state_at;
+        use std::time::Duration;
+
+        let mut app = test_app();
+        app.current_session_id = Some("roster-owner".to_string());
+        app.agent_roster_session_id = app.current_session_id.clone();
+        let mut done = running_agent("done");
+        done.status = SubAgentStatus::Completed;
+        app.subagent_cache = vec![done, running_agent("live")];
+        let receipt =
+            retained_agent_receipt("done", AgentWorkerStatus::Completed, RosterState::Done);
+        app.agent_roster = vec![
+            receipt.clone(),
+            receipt,
+            // A newer live cache row must win over an older retained receipt.
+            retained_agent_receipt("live", AgentWorkerStatus::Completed, RosterState::Done),
+        ];
+        app.agent_progress
+            .insert("done".to_string(), "old progress".to_string());
+        let observed = Instant::now();
+        reconcile_subagent_activity_state_at(&mut app, observed);
+        reconcile_subagent_activity_state_at(&mut app, observed + Duration::from_secs(46));
+        assert!(
+            !app.subagent_cache
+                .iter()
+                .any(|agent| agent.agent_id == "done")
+        );
+        assert!(!app.agent_progress.contains_key("done"));
+
+        let rows = agents_view_rows(&mut app);
+        let workers: Vec<_> = rows.iter().filter(|row| row.agent.is_some()).collect();
+        assert_eq!(workers.len(), 2, "one row per retained/live worker ID");
+        let done = workers
+            .iter()
+            .find(|row| row.id.0 == "worker:done")
+            .unwrap();
+        assert_eq!(done.agent.as_ref().unwrap().status, "completed");
+        assert_eq!(done.agent.as_ref().unwrap().elapsed_secs, Some(3));
+        assert_eq!(done.agent.as_ref().unwrap().tokens, None);
+        let live = workers
+            .iter()
+            .find(|row| row.id.0 == "worker:live")
+            .unwrap();
+        assert_eq!(live.agent.as_ref().unwrap().status, "running");
+
+        app.subagent_cache[0].status = SubAgentStatus::Completed;
+        reconcile_subagent_activity_state_at(&mut app, observed + Duration::from_secs(47));
+        reconcile_subagent_activity_state_at(&mut app, observed + Duration::from_secs(93));
+        assert!(app.subagent_cache.is_empty());
+        assert!(app.agent_progress.is_empty());
+        let rows = agents_view_rows(&mut app);
+        assert_eq!(rows.iter().filter(|row| row.agent.is_some()).count(), 2);
+    }
+
+    #[test]
+    fn retained_agents_register_is_bound_to_the_exact_parent_session() {
+        use crate::tui::agent_roster::RosterState;
+
+        let mut app = test_app();
+        app.agent_roster_session_id = Some("original-session".to_string());
+        app.agent_roster = vec![retained_agent_receipt(
+            "private-worker",
+            AgentWorkerStatus::Completed,
+            RosterState::Done,
+        )];
+        app.agent_roster[0].cost_microusd = Some(1_000);
+        for owner in [None, Some(""), Some("other-session")] {
+            app.current_session_id = owner.map(str::to_string);
+            assert!(app.current_agent_roster().is_empty());
+            assert!(
+                !agents_view_rows(&mut app)
+                    .iter()
+                    .any(|row| row.agent.is_some())
+            );
+            assert!(
+                !super::super::views::price_rows(&mut app)
+                    .iter()
+                    .any(|row| row.id.0.starts_with("price:agent:"))
+            );
+        }
+        app.current_session_id = Some("original-session".to_string());
+        assert_eq!(app.current_agent_roster().len(), 1);
+        assert_eq!(
+            agents_view_rows(&mut app)
+                .iter()
+                .filter(|row| row.agent.is_some())
+                .count(),
+            1
+        );
+    }
+
+    #[test]
+    fn retained_agents_register_distinguishes_parked_from_waiting_for_input() {
+        use crate::tui::agent_roster::RosterState;
+
+        let mut app = test_app();
+        app.current_session_id = Some("roster-owner".to_string());
+        app.agent_roster_session_id = app.current_session_id.clone();
+        app.agent_roster = vec![
+            retained_agent_receipt(
+                "parked",
+                AgentWorkerStatus::WaitingForUser,
+                RosterState::Parked,
+            ),
+            retained_agent_receipt(
+                "asked",
+                AgentWorkerStatus::WaitingForUser,
+                RosterState::Waiting,
+            ),
+        ];
+        let rows = agents_view_rows(&mut app);
+        let parked = rows.iter().find(|row| row.id.0 == "worker:parked").unwrap();
+        let asked = rows.iter().find(|row| row.id.0 == "worker:asked").unwrap();
+        assert_eq!(parked.mark, RosterState::Parked.glyph());
+        assert_eq!(parked.tone, WorkTone::Muted);
+        assert_ne!(parked.agent.as_ref().unwrap().status, "waiting for input");
+        assert_eq!(asked.agent.as_ref().unwrap().status, "waiting for input");
+        assert_eq!(asked.tone, WorkTone::Attention);
+        assert!(
+            matches!(&parked.primary_action, Some(SidebarRowAction::OpenAgentTranscript { agent_id }) if agent_id == "parked")
+        );
+    }
+
     /// #5287: the identity column leads with the name the lane was dispatched
     /// under; the whale only names an agent that was dispatched without one.
     #[test]

crates/tui/src/tui/work_surface/views.rs

@@ -200,7 +200,7 @@ pub(super) fn price_rows(app: &mut App) -> Vec<WorkRow> {
         primary_action: Some(SidebarRowAction::Command("/cost".to_string())),
         agent: None,
     });
-    let roster = app.agent_roster.clone();
+    let roster = app.current_agent_roster().to_vec();
     let priced: Vec<_> = roster
         .iter()
         .filter(|row| row.cost_microusd.is_some())