fix: distinguish agent completion from startup and session changes (#4457)

refs #3338
refs #3990
refs #3916

20 files changed, 1050 insertions, 317 deletions

diff --git a/src/api/schema/agents.rs b/src/api/schema/agents.rs
index ee496d5d..48450c30 100644
--- a/src/api/schema/agents.rs
+++ b/src/api/schema/agents.rs
@@ -218,6 +218,9 @@ pub struct AgentInfo {
     pub interactive_ready: bool,
     #[serde(default)]
     pub state_change_seq: u64,
+    /// The current idle transition completed work, independently of who has viewed it.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub completion_seq: Option<u64>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub cwd: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
diff --git a/src/app/actions.rs b/src/app/actions.rs
index 85314331..953bcd94 100644
--- a/src/app/actions.rs
+++ b/src/app/actions.rs
@@ -25,12 +25,7 @@ fn is_background_completion_transition(prev_state: AgentState, new_state: AgentS
 }
 
 fn is_completion_transition(change: &EffectiveStateChange) -> bool {
-    is_completion_transition_parts(
-        change.previous_state,
-        change.state,
-        change.previous_agent_label.as_deref(),
-        change.agent_label.as_deref(),
-    )
+    is_background_completion_transition(change.previous_state, change.state)
 }
 
 fn public_tab_id_for_index(ws: &crate::workspace::Workspace, tab_idx: usize) -> Option<String> {
@@ -40,19 +35,6 @@ fn public_tab_id_for_index(ws: &crate::workspace::Workspace, tab_idx: usize) ->
     ))
 }
 
-pub fn is_completion_transition_parts(
-    previous_state: AgentState,
-    state: AgentState,
-    previous_agent_label: Option<&str>,
-    agent_label: Option<&str>,
-) -> bool {
-    is_background_completion_transition(previous_state, state)
-        || (previous_state == AgentState::Unknown
-            && state == AgentState::Idle
-            && previous_agent_label.is_some()
-            && previous_agent_label == agent_label)
-}
-
 pub fn active_tab_suppresses_notifications(
     is_active_tab: bool,
     outer_terminal_focus: Option<bool>,
@@ -60,7 +42,6 @@ pub fn active_tab_suppresses_notifications(
     is_active_tab && outer_terminal_focus != Some(false)
 }
 
-#[cfg(test)]
 pub fn notification_sound_for_state_change(
     suppress_active_tab_notifications: bool,
     prev_state: AgentState,
@@ -82,58 +63,21 @@ pub fn notification_sound_for_state_change(
     }
 }
 
-pub fn notification_sound_for_state_change_with_agent_labels(
-    suppress_active_tab_notifications: bool,
-    prev_state: AgentState,
-    new_state: AgentState,
-    previous_agent_label: Option<&str>,
-    agent_label: Option<&str>,
-) -> Option<crate::sound::Sound> {
-    if new_state == prev_state {
-        return None;
-    }
-
-    match new_state {
-        AgentState::Blocked => Some(crate::sound::Sound::Request),
-        AgentState::Idle
-            if is_completion_transition_parts(
-                prev_state,
-                new_state,
-                previous_agent_label,
-                agent_label,
-            ) && !suppress_active_tab_notifications =>
-        {
-            Some(crate::sound::Sound::Done)
-        }
-        _ => None,
-    }
-}
-
 fn notification_sound_for_effective_state_change(
     suppress_active_tab_notifications: bool,
     change: &EffectiveStateChange,
 ) -> Option<crate::sound::Sound> {
-    if change.state == change.previous_state {
-        return None;
-    }
-
-    match change.state {
-        AgentState::Blocked => Some(crate::sound::Sound::Request),
-        AgentState::Idle
-            if is_completion_transition(change) && !suppress_active_tab_notifications =>
-        {
-            Some(crate::sound::Sound::Done)
-        }
-        _ => None,
-    }
+    notification_sound_for_state_change(
+        suppress_active_tab_notifications,
+        change.previous_state,
+        change.state,
+    )
 }
 
-pub fn notification_toast_for_state_change_with_agent_labels(
+pub fn notification_toast_for_state_change(
     suppress_active_tab_notifications: bool,
     prev_state: AgentState,
     new_state: AgentState,
-    previous_agent_label: Option<&str>,
-    agent_label: Option<&str>,
 ) -> Option<ToastKind> {
     if suppress_active_tab_notifications || new_state == prev_state {
         return None;
@@ -141,14 +85,7 @@ pub fn notification_toast_for_state_change_with_agent_labels(
 
     match new_state {
         AgentState::Blocked => Some(ToastKind::NeedsAttention),
-        AgentState::Idle
-            if is_completion_transition_parts(
-                prev_state,
-                new_state,
-                previous_agent_label,
-                agent_label,
-            ) =>
-        {
+        AgentState::Idle if is_background_completion_transition(prev_state, new_state) => {
             Some(ToastKind::Finished)
         }
         _ => None,
@@ -159,15 +96,11 @@ fn notification_toast_for_effective_state_change(
     suppress_active_tab_notifications: bool,
     change: &EffectiveStateChange,
 ) -> Option<ToastKind> {
-    if suppress_active_tab_notifications || change.state == change.previous_state {
-        return None;
-    }
-
-    match change.state {
-        AgentState::Blocked => Some(ToastKind::NeedsAttention),
-        AgentState::Idle if is_completion_transition(change) => Some(ToastKind::Finished),
-        _ => None,
-    }
+    notification_toast_for_state_change(
+        suppress_active_tab_notifications,
+        change.previous_state,
+        change.state,
+    )
 }
 
 pub fn notification_toast_for_pane_state_update(
@@ -181,12 +114,10 @@ pub fn notification_toast_for_pane_state_update(
         return None;
     }
 
-    notification_toast_for_state_change_with_agent_labels(
+    notification_toast_for_state_change(
         suppress_active_tab_notifications,
         update.previous_state,
         update.state,
-        update.previous_agent_label.as_deref(),
-        update.agent_label.as_deref(),
     )
 }
 
@@ -1709,39 +1640,55 @@ impl AppState {
             managed_changed,
             agent_name_changed,
             unchanged_change,
-            managed_launch_pending,
             suppress_acquisition_completion,
+            completion_reset,
         ) = {
             let terminal = self.terminals.get_mut(&terminal_id)?;
             let previous_agent_name = terminal.agent_name.clone();
-            let managed_launch_pending = terminal.managed_agent_launch_pending();
+            let had_completion = terminal.last_agent_completion_seq.is_some() || !previous_seen;
             let mutation = update(terminal)?;
+            let completion_reset = mutation.session_ref_changed
+                || mutation
+                    .effective_state_change
+                    .as_ref()
+                    .is_some_and(|change| change.previous_agent_label != change.agent_label);
+            if completion_reset {
+                terminal.last_agent_completion_seq = None;
+            }
             let managed_changed = terminal.reconcile_managed_agent_at(now, false);
             let suppress_acquisition_completion = terminal.finish_agent_process_acquisition();
             let agent_name_changed = terminal.agent_name != previous_agent_name;
-            let unchanged_change = (mutation.agent_released || agent_name_changed)
+            let unchanged_change = (mutation.agent_released
+                || agent_name_changed
+                || (completion_reset && had_completion))
                 .then(|| terminal.unchanged_effective_state_change_at(now));
             (
                 mutation,
                 managed_changed,
                 agent_name_changed,
                 unchanged_change,
-                managed_launch_pending,
                 suppress_acquisition_completion,
+                completion_reset,
             )
         };
+        if completion_reset {
+            self.pending_agent_notifications.remove(&pane_id);
+            self.workspaces[ws_idx].pane_state_mut(pane_id)?.seen = true;
+        }
         if mutation.session_ref_changed || managed_changed || agent_name_changed {
             self.mark_session_dirty();
         }
         let agent_released = mutation.agent_released;
         let change = mutation.effective_state_change.or(unchanged_change)?;
         let suppress_completion = force_suppress_completion
-            || (change.state == AgentState::Idle
-                && (managed_launch_pending || suppress_acquisition_completion));
+            || (change.state == AgentState::Idle && suppress_acquisition_completion);
         if change.previous_state != change.state {
             self.next_agent_state_change_seq += 1;
             if let Some(terminal) = self.terminals.get_mut(&terminal_id) {
                 terminal.last_agent_state_change_seq = Some(self.next_agent_state_change_seq);
+                terminal.last_agent_completion_seq = (!suppress_completion
+                    && is_completion_transition(&change))
+                .then_some(self.next_agent_state_change_seq);
             }
         }
         let seen = self.apply_pane_state_change(ws_idx, pane_id, &change, suppress_completion)?;
@@ -3054,34 +3001,217 @@ mod tests {
         assert!(pane.seen);
     }
 
+    fn assert_completion_guard_sequence(
+        acquired: bool,
+        states: &[AgentState],
+        expect_completion: bool,
+    ) {
+        let mut app = app_with_workspaces(&["active", "background"]);
+        app.active = Some(0);
+        app.toast_config.delivery = crate::config::ToastDelivery::Herdr;
+        app.toast_config.delay_seconds = 0;
+        let pane_id = app.workspaces[1].tabs[0].root_pane;
+        if acquired {
+            app.handle_app_event(AppEvent::AgentProcessDetected {
+                pane_id,
+                agent: Agent::Pi,
+                observed_at: Instant::now(),
+            });
+        }
+        for &state in states {
+            app.handle_app_event(AppEvent::StateChanged {
+                pane_id,
+                agent: Some(Agent::Pi),
+                state,
+                visible_blocker: state == AgentState::Blocked,
+                visible_working: state == AgentState::Working,
+                process_exited: false,
+                observed_at: Instant::now(),
+            });
+        }
+        let finished = app
+            .toast
+            .as_ref()
+            .is_some_and(|toast| toast.kind == ToastKind::Finished);
+        assert_eq!(
+            finished, expect_completion,
+            "completion notification: {states:?}"
+        );
+        assert_eq!(
+            !app.workspaces[1].panes[&pane_id].seen, expect_completion,
+            "unseen completion: {states:?}"
+        );
+        let terminal = &app.terminals[&app.workspaces[1].panes[&pane_id].attached_terminal_id];
+        assert_eq!(
+            terminal.last_agent_completion_seq.is_some(),
+            expect_completion
+        );
+        if expect_completion {
+            assert_eq!(
+                terminal.last_agent_completion_seq,
+                terminal.last_agent_state_change_seq
+            );
+        }
+        app.assert_invariants_for_test();
+    }
+
     #[test]
-    fn idle_after_known_unknown_agent_in_background_marks_done() {
-        let mut state = app_with_workspaces(&["active", "background"]);
-        state.toast_config.delivery = crate::config::ToastDelivery::Herdr;
-        state.active = Some(0);
-        let bg_pane_id = *state.workspaces[1].panes.keys().next().unwrap();
+    fn completion_guard_unknown_to_idle_is_not_completed_work() {
+        assert_completion_guard_sequence(false, &[AgentState::Unknown, AgentState::Idle], false);
+    }
 
-        state.handle_app_event(AppEvent::StateChanged {
-            pane_id: bg_pane_id,
-            agent: Some(Agent::Pi),
-            state: AgentState::Unknown,
-            visible_blocker: false,
-            visible_working: false,
-            process_exited: false,
-            observed_at: std::time::Instant::now(),
-        });
-        state.handle_app_event(AppEvent::StateChanged {
-            pane_id: bg_pane_id,
-            agent: Some(Agent::Pi),
-            state: AgentState::Idle,
-            visible_blocker: false,
-            visible_working: false,
-            process_exited: false,
-            observed_at: std::time::Instant::now(),
-        });
+    #[test]
+    fn completion_guard_first_work_finishes_without_prior_idle() {
+        assert_completion_guard_sequence(true, &[AgentState::Working, AgentState::Idle], true);
+    }
 
-        let pane = state.workspaces[1].panes.get(&bg_pane_id).unwrap();
-        assert!(!pane.seen);
+    #[test]
+    fn completion_guard_first_work_can_pause_for_permission() {
+        assert_completion_guard_sequence(
+            true,
+            &[AgentState::Working, AgentState::Blocked, AgentState::Idle],
+            true,
+        );
+    }
+
+    #[test]
+    fn completion_guard_startup_trust_is_not_completed_work() {
+        assert_completion_guard_sequence(true, &[AgentState::Blocked, AgentState::Idle], false);
+    }
+
+    #[test]
+    fn completion_guard_managed_launch_readiness_does_not_swallow_work() {
+        for first_state in [AgentState::Blocked, AgentState::Working] {
+            let mut app = app_with_workspaces(&["active", "background"]);
+            app.active = Some(0);
+            let pane_id = app.workspaces[1].tabs[0].root_pane;
+            let terminal_id = app.workspaces[1].panes[&pane_id]
+                .attached_terminal_id
+                .clone();
+            app.terminals
+                .get_mut(&terminal_id)
+                .unwrap()
+                .begin_managed_agent(
+                    "worker".into(),
+                    Agent::Pi,
+                    Instant::now(),
+                    std::time::Duration::ZERO,
+                    std::time::Duration::from_secs(60),
+                );
+            for state in [first_state, AgentState::Idle] {
+                app.handle_app_event(AppEvent::StateChanged {
+                    pane_id,
+                    agent: Some(Agent::Pi),
+                    state,
+                    visible_blocker: state == AgentState::Blocked,
+                    visible_working: state == AgentState::Working,
+                    process_exited: false,
+                    observed_at: Instant::now(),
+                });
+            }
+            let terminal = &app.terminals[&terminal_id];
+            assert!(terminal.managed_agent_interactive_ready());
+            assert_eq!(
+                terminal.last_agent_completion_seq.is_some(),
+                first_state == AgentState::Working
+            );
+            assert_eq!(
+                !app.workspaces[1].panes[&pane_id].seen,
+                first_state == AgentState::Working
+            );
+        }
+    }
+
+    #[test]
+    fn completion_guard_same_state_agent_replacement_clears_old_work() {
+        let mut app = app_with_workspaces(&["active", "background"]);
+        app.active = Some(0);
+        app.toast_config.delay_seconds = 5;
+        let pane_id = app.workspaces[1].tabs[0].root_pane;
+        for (seq, label, state) in [
+            (1, "old", AgentState::Working),
+            (2, "old", AgentState::Idle),
+            (3, "new", AgentState::Idle),
+        ] {
+            app.handle_app_event(AppEvent::HookStateReported {
+                pane_id,
+                source: "custom:worker".into(),
+                agent_label: label.into(),
+                state,
+                message: None,
+                seq: Some(seq),
+                session_ref: None,
+            });
+            if seq == 2 {
+                assert!(!app.workspaces[1].panes[&pane_id].seen);
+                assert!(app.pending_agent_notifications.contains_key(&pane_id));
+            }
+        }
+        let terminal = &app.terminals[&app.workspaces[1].panes[&pane_id].attached_terminal_id];
+        assert_eq!(terminal.effective_agent_label(), Some("new"));
+        assert!(terminal.last_agent_completion_seq.is_none());
+        assert!(app.workspaces[1].panes[&pane_id].seen);
+        assert!(!app.pending_agent_notifications.contains_key(&pane_id));
+    }
+
+    #[test]
+    fn completion_guard_idle_session_replacement_clears_seen_and_pending_delivery() {
+        let mut app = app_with_workspaces(&["active", "background"]);
+        app.active = Some(0);
+        app.toast_config.delay_seconds = 5;
+        let pane_id = app.workspaces[1].tabs[0].root_pane;
+        for (seq, session, reason) in [(1, "old-session", "startup"), (2, "new-session", "clear")] {
+            let updates = app.handle_app_event(AppEvent::AgentSessionReported {
+                pane_id,
+                source: "herdr:claude".into(),
+                agent_label: "claude".into(),
+                seq: Some(seq),
+                session_ref: crate::agent_resume::AgentSessionRef::id(session),
+                session_start_source: Some(reason.into()),
+            });
+            if seq == 1 {
+                for state in [AgentState::Working, AgentState::Idle] {
+                    app.handle_app_event(AppEvent::StateChanged {
+                        pane_id,
+                        agent: Some(Agent::Claude),
+                        state,
+                        visible_blocker: false,
+                        visible_working: state == AgentState::Working,
+                        process_exited: false,
+                        observed_at: Instant::now(),
+                    });
+                }
+                assert!(!app.workspaces[1].panes[&pane_id].seen);
+                assert!(app.pending_agent_notifications.contains_key(&pane_id));
+            } else {
+                assert!(
+                    !updates.is_empty(),
+                    "session replacement must publish attention reset"
+                );
+            }
+        }
+        let terminal = &app.terminals[&app.workspaces[1].panes[&pane_id].attached_terminal_id];
+        assert_eq!(
+            terminal
+                .persisted_agent_session
+                .as_ref()
+                .unwrap()
+                .session_ref
+                .value,
+            "new-session"
+        );
+        assert!(terminal.last_agent_completion_seq.is_none());
+        assert!(app.workspaces[1].panes[&pane_id].seen);
+        assert!(!app.pending_agent_notifications.contains_key(&pane_id));
+    }
+
+    #[test]
+    fn completion_guard_normal_turn_still_finishes() {
+        assert_completion_guard_sequence(
+            true,
+            &[AgentState::Idle, AgentState::Working, AgentState::Idle],
+            true,
+        );
     }
 
     #[test]
@@ -3139,9 +3269,9 @@ mod tests {
             .pop()
             .expect("idle state update");
 
-        assert!(update.suppress_completion);
-        assert!(state.workspaces[1].panes[&pane_id].seen);
-        assert!(!matches!(
+        assert!(!update.suppress_completion);
+        assert!(!state.workspaces[1].panes[&pane_id].seen);
+        assert!(matches!(
             state.toast.as_ref().map(|toast| toast.kind),
             Some(ToastKind::Finished)
         ));
diff --git a/src/app/agents.rs b/src/app/agents.rs
index 5033505d..fd2a58ed 100644
--- a/src/app/agents.rs
+++ b/src/app/agents.rs
@@ -395,6 +395,7 @@ impl App {
             launch_pending: terminal.managed_agent_launch_pending(),
             interactive_ready: terminal.managed_agent_interactive_ready(),
             state_change_seq: terminal.last_agent_state_change_seq.unwrap_or(0),
+            completion_seq: terminal.last_agent_completion_seq,
             cwd: pane.cwd,
             foreground_cwd: pane.foreground_cwd,
             revision: pane.revision,
diff --git a/src/client/endpoint/control.rs b/src/client/endpoint/control.rs
index 6eef8dc8..f2fe1b76 100644
--- a/src/client/endpoint/control.rs
+++ b/src/client/endpoint/control.rs
@@ -9,6 +9,7 @@ pub(crate) struct DecodedAgentViewProjection {
 pub(crate) enum EndpointControlMessage {
     HealthPong,
     AgentViewProjection(DecodedAgentViewProjection),
+    AgentCompletions(crate::protocol::endpoint::EndpointAgentCompletions),
     Snapshot(Box<crate::protocol::ClientShellSnapshot>),
     Ignored,
 }
@@ -20,6 +21,11 @@ pub(crate) fn decode_endpoint_control(
     if kind == crate::protocol::endpoint::HEALTH_PONG_KIND {
         return Ok(EndpointControlMessage::HealthPong);
     }
+    if kind == crate::protocol::endpoint::AGENT_COMPLETIONS_KIND {
+        return Ok(serde_json::from_str(data)
+            .map(EndpointControlMessage::AgentCompletions)
+            .unwrap_or(EndpointControlMessage::Ignored));
+    }
     if kind == crate::protocol::endpoint::AGENT_VIEW_PROJECTION_KIND {
         let Ok(projection): Result<crate::protocol::endpoint::EndpointAgentViewProjection, _> =
             serde_json::from_str(data)
@@ -78,6 +84,30 @@ mod tests {
     }
 
     #[test]
+    fn completion_guard_optional_control_round_trips_without_changing_snapshot_codec() {
+        let projection = crate::protocol::endpoint::EndpointAgentCompletions {
+            boot_id: "boot".into(),
+            revision: 3,
+            completions: [("pane".into(), 7)].into_iter().collect(),
+        };
+        let crate::protocol::ServerMessage::EndpointControl { kind, data } =
+            crate::protocol::endpoint::agent_completions_message(&projection).unwrap()
+        else {
+            panic!("expected optional control");
+        };
+        let EndpointControlMessage::AgentCompletions(decoded) =
+            decode_endpoint_control(&kind, &data).unwrap()
+        else {
+            panic!("expected completion projection");
+        };
+        assert_eq!(decoded, projection);
+        assert!(matches!(
+            decode_endpoint_control(&kind, "invalid").unwrap(),
+            EndpointControlMessage::Ignored
+        ));
+    }
+
+    #[test]
     fn agent_view_projection_decodes_and_validates_the_view() {
         let view = crate::api::schema::AgentViewSetParams {
             source: "example.views".into(),
diff --git a/src/client/mod.rs b/src/client/mod.rs
index b9335e06..68e569f9 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -1913,6 +1913,16 @@ async fn run_client_loop(
                                 }
                                 continue;
                             }
+                            Ok(endpoint::EndpointControlMessage::AgentCompletions(projection)) => {
+                                if let Some(shell) = state.shell.as_mut() {
+                                    shell.set_endpoint_agent_completions(
+                                        &endpoint_id,
+                                        generation,
+                                        projection,
+                                    );
+                                }
+                                continue;
+                            }
                             Ok(endpoint::EndpointControlMessage::Ignored) => {
                                 debug!(%kind, "ignoring unknown endpoint control message");
                                 continue;
diff --git a/src/client/shell/endpoint_agent_state.rs b/src/client/shell/endpoint_agent_state.rs
index f8ba7a9b..1ebcd35d 100644
--- a/src/client/shell/endpoint_agent_state.rs
+++ b/src/client/shell/endpoint_agent_state.rs
@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use crate::api::schema::AgentStatus;
 use crate::protocol::{ClientShellAgent, ClientShellSnapshot, PaneSurfaceFrame};
@@ -7,13 +7,49 @@ use crate::protocol::{ClientShellAgent, ClientShellSnapshot, PaneSurfaceFrame};
 pub(super) struct EndpointAgentPresentation {
     boot_id: Option<String>,
     acknowledged: HashMap<String, u64>,
+    completed: HashMap<String, u64>,
+    working: HashSet<String>,
+    pending_completions: Option<(
+        Option<u64>,
+        crate::protocol::endpoint::EndpointAgentCompletions,
+    )>,
 }
 
 impl EndpointAgentPresentation {
+    pub(super) fn receive_completions(
+        &mut self,
+        generation: Option<u64>,
+        completions: crate::protocol::endpoint::EndpointAgentCompletions,
+    ) {
+        if self
+            .pending_completions
+            .as_ref()
+            .is_some_and(|(current_generation, current)| {
+                *current_generation == generation
+                    && current.boot_id == completions.boot_id
+                    && current.revision >= completions.revision
+            })
+        {
+            return;
+        }
+        self.pending_completions = Some((generation, completions));
+    }
+
+    #[cfg(test)]
     pub(super) fn project_snapshot(&mut self, snapshot: &mut ClientShellSnapshot) {
+        self.project_snapshot_for_generation(snapshot, None);
+    }
+
+    pub(super) fn project_snapshot_for_generation(
+        &mut self,
+        snapshot: &mut ClientShellSnapshot,
+        generation: Option<u64>,
+    ) {
         if self.boot_id.as_deref() != Some(snapshot.boot_id.as_str()) {
             self.boot_id = Some(snapshot.boot_id.clone());
             self.acknowledged.clear();
+            self.completed.clear();
+            self.working.clear();
             self.acknowledged.extend(
                 snapshot
                     .agents
@@ -21,13 +57,59 @@ impl EndpointAgentPresentation {
                     .map(|agent| (agent.pane_id.clone(), agent.state_change_seq)),
             );
         }
-        self.acknowledged.retain(|pane_id, _| {
-            snapshot
-                .agents
-                .iter()
-                .any(|agent| &agent.pane_id == pane_id)
-        });
+        let pane_ids: HashSet<&str> = snapshot
+            .agents
+            .iter()
+            .map(|agent| agent.pane_id.as_str())
+            .collect();
+        self.acknowledged
+            .retain(|pane_id, _| pane_ids.contains(pane_id.as_str()));
+        self.completed
+            .retain(|pane_id, _| pane_ids.contains(pane_id.as_str()));
+        self.working
+            .retain(|pane_id| pane_ids.contains(pane_id.as_str()));
+        let completions = self
+            .pending_completions
+            .take()
+            .filter(|(received_generation, projection)| {
+                *received_generation == generation
+                    && projection.boot_id == snapshot.boot_id
+                    && projection.revision == snapshot.revision
+            })
+            .map(|(_, projection)| projection.completions);
         for agent in &mut snapshot.agents {
+            match agent.agent_status {
+                AgentStatus::Working => {
+                    self.working.insert(agent.pane_id.clone());
+                    self.completed.remove(&agent.pane_id);
+                }
+                AgentStatus::Blocked => {
+                    self.completed.remove(&agent.pane_id);
+                }
+                AgentStatus::Idle | AgentStatus::Done => {
+                    let observed_work = self.working.remove(&agent.pane_id);
+                    let completed = completions.as_ref().map_or_else(
+                        || {
+                            observed_work
+                                || self.completed.get(&agent.pane_id)
+                                    == Some(&agent.state_change_seq)
+                        },
+                        |completions| {
+                            completions.get(&agent.pane_id) == Some(&agent.state_change_seq)
+                        },
+                    );
+                    if completed {
+                        self.completed
+                            .insert(agent.pane_id.clone(), agent.state_change_seq);
+                    } else {
+                        self.completed.remove(&agent.pane_id);
+                    }
+                }
+                _ => {
+                    self.working.remove(&agent.pane_id);
+                    self.completed.remove(&agent.pane_id);
+                }
+            }
             agent.agent_status = self.projected_status(agent);
         }
         project_aggregate_status(snapshot);
@@ -72,9 +154,11 @@ impl EndpointAgentPresentation {
     }
 
     pub(super) fn seen(&self, agent: &ClientShellAgent) -> bool {
-        self.acknowledged
-            .get(&agent.pane_id)
-            .is_some_and(|sequence| *sequence >= agent.state_change_seq)
+        self.completed.get(&agent.pane_id).is_none_or(|completion| {
+            self.acknowledged
+                .get(&agent.pane_id)
+                .is_some_and(|sequence| sequence >= completion)
+        })
     }
 
     fn projected_status(&self, agent: &ClientShellAgent) -> AgentStatus {
@@ -119,7 +203,9 @@ fn project_aggregate_status(snapshot: &mut ClientShellSnapshot) {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::protocol::{FrameData, PaneSurfacePane, SurfaceRect};
+    use crate::protocol::{
+        endpoint::EndpointAgentCompletions, FrameData, PaneSurfacePane, SurfaceRect,
+    };
 
     fn agent(status: AgentStatus, sequence: u64) -> ClientShellAgent {
         ClientShellAgent {
@@ -197,6 +283,133 @@ mod tests {
         assert_eq!(snapshot.agents[0].agent_status, AgentStatus::Idle);
     }
 
+    fn assert_idle_sequence(states: &[(AgentStatus, u64)]) {
+        let mut presentation = EndpointAgentPresentation::default();
+        let mut projected = AgentStatus::Unknown;
+        for (revision, &(status, seq)) in states.iter().enumerate() {
+            let mut snapshot = snapshot(status, seq, revision as u64 + 1);
+            presentation.project_snapshot(&mut snapshot);
+            projected = snapshot.agents[0].agent_status;
+        }
+        assert_eq!(projected, AgentStatus::Idle, "{states:?}");
+    }
+
+    fn completions(boot: &str, revision: u64, seq: Option<u64>) -> EndpointAgentCompletions {
+        EndpointAgentCompletions {
+            boot_id: boot.into(),
+            revision,
+            completions: seq
+                .map(|seq| ("agent-pane".into(), seq))
+                .into_iter()
+                .collect(),
+        }
+    }
+
+    #[test]
+    fn completion_guard_initial_unknown_to_idle_does_not_project_done() {
+        assert_idle_sequence(&[(AgentStatus::Unknown, 0), (AgentStatus::Idle, 1)]);
+    }
+
+    #[test]
+    fn completion_guard_new_idle_agent_is_not_completed_work() {
+        let mut presentation = EndpointAgentPresentation::default();
+        let mut baseline = snapshot(AgentStatus::Idle, 0, 1);
+        baseline.agents.clear();
+        presentation.project_snapshot(&mut baseline);
+        let mut ready = snapshot(AgentStatus::Idle, 1, 2);
+
+        presentation.project_snapshot(&mut ready);
+
+        assert_eq!(ready.agents[0].agent_status, AgentStatus::Idle);
+    }
+
+    #[test]
+    fn completion_guard_session_rebind_does_not_project_done() {
+        assert_idle_sequence(&[
+            (AgentStatus::Idle, 4),
+            (AgentStatus::Unknown, 5),
+            (AgentStatus::Idle, 6),
+        ]);
+    }
+
+    #[test]
+    fn completion_guard_startup_blocker_does_not_project_done() {
+        assert_idle_sequence(&[
+            (AgentStatus::Unknown, 0),
+            (AgentStatus::Blocked, 1),
+            (AgentStatus::Idle, 2),
+        ]);
+    }
+
+    #[test]
+    fn completion_guard_distinguishes_coalesced_work_from_rebind() {
+        for completion in [None, Some(6)] {
+            let mut presentation = EndpointAgentPresentation::default();
+            let mut baseline = snapshot(AgentStatus::Idle, 4, 1);
+            presentation.project_snapshot(&mut baseline);
+            presentation.receive_completions(None, completions(&baseline.boot_id, 2, completion));
+            let mut settled = snapshot(AgentStatus::Idle, 6, 2);
+
+            presentation.project_snapshot(&mut settled);
+
+            let expected = if completion.is_some() {
+                AgentStatus::Done
+            } else {
+                AgentStatus::Idle
+            };
+            assert_eq!(settled.agents[0].agent_status, expected);
+            presentation.project_snapshot(&mut settled);
+            assert_eq!(settled.agents[0].agent_status, expected);
+        }
+    }
+
+    #[test]
+    fn completion_guard_server_suppression_overrides_client_observed_work() {
+        let mut presentation = EndpointAgentPresentation::default();
+        let mut baseline = snapshot(AgentStatus::Working, 1, 1);
+        presentation.project_snapshot(&mut baseline);
+        presentation.receive_completions(None, completions(&baseline.boot_id, 2, None));
+        let mut settled = snapshot(AgentStatus::Idle, 2, 2);
+
+        presentation.project_snapshot(&mut settled);
+
+        assert_eq!(settled.agents[0].agent_status, AgentStatus::Idle);
+    }
+
+    #[test]
+    fn completion_guard_companion_is_scoped_to_boot_revision_and_connection() {
+        for (boot, revision, generation, seq) in [
+            ("old-boot", 2, Some(1), 6),
+            ("endpoint-boot", 1, Some(1), 6),
+            ("endpoint-boot", 2, Some(0), 6),
+            ("endpoint-boot", 2, Some(1), 5),
+        ] {
+            let mut presentation = EndpointAgentPresentation::default();
+            let mut baseline = snapshot(AgentStatus::Idle, 4, 1);
+            presentation.project_snapshot_for_generation(&mut baseline, Some(1));
+            presentation.receive_completions(generation, completions(boot, revision, Some(seq)));
+            let mut settled = snapshot(AgentStatus::Idle, 6, 2);
+
+            presentation.project_snapshot_for_generation(&mut settled, Some(1));
+
+            assert_eq!(settled.agents[0].agent_status, AgentStatus::Idle);
+        }
+    }
+
+    #[test]
+    fn completion_guard_new_boot_baselines_existing_completions() {
+        let mut presentation = EndpointAgentPresentation::default();
+        let mut old = snapshot(AgentStatus::Working, 1, 1);
+        old.boot_id = "old-boot".into();
+        presentation.project_snapshot(&mut old);
+        presentation.receive_completions(None, completions("endpoint-boot", 1, Some(6)));
+        let mut restored = snapshot(AgentStatus::Idle, 6, 1);
+
+        presentation.project_snapshot(&mut restored);
+
+        assert_eq!(restored.agents[0].agent_status, AgentStatus::Idle);
+    }
+
     #[test]
     fn unpresented_working_completion_projects_done() {
         let mut presentation = EndpointAgentPresentation::default();
diff --git a/src/client/shell/endpoints.rs b/src/client/shell/endpoints.rs
index 513842ba..dd6fa3cb 100644
--- a/src/client/shell/endpoints.rs
+++ b/src/client/shell/endpoints.rs
@@ -309,6 +309,23 @@ impl ClientShellState {
             .map(|snapshot| (snapshot.boot_id.as_str(), snapshot.revision))
     }
 
+    pub(crate) fn set_endpoint_agent_completions(
+        &mut self,
+        endpoint_id: &ClientEndpointId,
+        generation: u64,
+        projection: crate::protocol::endpoint::EndpointAgentCompletions,
+    ) {
+        if let Some(endpoint) = self
+            .endpoints
+            .iter_mut()
+            .find(|endpoint| &endpoint.endpoint_id == endpoint_id)
+        {
+            endpoint
+                .agent_presentation
+                .receive_completions(Some(generation), projection);
+        }
+    }
+
     pub(crate) fn set_endpoint_agent_view_projection_for_generation(
         &mut self,
         endpoint_id: &ClientEndpointId,
@@ -546,7 +563,7 @@ impl ClientShellState {
         }
         self.endpoints[index]
             .agent_presentation
-            .project_snapshot(&mut snapshot);
+            .project_snapshot_for_generation(&mut snapshot, generation);
         let presented_surface = if acknowledge_surface && endpoint_id == &self.active_endpoint_id {
             self.pane_surface.as_ref()
         } else {
diff --git a/src/protocol/endpoint.rs b/src/protocol/endpoint.rs
index df83b242..2704bf65 100644
--- a/src/protocol/endpoint.rs
+++ b/src/protocol/endpoint.rs
@@ -29,6 +29,15 @@ pub const HEALTH_PING_KIND: &str = "endpoint.health.ping.v1";
 pub const HEALTH_PONG_KIND: &str = "endpoint.health.pong.v1";
 pub const AGENT_VIEW_PROJECTION_CAPABILITY: &str = "agent_view_projection";
 pub const AGENT_VIEW_PROJECTION_KIND: &str = "endpoint.agent-view.v1";
+pub const AGENT_COMPLETIONS_CAPABILITY: &str = "agent_completions";
+pub const AGENT_COMPLETIONS_KIND: &str = "endpoint.agent-completions.v1";
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct EndpointAgentCompletions {
+    pub boot_id: String,
+    pub revision: u64,
+    pub completions: std::collections::BTreeMap<String, u64>,
+}
 
 fn default_true() -> bool {
     true
@@ -101,6 +110,15 @@ pub fn snapshot_message(snapshot: &ClientShellSnapshot) -> serde_json::Result<Se
     })
 }
 
+pub fn agent_completions_message(
+    projection: &EndpointAgentCompletions,
+) -> serde_json::Result<ServerMessage> {
+    Ok(ServerMessage::EndpointControl {
+        kind: AGENT_COMPLETIONS_KIND.into(),
+        data: serde_json::to_string(projection)?,
+    })
+}
+
 pub fn agent_view_projection_message(
     boot_id: &str,
     revision: u64,
@@ -151,6 +169,7 @@ impl EndpointServerWelcome {
                 PRESENTATION_EFFECTS_FENCE_CAPABILITY.into(),
                 HEALTH_CHECK_CAPABILITY.into(),
                 AGENT_VIEW_PROJECTION_CAPABILITY.into(),
+                AGENT_COMPLETIONS_CAPABILITY.into(),
             ],
             error: None,
         }
@@ -357,6 +376,7 @@ mod tests {
                 PRESENTATION_EFFECTS_FENCE_CAPABILITY.to_string(),
                 HEALTH_CHECK_CAPABILITY.to_string(),
                 AGENT_VIEW_PROJECTION_CAPABILITY.to_string(),
+                AGENT_COMPLETIONS_CAPABILITY.to_string(),
             ]
         );
     }
diff --git a/src/server/client_shell.rs b/src/server/client_shell.rs
index 37470693..554e8167 100644
--- a/src/server/client_shell.rs
+++ b/src/server/client_shell.rs
@@ -3,6 +3,7 @@ use ratatui::layout::Rect;
 use crate::app;
 use crate::protocol::{self, FrameData};
 
+#[cfg(test)]
 pub(super) fn snapshot(
     app: &app::App,
     boot_id: &str,
@@ -10,7 +11,29 @@ pub(super) fn snapshot(
     config_diagnostic: Option<&str>,
     location: Option<&crate::server::clients::ClientShellLocation>,
 ) -> protocol::ClientShellSnapshot {
+    snapshot_with_completions(app, boot_id, revision, config_diagnostic, location).0
+}
+
+pub(super) fn snapshot_with_completions(
+    app: &app::App,
+    boot_id: &str,
+    revision: u64,
+    config_diagnostic: Option<&str>,
+    location: Option<&crate::server::clients::ClientShellLocation>,
+) -> (
+    protocol::ClientShellSnapshot,
+    protocol::endpoint::EndpointAgentCompletions,
+) {
     let snapshot = app.session_snapshot();
+    let completions = protocol::endpoint::EndpointAgentCompletions {
+        boot_id: boot_id.to_owned(),
+        revision,
+        completions: snapshot
+            .agents
+            .iter()
+            .filter_map(|agent| agent.completion_seq.map(|seq| (agent.pane_id.clone(), seq)))
+            .collect(),
+    };
     let focused_workspace_id = location
         .and_then(|location| location.focused_workspace_id.clone())
         .or_else(|| snapshot.focused_workspace_id.clone());
@@ -216,7 +239,7 @@ pub(super) fn snapshot(
                 preview: notes.preview,
             });
 
-    protocol::ClientShellSnapshot {
+    let shell = protocol::ClientShellSnapshot {
         boot_id: boot_id.to_owned(),
         revision,
         config_diagnostic: config_diagnostic.map(str::to_owned),
@@ -240,7 +263,8 @@ pub(super) fn snapshot(
         panes,
         agents,
         commands: app.client_shell_command_manifest(),
-    }
+    };
+    (shell, completions)
 }
 
 pub(super) struct RenderedPaneSurface {
diff --git a/src/server/clients.rs b/src/server/clients.rs
index aeda3822..65e875b8 100644
--- a/src/server/clients.rs
+++ b/src/server/clients.rs
@@ -177,6 +177,7 @@ pub(crate) struct ClientConnection {
     pub(crate) shell_location: Option<ClientShellLocation>,
     /// Last coherent shell replacement sent to this client.
     pub(crate) shell_snapshot: Option<crate::protocol::ClientShellSnapshot>,
+    pub(crate) shell_agent_completions: Option<crate::protocol::endpoint::EndpointAgentCompletions>,
     /// View policy paired with the last coherent shell replacement.
     pub(crate) shell_agent_view: Option<crate::api::schema::AgentViewSetParams>,
     /// Monotonic shell replacement revision for this connection.
@@ -246,6 +247,7 @@ impl ClientConnection {
             staged_clipboard_files: Vec::new(),
             shell_location: None,
             shell_snapshot: None,
+            shell_agent_completions: None,
             shell_agent_view: None,
             shell_projection_revision: 0,
             shell_endpoint_command_in_flight: false,
diff --git a/src/server/headless.rs b/src/server/headless.rs
index 638559ad..c5a644ce 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -51,7 +51,8 @@ use crate::server::client_accept::{
     accept_pending_client_connections, reject_pending_client_connections,
 };
 use crate::server::client_shell::{
-    render_pane_surface as render_client_shell_pane_surface, snapshot as client_shell_snapshot,
+    render_pane_surface as render_client_shell_pane_surface,
+    snapshot_with_completions as client_shell_snapshot,
 };
 use crate::server::client_transport::ServerEvent;
 use crate::server::clients::{
@@ -2041,7 +2042,7 @@ impl HeadlessServer {
                 } else {
                     self.server_config_diagnostic_without_keybindings.as_deref()
                 };
-                let seed_snapshot = client_shell_snapshot(
+                let (seed_snapshot, completion_projection) = client_shell_snapshot(
                     &self.app,
                     &self.client_shell_boot_id,
                     connection.shell_projection_revision,
@@ -2073,8 +2074,18 @@ impl HeadlessServer {
                             return false;
                         }
                     };
+                let completion_message = match crate::protocol::endpoint::agent_completions_message(
+                    &completion_projection,
+                ) {
+                    Ok(message) => message,
+                    Err(err) => {
+                        warn!(client_id, err = %err, "failed to encode agent completions");
+                        return false;
+                    }
+                };
                 connection.shell_location = Some(location);
                 connection.shell_snapshot = Some(seed_snapshot);
+                connection.shell_agent_completions = Some(completion_projection);
                 connection.shell_agent_view = agent_view;
                 self.clients.insert(client_id, connection);
                 if self.app.state.popup_pane.is_some() && self.popup_owner_tab_id.is_none() {
@@ -2083,6 +2094,7 @@ impl HeadlessServer {
                 if let Some(message) = projection_message {
                     self.send_to_client(client_id, message);
                 }
+                self.send_to_client(client_id, completion_message);
                 self.send_to_client(client_id, snapshot_message);
                 if surface_active {
                     self.foreground_client_id = Some(client_id);
@@ -3241,7 +3253,10 @@ impl HeadlessServer {
             };
 
             let new_state = terminal_after.state;
-            if new_state == *prev_state {
+            if new_state == *prev_state
+                || (new_state == crate::detect::AgentState::Idle
+                    && terminal_after.last_agent_completion_seq.is_none())
+            {
                 continue;
             }
 
@@ -3274,15 +3289,11 @@ impl HeadlessServer {
                 && self.app.state.toast_config.delay_seconds == 0
                 && should_forward_toast_to_clients(self.app.state.toast_config.delivery)
             {
-                if let Some(kind) =
-                    crate::app::actions::notification_toast_for_state_change_with_agent_labels(
-                        suppress_active_tab_notifications,
-                        *prev_state,
-                        new_state,
-                        prev_agent_label.as_deref(),
-                        agent_label.as_deref(),
-                    )
-                {
+                if let Some(kind) = crate::app::actions::notification_toast_for_state_change(
+                    suppress_active_tab_notifications,
+                    *prev_state,
+                    new_state,
+                ) {
                     if let Some(agent_label) = self
                         .app
                         .state
@@ -3319,15 +3330,11 @@ impl HeadlessServer {
             // Clients still decide locally whether they can execute the side effect.
             if self.app.state.toast_config.delay_seconds == 0 && self.app.state.sound.allows(agent)
             {
-                if let Some(sound) =
-                    crate::app::actions::notification_sound_for_state_change_with_agent_labels(
-                        suppress_active_tab_notifications,
-                        *prev_state,
-                        new_state,
-                        prev_agent_label.as_deref(),
-                        agent_label.as_deref(),
-                    )
-                {
+                if let Some(sound) = crate::app::actions::notification_sound_for_state_change(
+                    suppress_active_tab_notifications,
+                    *prev_state,
+                    new_state,
+                ) {
                     debug!(sound = ?sound, "forwarding sound notification from API request");
                     self.send_notify_to_foreground_client(
                         protocol::NotifyKind::Sound,
diff --git a/src/server/headless/notifications.rs b/src/server/headless/notifications.rs
index 722f95c5..a575cfe1 100644
--- a/src/server/headless/notifications.rs
+++ b/src/server/headless/notifications.rs
@@ -19,20 +19,6 @@ impl HeadlessServer {
             .unwrap_or(crate::detect::AgentState::Unknown)
     }
 
-    fn pane_effective_agent_label(&self, pane_id: crate::layout::PaneId) -> Option<String> {
-        self.app.state.workspaces.iter().find_map(|ws| {
-            ws.tabs.iter().find_map(|tab| {
-                let pane = tab.panes.get(&pane_id)?;
-                self.app
-                    .state
-                    .terminals
-                    .get(&pane.attached_terminal_id)
-                    .and_then(|terminal| terminal.effective_agent_label())
-                    .map(str::to_string)
-            })
-        })
-    }
-
     fn forward_semantic_agent_notification(
         &mut self,
         update: &crate::app::actions::PaneStateUpdate,
@@ -61,13 +47,9 @@ impl HeadlessServer {
         agent_label: Option<&str>,
         known_agent: Option<crate::detect::Agent>,
     ) -> bool {
-        let Some(kind) = crate::app::actions::notification_toast_for_state_change_with_agent_labels(
-            false,
-            previous_state,
-            state,
-            previous_agent_label,
-            agent_label,
-        ) else {
+        let Some(kind) =
+            crate::app::actions::notification_toast_for_state_change(false, previous_state, state)
+        else {
             return false;
         };
         let Some(workspace) = self.app.state.workspaces.get(ws_idx) else {
@@ -142,15 +124,11 @@ impl HeadlessServer {
             self.active_tab_suppresses_notifications(is_active_tab);
 
         if !update.suppress_completion && self.app.state.sound.allows(update.known_agent) {
-            if let Some(sound) =
-                crate::app::actions::notification_sound_for_state_change_with_agent_labels(
-                    suppress_active_tab_notifications,
-                    update.previous_state,
-                    update.state,
-                    update.previous_agent_label.as_deref(),
-                    update.agent_label.as_deref(),
-                )
-            {
+            if let Some(sound) = crate::app::actions::notification_sound_for_state_change(
+                suppress_active_tab_notifications,
+                update.previous_state,
+                update.state,
+            ) {
                 self.send_notify_to_foreground_client(
                     protocol::NotifyKind::Sound,
                     sound_notify_message(sound),
@@ -360,11 +338,8 @@ impl HeadlessServer {
                 let pane_id_val = *pane_id;
                 let agent_val = *agent;
 
-                // Find the previous effective state of this pane before the event
-                // is processed. Notifications must follow effective state changes,
-                // not raw fallback reports that may be masked by hook authority.
+                // Notifications follow effective changes, not fallback reports masked by hooks.
                 let prev_state = self.pane_effective_state(pane_id_val);
-                let prev_agent_label = self.pane_effective_agent_label(pane_id_val);
 
                 // Handle the state change (updates pane state, sets toast on AppState).
                 // Headless mode disables local sound playback separately from the
@@ -396,21 +371,16 @@ impl HeadlessServer {
                     self.active_tab_suppresses_notifications(is_active_tab);
 
                 let next_state = self.pane_effective_state(pane_id_val);
-                let next_agent_label = self.pane_effective_agent_label(pane_id_val);
 
                 if !suppress_completion
                     && self.app.state.toast_config.delay_seconds == 0
                     && self.app.state.sound.allows(agent_val)
                 {
-                    if let Some(sound) =
-                        crate::app::actions::notification_sound_for_state_change_with_agent_labels(
-                            suppress_active_tab_notifications,
-                            prev_state,
-                            next_state,
-                            prev_agent_label.as_deref(),
-                            next_agent_label.as_deref(),
-                        )
-                    {
+                    if let Some(sound) = crate::app::actions::notification_sound_for_state_change(
+                        suppress_active_tab_notifications,
+                        prev_state,
+                        next_state,
+                    ) {
                         self.send_notify_to_foreground_client(
                             protocol::NotifyKind::Sound,
                             sound_notify_message(sound),
@@ -437,7 +407,6 @@ impl HeadlessServer {
                             suppress_active_tab_notifications,
                             prev_state,
                             next_state,
-                            prev_agent_label.as_deref(),
                         )
                     }
                 } else {
@@ -465,11 +434,8 @@ impl HeadlessServer {
                 let pane_id_val = *pane_id;
                 let agent_val = crate::detect::parse_agent_label(agent_label);
 
-                // Capture the previous effective state for this pane. Hook reports
-                // are already folded into pane.state; raw hook transitions must not
-                // produce a second notification path.
+                // Hook reports are already folded into the effective state.
                 let prev_state = self.pane_effective_state(pane_id_val);
-                let prev_agent_label = self.pane_effective_agent_label(pane_id_val);
 
                 self.sync_foreground_client_state();
                 let pane_updates = self.app.handle_internal_event_with_pane_updates(ev);
@@ -499,21 +465,16 @@ impl HeadlessServer {
                     self.active_tab_suppresses_notifications(is_active_tab);
 
                 let next_state = self.pane_effective_state(pane_id_val);
-                let next_agent_label = self.pane_effective_agent_label(pane_id_val);
 
                 if !suppress_completion
                     && self.app.state.toast_config.delay_seconds == 0
                     && self.app.state.sound.allows(agent_val)
                 {
-                    if let Some(sound) =
-                        crate::app::actions::notification_sound_for_state_change_with_agent_labels(
-                            suppress_active_tab_notifications,
-                            prev_state,
-                            next_state,
-                            prev_agent_label.as_deref(),
-                            next_agent_label.as_deref(),
-                        )
-                    {
+                    if let Some(sound) = crate::app::actions::notification_sound_for_state_change(
+                        suppress_active_tab_notifications,
+                        prev_state,
+                        next_state,
+                    ) {
                         self.send_notify_to_foreground_client(
                             protocol::NotifyKind::Sound,
                             sound_notify_message(sound),
@@ -540,7 +501,6 @@ impl HeadlessServer {
                             suppress_active_tab_notifications,
                             prev_state,
                             next_state,
-                            prev_agent_label.as_deref(),
                         )
                     }
                 } else {
diff --git a/src/server/headless/render.rs b/src/server/headless/render.rs
index dd2cce38..82564ba3 100644
--- a/src/server/headless/render.rs
+++ b/src/server/headless/render.rs
@@ -483,7 +483,7 @@ impl HeadlessServer {
                 let Some(client) = self.clients.get_mut(&client_id) else {
                     continue;
                 };
-                let mut candidate = client_shell_snapshot(
+                let (mut candidate, mut completions) = client_shell_snapshot(
                     &self.app,
                     &self.client_shell_boot_id,
                     client.shell_projection_revision,
@@ -497,11 +497,26 @@ impl HeadlessServer {
                 };
                 candidate.revision = client.shell_projection_revision;
                 if client.shell_snapshot.as_ref() != Some(&candidate)
+                    || client.shell_agent_completions.as_ref() != Some(&completions)
                     || client.shell_agent_view != agent_view
                 {
                     client.shell_projection_revision =
                         client.shell_projection_revision.saturating_add(1);
                     candidate.revision = client.shell_projection_revision;
+                    completions.revision = candidate.revision;
+                    let completion_framed =
+                        match crate::protocol::endpoint::agent_completions_message(&completions)
+                            .map_err(std::io::Error::other)
+                            .and_then(|message| {
+                                Self::frame_server_message(&message).map_err(std::io::Error::other)
+                            }) {
+                            Ok(message) => message,
+                            Err(err) => {
+                                warn!(client_id, err = %err, "failed to frame agent completions");
+                                broken_clients.push(client_id);
+                                continue;
+                            }
+                        };
                     let projection_message = if agent_view.is_some()
                         || client.shell_agent_view.is_some()
                     {
@@ -554,12 +569,14 @@ impl HeadlessServer {
                         continue;
                     };
                     if projection_framed.is_some_and(|framed| writer.control.send(framed).is_err())
+                        || writer.control.send(completion_framed).is_err()
                         || writer.control.send(snapshot_framed).is_err()
                     {
                         broken_clients.push(client_id);
                         continue;
                     }
                     client.shell_snapshot = Some(candidate);
+                    client.shell_agent_completions = Some(completions);
                     client.shell_agent_view = agent_view;
                 }
                 shell_projection_revision = client.shell_projection_revision;
diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs
index cf6d389b..6accdb09 100644
--- a/src/server/headless/tests/mod.rs
+++ b/src/server/headless/tests/mod.rs
@@ -9,12 +9,36 @@ mod surface_delta_tests;
 #[path = "surface_interest.rs"]
 mod surface_interest_tests;
 
-fn client_shell_snapshot(message: ServerMessage) -> Box<crate::protocol::ClientShellSnapshot> {
-    let ServerMessage::EndpointControl { kind, data } = message else {
-        panic!("expected client shell snapshot");
+fn client_shell_projection(
+    receiver: &std::sync::mpsc::Receiver<Vec<u8>>,
+) -> (
+    Box<protocol::ClientShellSnapshot>,
+    protocol::endpoint::EndpointAgentCompletions,
+) {
+    let read_control = |expected| {
+        let ServerMessage::EndpointControl { kind, data } = read_server_message(
+            receiver
+                .recv_timeout(Duration::from_secs(1))
+                .expect("endpoint projection"),
+        ) else {
+            panic!("expected endpoint control {expected}");
+        };
+        assert_eq!(kind, expected);
+        data
     };
-    assert_eq!(kind, crate::protocol::endpoint::ENDPOINT_SNAPSHOT_KIND);
-    Box::new(serde_json::from_str(&data).expect("decode client shell snapshot"))
+    let completions: protocol::endpoint::EndpointAgentCompletions =
+        serde_json::from_str(&read_control(protocol::endpoint::AGENT_COMPLETIONS_KIND)).unwrap();
+    let snapshot: Box<protocol::ClientShellSnapshot> =
+        serde_json::from_str(&read_control(protocol::endpoint::ENDPOINT_SNAPSHOT_KIND)).unwrap();
+    assert_eq!(completions.boot_id, snapshot.boot_id);
+    assert_eq!(completions.revision, snapshot.revision);
+    (snapshot, completions)
+}
+
+fn client_shell_snapshot(
+    receiver: &std::sync::mpsc::Receiver<Vec<u8>>,
+) -> Box<protocol::ClientShellSnapshot> {
+    client_shell_projection(receiver).0
 }
 
 fn client_agent_view_projection(
@@ -715,6 +739,64 @@ async fn client_shell_attach_seeds_workspace() {
 }
 
 #[tokio::test]
+async fn completion_guard_endpoint_pairs_runtime_completions_with_snapshots() {
+    let mut server = test_headless_server();
+    let workspace = crate::workspace::Workspace::test_new("endpoint");
+    let pane_id = workspace.tabs[0].root_pane;
+    server.app.state.workspaces = vec![workspace];
+    server.app.state.ensure_test_terminals();
+    server.app.state.active = Some(0);
+    let (writer, control_rx, _render_rx) = test_client_writer();
+    server.handle_server_event(ServerEvent::ClientShellConnected {
+        client_id: 78,
+        surface_cols: 80,
+        surface_rows: 24,
+        cell_width_px: 0,
+        cell_height_px: 0,
+        pixel_mouse: false,
+        direct_graphics: false,
+        endpoint_keybindings: false,
+        mouse_capture: false,
+        surface_active: false,
+        surface_reuse: false,
+        surface_delta: false,
+        writer,
+    });
+    let (_, initial) = client_shell_projection(&control_rx);
+    assert!(initial.completions.is_empty());
+    for first_state in [
+        crate::detect::AgentState::Working,
+        crate::detect::AgentState::Unknown,
+    ] {
+        for state in [first_state, crate::detect::AgentState::Idle] {
+            server.app.state.handle_app_event(AppEvent::StateChanged {
+                pane_id,
+                agent: Some(crate::detect::Agent::Pi),
+                state,
+                visible_blocker: false,
+                visible_working: state == crate::detect::AgentState::Working,
+                process_exited: false,
+                observed_at: Instant::now(),
+            });
+        }
+        server.render_and_stream();
+        let (snapshot, completions) = client_shell_projection(&control_rx);
+        assert_eq!(snapshot.agents.len(), 1);
+        let agent = &snapshot.agents[0];
+        let expected =
+            (first_state == crate::detect::AgentState::Working).then_some(agent.state_change_seq);
+        assert_eq!(
+            completions.completions.get(&agent.pane_id).copied(),
+            expected
+        );
+        assert_eq!(
+            server.app.session_snapshot().agents[0].completion_seq,
+            expected
+        );
+    }
+}
+
+#[tokio::test]
 async fn client_shell_endpoint_request_uses_the_selected_connection() {
     let mut server = test_headless_server();
     server.app.state.workspaces = vec![crate::workspace::Workspace::test_new("endpoint")];
@@ -739,7 +821,7 @@ async fn client_shell_endpoint_request_uses_the_selected_connection() {
             writer,
         })
     );
-    let _initial_snapshot = control_rx.recv().expect("initial shell snapshot");
+    let _initial_snapshot = client_shell_snapshot(&control_rx);
     let boot_id = server.client_shell_boot_id.clone();
 
     assert!(
@@ -857,9 +939,7 @@ async fn client_shell_pairs_agent_view_set_replacement_and_clear_with_snapshots(
             writer,
         })
     );
-    let initial = client_shell_snapshot(read_server_message(
-        control_rx.recv().expect("initial snapshot"),
-    ));
+    let initial = client_shell_snapshot(&control_rx);
 
     let mut view = AgentViewSetParams {
         source: "example.views".into(),
@@ -875,9 +955,7 @@ async fn client_shell_pairs_agent_view_set_replacement_and_clear_with_snapshots(
     let set = client_agent_view_projection(read_server_message(
         control_rx.recv().expect("set projection"),
     ));
-    let set_snapshot = client_shell_snapshot(read_server_message(
-        control_rx.recv().expect("set snapshot"),
-    ));
+    let set_snapshot = client_shell_snapshot(&control_rx);
     assert!(set.revision > initial.revision);
     assert_eq!(set.revision, set_snapshot.revision);
     assert_eq!(
@@ -894,9 +972,7 @@ async fn client_shell_pairs_agent_view_set_replacement_and_clear_with_snapshots(
     let replacement = client_agent_view_projection(read_server_message(
         control_rx.recv().expect("replacement projection"),
     ));
-    let replacement_snapshot = client_shell_snapshot(read_server_message(
-        control_rx.recv().expect("replacement snapshot"),
-    ));
+    let replacement_snapshot = client_shell_snapshot(&control_rx);
     assert!(replacement.revision > set.revision);
     assert_eq!(replacement.revision, replacement_snapshot.revision);
     assert_eq!(
@@ -913,9 +989,7 @@ async fn client_shell_pairs_agent_view_set_replacement_and_clear_with_snapshots(
     let cleared = client_agent_view_projection(read_server_message(
         control_rx.recv().expect("clear projection"),
     ));
-    let cleared_snapshot = client_shell_snapshot(read_server_message(
-        control_rx.recv().expect("clear snapshot"),
-    ));
+    let cleared_snapshot = client_shell_snapshot(&control_rx);
     assert!(cleared.revision > replacement.revision);
     assert_eq!(cleared.revision, cleared_snapshot.revision);
     assert!(cleared.view.is_none());
@@ -967,9 +1041,7 @@ async fn client_shell_receives_metadata_then_shell_free_pane_surface() {
             writer,
         })
     );
-    let snapshot = client_shell_snapshot(read_server_message(
-        control_rx.recv().expect("shell snapshot"),
-    ));
+    let snapshot = client_shell_snapshot(&control_rx);
     assert_eq!(snapshot.workspaces.len(), 1);
     assert_eq!(snapshot.workspaces[0].label, "shell-only-label");
     assert_eq!(
@@ -1599,9 +1671,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
             writer: local_writer,
         })
     );
-    let local_snapshot = client_shell_snapshot(read_server_message(
-        local_control.recv().expect("local shell snapshot"),
-    ));
+    let local_snapshot = client_shell_snapshot(&local_control);
     assert_eq!(
         local_snapshot.config_diagnostic.as_deref(),
         Some("theme warning")
@@ -1625,9 +1695,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() {
             writer: endpoint_writer,
         })
     );
-    let endpoint_snapshot = client_shell_snapshot(read_server_message(
-        endpoint_control.recv().expect("endpoint shell snapshot"),
-    ));
+    let endpoint_snapshot = client_shell_snapshot(&endpoint_control);
     assert_eq!(
         endpoint_snapshot.config_diagnostic.as_deref(),
         Some("server keybinding warning\ntheme warning")
@@ -1658,12 +1726,8 @@ async fn client_shell_tab_focus_changes_only_the_source_connection() {
 
     let (first_control, _first_render) = connect_test_shell(&mut server, 7, 100, 30);
     let (second_control, _second_render) = connect_test_shell(&mut server, 8, 80, 24);
-    let first_initial = client_shell_snapshot(read_server_message(
-        first_control.recv().expect("first snapshot"),
-    ));
-    let second_initial = client_shell_snapshot(read_server_message(
-        second_control.recv().expect("second snapshot"),
-    ));
+    let first_initial = client_shell_snapshot(&first_control);
+    let second_initial = client_shell_snapshot(&second_control);
     assert_eq!(
         first_initial.focused_tab_id.as_deref(),
         Some(first_tab_id.as_str())
@@ -1699,9 +1763,7 @@ async fn client_shell_tab_focus_changes_only_the_source_connection() {
         first_control.try_recv().is_err(),
         "another shell must not receive a navigation replacement"
     );
-    let second_replacement = client_shell_snapshot(read_server_message(
-        second_control.recv().expect("second replacement snapshot"),
-    ));
+    let second_replacement = client_shell_snapshot(&second_control);
     assert_eq!(
         second_replacement.focused_tab_id.as_deref(),
         Some(second_tab_id.as_str())
@@ -2433,13 +2495,11 @@ async fn public_agent_focus_replaces_a_diverged_client_shell_projection() {
     let second_tab_id = server.app.public_tab_id(1, 0).unwrap();
 
     let (control_rx, render_rx) = connect_test_shell(&mut server, 9, 80, 23);
-    let _ = control_rx.recv().expect("initial snapshot");
+    let _ = client_shell_snapshot(&control_rx);
     assert!(server.focus_shell_client_on_tab(9, &second_tab_id));
     assert!(server.claim_shell_tab_geometry(9, false));
     server.render_and_stream();
-    let diverged = client_shell_snapshot(read_server_message(
-        control_rx.recv().expect("diverged snapshot"),
-    ));
+    let diverged = client_shell_snapshot(&control_rx);
     assert_eq!(
         diverged.focused_workspace_id.as_deref(),
         Some(server.app.public_workspace_id(1).as_str())
@@ -2484,9 +2544,7 @@ async fn public_agent_focus_replaces_a_diverged_client_shell_projection() {
     assert_eq!(location.focused_tab_id(), Some(first_tab_id.as_str()));
 
     server.render_and_stream();
-    let replacement = client_shell_snapshot(read_server_message(
-        control_rx.recv().expect("agent focus replacement snapshot"),
-    ));
+    let replacement = client_shell_snapshot(&control_rx);
     assert_eq!(
         replacement.focused_workspace_id.as_deref(),
         Some(first_workspace_id.as_str())
@@ -2529,10 +2587,7 @@ async fn public_api_focus_replaces_every_client_shell_projection() {
             writer,
         })
     );
-    let initial_revision = client_shell_snapshot(read_server_message(
-        control_rx.recv().expect("initial snapshot"),
-    ))
-    .revision;
+    let initial_revision = client_shell_snapshot(&control_rx).revision;
 
     let (respond_to, _response_rx) = std::sync::mpsc::channel();
     server.handle_api_request_with_shutdown_check(crate::api::ApiRequestMessage {
@@ -2551,9 +2606,7 @@ async fn public_api_focus_replaces_every_client_shell_projection() {
     assert_eq!(server.app.state.active, Some(1));
     server.render_and_stream();
 
-    let replacement = client_shell_snapshot(read_server_message(
-        control_rx.recv().expect("replacement snapshot"),
-    ));
+    let replacement = client_shell_snapshot(&control_rx);
     assert!(replacement.revision > initial_revision);
     assert_eq!(
         replacement.focused_workspace_id.as_deref(),
@@ -2781,9 +2834,7 @@ async fn client_shell_streams_and_targets_popup_terminal_content() {
             writer,
         })
     );
-    let _snapshot = client_shell_snapshot(read_server_message(
-        control_rx.recv().expect("shell snapshot"),
-    ));
+    let _snapshot = client_shell_snapshot(&control_rx);
 
     server.render_and_stream();
     let ServerMessage::PaneSurface(surface) =
@@ -6726,6 +6777,225 @@ fn notification_show_api_includes_sound_in_semantic_event() {
     }
 }
 
+fn completion_guard_server(writer: ClientWriter) -> (HeadlessServer, crate::layout::PaneId) {
+    let mut server = test_headless_server();
+    server.app.state.workspaces = ["background", "active"]
+        .map(crate::workspace::Workspace::test_new)
+        .into();
+    server.app.state.ensure_test_terminals();
+    server.app.state.active = Some(1);
+    server.app.state.toast_config.delay_seconds = 0;
+    server.clients.insert(
+        1,
+        ClientConnection::new(
+            (80, 24),
+            Default::default(),
+            1,
+            RenderEncoding::SemanticFrame,
+            Some(writer),
+        ),
+    );
+    let pane_id = server.app.state.workspaces[0].tabs[0].root_pane;
+    (server, pane_id)
+}
+
+fn completion_guard_api_report(server: &mut HeadlessServer, method: api::schema::Method) {
+    let (respond_to, response_rx) = std::sync::mpsc::channel();
+    server.handle_api_request_with_shutdown_check(api::ApiRequestMessage {
+        request: api::schema::Request {
+            id: "completion-probe".into(),
+            method,
+        },
+        respond_to,
+        response_write_complete: None,
+        stream_active: None,
+    });
+    let response = response_rx
+        .recv_timeout(Duration::from_millis(100))
+        .unwrap();
+    serde_json::from_str::<api::schema::SuccessResponse>(&response).expect("successful report");
+}
+
+fn completion_guard_notifications(
+    server: &mut HeadlessServer,
+    receiver: &std::sync::mpsc::Receiver<Vec<u8>>,
+) -> Vec<protocol::SemanticNotificationKind> {
+    server.send_to_client(
+        1,
+        ServerMessage::EndpointControl {
+            kind: "test.completion-barrier".into(),
+            data: String::new(),
+        },
+    );
+    let mut messages = Vec::new();
+    loop {
+        let message = read_server_message(
+            receiver
+                .recv_timeout(Duration::from_secs(1))
+                .expect("notification barrier"),
+        );
+        if matches!(&message, ServerMessage::EndpointControl { kind, .. } if kind == "test.completion-barrier")
+        {
+            return messages;
+        }
+        if let ServerMessage::SemanticNotification(notification) = message {
+            messages.push(notification.kind);
+        }
+    }
+}
+
+#[test]
+fn completion_guard_api_startup_blocker_respects_suppression() {
+    use protocol::SemanticNotificationKind::{Finished, NeedsAttention};
+
+    let (writer, control_rx, _render_rx) = test_client_writer();
+    let (mut server, pane_id) = completion_guard_server(writer);
+    server.handle_internal_event_with_forwarding(AppEvent::AgentProcessDetected {
+        pane_id,
+        agent: crate::detect::Agent::Pi,
+        observed_at: Instant::now(),
+    });
+    let public_pane_id = server.app.public_pane_id(0, pane_id).unwrap();
+    for (seq, state) in [
+        api::schema::PaneAgentState::Blocked,
+        api::schema::PaneAgentState::Idle,
+    ]
+    .into_iter()
+    .enumerate()
+    {
+        completion_guard_api_report(
+            &mut server,
+            api::schema::Method::PaneReportAgent(api::schema::PaneReportAgentParams {
+                pane_id: public_pane_id.clone(),
+                source: "custom:pi".into(),
+                agent: "pi".into(),
+                state,
+                message: None,
+                seq: Some(seq as u64 + 1),
+                agent_session_id: None,
+                agent_session_path: None,
+            }),
+        );
+    }
+    let notifications = completion_guard_notifications(&mut server, &control_rx);
+    assert!(notifications.contains(&NeedsAttention));
+    assert!(
+        !notifications.contains(&Finished),
+        "startup: {notifications:?}"
+    );
+}
+
+#[test]
+fn completion_guard_api_session_replacement_does_not_notify_finished() {
+    use crate::agent_resume::{AgentSessionRef, PersistedAgentSession};
+    use crate::detect::{Agent, AgentState};
+    use api::schema::{
+        Method, PaneAgentState, PaneReportAgentParams, PaneReportAgentSessionParams,
+    };
+    use protocol::SemanticNotificationKind::Finished;
+
+    for fallback_state in [
+        AgentState::Unknown,
+        AgentState::Working,
+        AgentState::Blocked,
+    ] {
+        for reason in ["new", "resume", "fork"] {
+            let (writer, control_rx, _render_rx) = test_client_writer();
+            let (mut server, pane_id) = completion_guard_server(writer);
+            let terminal_id = server.app.state.workspaces[0].panes[&pane_id]
+                .attached_terminal_id
+                .clone();
+            let terminal = server.app.state.terminals.get_mut(&terminal_id).unwrap();
+            terminal.set_detected_state(Some(Agent::Pi), fallback_state);
+            let cwd = std::env::current_dir().unwrap();
+            let old_session =
+                AgentSessionRef::path(cwd.join("completion-old.jsonl").display().to_string())
+                    .unwrap();
+            let new_session = cwd.join("completion-new.jsonl").display().to_string();
+            terminal.set_persisted_agent_session(PersistedAgentSession {
+                source: "herdr:pi".into(),
+                agent: "pi".into(),
+                session_ref: old_session.clone(),
+            });
+            terminal
+                .set_hook_authority_with_session_ref(
+                    "herdr:pi".into(),
+                    "pi".into(),
+                    AgentState::Idle,
+                    None,
+                    Some(old_session),
+                    Some(10),
+                )
+                .expect("initial authority");
+            let public_pane_id = server.app.public_pane_id(0, pane_id).unwrap();
+            completion_guard_api_report(
+                &mut server,
+                Method::PaneReportAgentSession(PaneReportAgentSessionParams {
+                    pane_id: public_pane_id.clone(),
+                    source: "herdr:pi".into(),
+                    agent: "pi".into(),
+                    seq: Some(11),
+                    agent_session_id: None,
+                    agent_session_path: Some(new_session.clone()),
+                    session_start_source: Some(reason.into()),
+                }),
+            );
+            let mut report = PaneReportAgentParams {
+                pane_id: public_pane_id,
+                source: "herdr:pi".into(),
+                agent: "pi".into(),
+                state: PaneAgentState::Idle,
+                message: None,
+                seq: Some(12),
+                agent_session_id: None,
+                agent_session_path: Some(new_session.clone()),
+            };
+            completion_guard_api_report(&mut server, Method::PaneReportAgent(report.clone()));
+            let terminal = &server.app.state.terminals[&terminal_id];
+            assert_eq!(terminal.state, AgentState::Idle);
+            assert_eq!(
+                terminal
+                    .hook_authority
+                    .as_ref()
+                    .unwrap()
+                    .session_ref
+                    .as_ref()
+                    .unwrap()
+                    .value,
+                new_session
+            );
+            let notifications = completion_guard_notifications(&mut server, &control_rx);
+            assert!(
+                !notifications.contains(&Finished),
+                "{reason}: {notifications:?}"
+            );
+            assert!(server.app.state.workspaces[0].panes[&pane_id].seen);
+            for (seq, state, expected) in [
+                (13, PaneAgentState::Working, AgentState::Working),
+                (14, PaneAgentState::Idle, AgentState::Idle),
+            ] {
+                report.seq = Some(seq);
+                report.state = state;
+                completion_guard_api_report(&mut server, Method::PaneReportAgent(report.clone()));
+                let terminal = &server.app.state.terminals[&terminal_id];
+                assert_eq!(terminal.state, expected);
+                assert_eq!(
+                    terminal.last_agent_completion_seq.is_some(),
+                    expected == AgentState::Idle
+                );
+            }
+            let finished = completion_guard_notifications(&mut server, &control_rx)
+                .into_iter()
+                .filter(|kind| *kind == Finished)
+                .count();
+            assert_eq!(
+                finished, 1,
+                "the first real turn after {reason} must finish"
+            );
+        }
+    }
+}
+
 #[test]
 fn startup_idle_does_not_forward_completion() {
     let mut server = test_headless_server();
diff --git a/src/server/headless/tests/pane_graphics.rs b/src/server/headless/tests/pane_graphics.rs
index 0b7a8095..3e36a8fc 100644
--- a/src/server/headless/tests/pane_graphics.rs
+++ b/src/server/headless/tests/pane_graphics.rs
@@ -56,7 +56,8 @@ async fn client_shell_surface_sends_complete_placements_and_each_live_asset_once
 
 #[tokio::test]
 async fn client_shell_asset_delivery_is_bounded_to_the_current_live_scene() {
-    let (mut server, client_rx, pane_id) = retained_test_server(b"client shell graphics");
+    let (mut server, _control_rx, client_rx, pane_id) =
+        retained_test_server_with_control(b"client shell graphics");
     let client = server.clients.get_mut(&1).unwrap();
     client.mode = ClientConnectionMode::ClientShell;
     client.render_state =
@@ -92,7 +93,8 @@ async fn client_shell_asset_delivery_is_bounded_to_the_current_live_scene() {
 
 #[tokio::test]
 async fn first_kitty_image_updates_retained_surface_without_full_redraw() {
-    let (mut server, client_rx, pane_id) = retained_test_server(b"text before image");
+    let (mut server, _control_rx, client_rx, pane_id) =
+        retained_test_server_with_control(b"text before image");
     let client = server.clients.get_mut(&1).unwrap();
     client.mode = ClientConnectionMode::ClientShell;
     client.render_state =
@@ -183,7 +185,8 @@ async fn first_kitty_image_updates_retained_surface_without_full_redraw() {
 
 #[tokio::test]
 async fn retained_unicode_image_arrives_after_fragmented_upload_without_reupload() {
-    let (mut server, client_rx, pane_id) = retained_test_server(b"\x1b[?1049h");
+    let (mut server, _control_rx, client_rx, pane_id) =
+        retained_test_server_with_control(b"\x1b[?1049h");
     let client = server.clients.get_mut(&1).unwrap();
     client.mode = ClientConnectionMode::ClientShell;
     client.cell_size = crate::kitty_graphics::HostCellSize {
@@ -260,7 +263,8 @@ async fn render_scale_profile_retained_graphics() {
     for retained in [false, true] {
         for with_image in [false, true] {
             for count in [1, 15] {
-                let (mut server, client_rx, root) = retained_test_server(b"populated terminal\r\n");
+                let (mut server, _control_rx, client_rx, root) =
+                    retained_test_server_with_control(b"populated terminal\r\n");
                 let mut pane_ids = vec![root];
                 for index in 1..count {
                     let workspace = &mut server.app.state.workspaces[0];
@@ -336,8 +340,9 @@ async fn render_scale_profile_retained_graphics() {
 
 #[tokio::test]
 async fn client_shell_surface_projects_terminal_kitty_images_from_authoritative_runtime() {
-    let (mut server, client_rx, _pane_id) =
-        retained_test_server(b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\");
+    let (mut server, _control_rx, client_rx, _pane_id) = retained_test_server_with_control(
+        b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\",
+    );
     let client = server.clients.get_mut(&1).unwrap();
     client.mode = ClientConnectionMode::ClientShell;
     client.render_state =
@@ -366,7 +371,7 @@ async fn client_shell_surface_projects_terminal_kitty_images_from_authoritative_
 
 #[tokio::test]
 async fn client_shell_delivers_equal_pixels_for_distinct_terminal_image_ids() {
-    let (mut server, client_rx, _pane_id) = retained_test_server(
+    let (mut server, _control_rx, client_rx, _pane_id) = retained_test_server_with_control(
         b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\\x1b_Ga=T,f=32,t=d,i=8,p=4,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\",
     );
     let client = server.clients.get_mut(&1).unwrap();
@@ -406,7 +411,8 @@ async fn client_shell_delivers_equal_pixels_for_distinct_terminal_image_ids() {
 
 #[tokio::test]
 async fn full_client_shell_render_lane_does_not_commit_graphics_delivery() {
-    let (mut server, client_rx, pane_id) = retained_test_server(b"client shell graphics");
+    let (mut server, _control_rx, client_rx, pane_id) =
+        retained_test_server_with_control(b"client shell graphics");
     let client = server.clients.get_mut(&1).unwrap();
     client.mode = ClientConnectionMode::ClientShell;
     client.render_state =
@@ -975,7 +981,8 @@ fn resident_direct_stream_survives_non_direct_client_becoming_foreground() {
 #[cfg(unix)]
 #[tokio::test]
 async fn client_shell_direct_graphics_uploads_without_server_authored_coordinates() {
-    let (mut server, client_rx, pane_id) = retained_test_server(b"client shell direct");
+    let (mut server, _control_rx, client_rx, pane_id) =
+        retained_test_server_with_control(b"client shell direct");
     server.app.state.kitty_graphics_enabled = true;
     let client = server.clients.get_mut(&1).unwrap();
     client.mode = ClientConnectionMode::ClientShell;
diff --git a/src/server/headless/tests/pane_move.rs b/src/server/headless/tests/pane_move.rs
index c18d0f1f..789b144f 100644
--- a/src/server/headless/tests/pane_move.rs
+++ b/src/server/headless/tests/pane_move.rs
@@ -55,7 +55,7 @@ async fn public_pane_move_focus_follows_the_moved_pane() {
     let source_id = server.app.public_pane_id(0, source).unwrap();
     let destination_id = server.app.public_workspace_id(1);
     let (control_rx, render_rx) = connect_test_shell(&mut server, 9, 80, 23);
-    let initial = client_shell_snapshot(read_server_message(control_rx.recv().unwrap()));
+    let initial = client_shell_snapshot(&control_rx);
 
     let moved = public_move(
         &mut server,
@@ -81,7 +81,7 @@ async fn public_pane_move_focus_follows_the_moved_pane() {
     assert_eq!(location.focused_tab_id(), Some(moved.pane.tab_id.as_str()));
 
     server.render_and_stream();
-    let snapshot = client_shell_snapshot(read_server_message(control_rx.recv().unwrap()));
+    let snapshot = client_shell_snapshot(&control_rx);
     assert!(snapshot.revision > initial.revision);
     assert_eq!(
         snapshot.focused_workspace_id.as_deref(),
diff --git a/src/server/headless/tests/surface_interest.rs b/src/server/headless/tests/surface_interest.rs
index 7c8b33f8..f3acc27b 100644
--- a/src/server/headless/tests/surface_interest.rs
+++ b/src/server/headless/tests/surface_interest.rs
@@ -79,10 +79,7 @@ async fn metadata_only_shell_is_isolated_until_surface_activation() {
             writer,
         })
     );
-    assert!(matches!(
-        read_server_message(control_rx.recv().expect("metadata snapshot")),
-        ServerMessage::EndpointControl { .. }
-    ));
+    let _ = client_shell_snapshot(&control_rx);
     assert_eq!(server.foreground_client_id, None);
     assert_eq!(server.effective_size, original_size);
 
@@ -410,7 +407,7 @@ async fn presentation_sync_epoch_replays_modes_and_title() {
             writer,
         })
     );
-    let _ = control_rx.recv().expect("initial snapshot");
+    let _ = client_shell_snapshot(&control_rx);
     server.api_window_title = Some("target title".into());
     {
         let client = server.clients.get_mut(&client_id).unwrap();
@@ -526,11 +523,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
             writer: source_writer,
         })
     );
-    let source_snapshot = client_shell_snapshot(read_server_message(
-        source_control
-            .recv()
-            .expect("real source metadata snapshot"),
-    ));
+    let source_snapshot = client_shell_snapshot(&source_control);
 
     let mut target_server = test_headless_server();
     let _target_input = install_focused_test_runtime(&mut target_server, b"remote target");
@@ -553,11 +546,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
             writer: target_writer,
         })
     );
-    let remote_snapshot = client_shell_snapshot(read_server_message(
-        target_control
-            .recv()
-            .expect("real target metadata snapshot"),
-    ));
+    let remote_snapshot = client_shell_snapshot(&target_control);
 
     let profile = SavedSshEndpoint {
         id: ProfileId::parse("0123456789abcdef0123456789abcdef").unwrap(),
@@ -685,9 +674,7 @@ async fn two_headless_servers_drive_atomic_endpoint_handoff() {
     );
 
     target_server.render_and_stream();
-    let coherent_snapshot = client_shell_snapshot(read_server_message(
-        target_control.recv().expect("target replacement snapshot"),
-    ));
+    let coherent_snapshot = client_shell_snapshot(&target_control);
     let snapshot_progress = activation.receive_snapshot(&target_id, 7, &coherent_snapshot);
     shell.set_endpoint_snapshot(&target_id, coherent_snapshot);
     assert_eq!(
diff --git a/src/server/notifications.rs b/src/server/notifications.rs
index 4ce33a24..724156b8 100644
--- a/src/server/notifications.rs
+++ b/src/server/notifications.rs
@@ -25,7 +25,6 @@ pub(crate) fn toast_message_from_state_change(
     suppress_active_tab_notifications: bool,
     prev_state: AgentState,
     new_state: AgentState,
-    previous_agent_label: Option<&str>,
 ) -> Option<String> {
     state
         .workspaces
@@ -38,12 +37,10 @@ pub(crate) fn toast_message_from_state_change(
                     .terminals
                     .get(&pane.attached_terminal_id)
                     .and_then(|terminal| terminal.effective_agent_label())?;
-                let kind = app::actions::notification_toast_for_state_change_with_agent_labels(
+                let kind = app::actions::notification_toast_for_state_change(
                     suppress_active_tab_notifications,
                     prev_state,
                     new_state,
-                    previous_agent_label,
-                    Some(agent_label),
                 )?;
                 let workspace_label = ws.display_name_from(&state.terminals, terminal_runtimes);
                 Some(format!(
@@ -142,7 +139,6 @@ mod tests {
             false,
             AgentState::Working,
             AgentState::Idle,
-            Some("codex"),
         );
 
         assert_eq!(
diff --git a/src/server/render_scale_benchmark.rs b/src/server/render_scale_benchmark.rs
index 5c3935de..dcafa1a9 100644
--- a/src/server/render_scale_benchmark.rs
+++ b/src/server/render_scale_benchmark.rs
@@ -233,13 +233,36 @@ fn profile_snapshot_encoding(
     count: usize,
     client_count: usize,
 ) -> StageStats {
-    let pipeline = RenderPipeline::new(build(count));
+    let mut pipeline = RenderPipeline::new(build(count));
+    pipeline.app.state.ensure_test_terminals();
+    for terminal in pipeline.app.state.terminals.values_mut() {
+        terminal.set_detected_state(
+            Some(crate::detect::Agent::Pi),
+            crate::detect::AgentState::Idle,
+        );
+        terminal.last_agent_state_change_seq = Some(1);
+        terminal.last_agent_completion_seq = Some(1);
+    }
     let run = || {
         let started = Instant::now();
-        let template = super::client_shell::snapshot(&pipeline.app, "bench-boot", 1, None, None);
+        let (template, completions) = super::client_shell::snapshot_with_completions(
+            &pipeline.app,
+            "bench-boot",
+            1,
+            None,
+            None,
+        );
         for client_index in 0..client_count {
             let mut snapshot = template.clone();
             snapshot.revision = client_index as u64 + 1;
+            let mut completions = completions.clone();
+            completions.revision = snapshot.revision;
+            let companion = crate::protocol::endpoint::agent_completions_message(&completions)
+                .expect("benchmark completion projection should serialize");
+            black_box(
+                bincode::serde::encode_to_vec(companion, bincode::config::standard())
+                    .expect("benchmark completion projection should frame"),
+            );
             let message = crate::protocol::endpoint::snapshot_message(&snapshot)
                 .expect("benchmark snapshot should serialize");
             black_box(
@@ -256,7 +279,7 @@ fn profile_snapshot_encoding(
 }
 
 fn print_snapshot_encoding_profiles(label: &str, build: fn(usize) -> Vec<Workspace>) {
-    println!("{label} snapshot projection + JSON framing");
+    println!("{label} populated agent snapshots + completions + JSON framing");
     println!("       panes  clients  median_us  p95_us  max_us");
     for count in CARDINALITIES {
         for client_count in CLIENT_CARDINALITIES {
diff --git a/src/terminal/state.rs b/src/terminal/state.rs
index c084b034..119916c8 100644
--- a/src/terminal/state.rs
+++ b/src/terminal/state.rs
@@ -142,6 +142,7 @@ pub struct TerminalState {
     metadata_token_sequence_sources: std::collections::HashSet<String>,
     pub state: AgentState,
     pub last_agent_state_change_seq: Option<u64>,
+    pub last_agent_completion_seq: Option<u64>,
     pub revision: u64,
     pub launch_argv: Option<Vec<String>>,
     pub respawn_shell_on_exit: bool,
@@ -178,6 +179,7 @@ impl TerminalState {
             metadata_token_sequence_sources: std::collections::HashSet::new(),
             state: AgentState::Unknown,
             last_agent_state_change_seq: None,
+            last_agent_completion_seq: None,
             revision: 0,
             launch_argv: None,
             respawn_shell_on_exit: false,
@@ -584,14 +586,18 @@ impl TerminalState {
         if agent.is_none() && self.recent_agent_process_exit.is_some() {
             self.clear_agent_name();
         }
+        let effective_state_change = self.recompute_effective_state(
+            previous_agent_label,
+            previous_known_agent,
+            previous_state,
+            previous_presentation,
+            now,
+        );
+        if fallback_state == AgentState::Working && self.state == AgentState::Working {
+            self.agent_process_acquisition_pending = false;
+        }
         TerminalStateMutation {
-            effective_state_change: self.recompute_effective_state(
-                previous_agent_label,
-                previous_known_agent,
-                previous_state,
-                previous_presentation,
-                now,
-            ),
+            effective_state_change,
             session_ref_changed: previous_session
                 != self.current_session_identity_for_persistence(),
             agent_released,
@@ -742,14 +748,18 @@ impl TerminalState {
             session_ref,
         });
         let current_session = self.current_session_identity_for_persistence();
+        let effective_state_change = self.recompute_effective_state(
+            previous_agent_label,
+            previous_known_agent,
+            previous_state,
+            previous_presentation,
+            now,
+        );
+        if state == AgentState::Working && self.state == AgentState::Working {
+            self.agent_process_acquisition_pending = false;
+        }
         Some(TerminalStateMutation {
-            effective_state_change: self.recompute_effective_state(
-                previous_agent_label,
-                previous_known_agent,
-                previous_state,
-                previous_presentation,
-                now,
-            ),
+            effective_state_change,
             session_ref_changed: previous_session != current_session,
             agent_released: false,
         })
@@ -1615,6 +1625,10 @@ impl TerminalState {
         }
         self.persisted_agent_session = Some(persisted_session);
         let current_session = self.current_session_identity_for_persistence();
+        if previous_session.is_some() && previous_session != current_session {
+            // Rebinding can expose a cached Working screen; only a fresh report ends acquisition.
+            self.agent_process_acquisition_pending = true;
+        }
         Some(TerminalStateMutation {
             effective_state_change: self.recompute_effective_state(
                 previous_agent_label,
@@ -1925,6 +1939,7 @@ impl TerminalState {
         timeout: Duration,
     ) {
         self.set_agent_name(name);
+        self.agent_process_acquisition_pending = true;
         self.agent_name_owner = Some(AgentNameOwner {
             agent_label: crate::detect::agent_label(kind).to_string(),
             session_ref: None,
@@ -2092,6 +2107,7 @@ impl TerminalState {
         self.stale_full_lifecycle_hook_sessions.clear();
         self.state = AgentState::Unknown;
         self.last_agent_state_change_seq = None;
+        self.last_agent_completion_seq = None;
         self.launch_argv = None;
         self.respawn_shell_on_exit = false;
         self.recent_agent_process_exit = None;