Merge pull request #890 from l0ng-ai/fix/unread-badge-on-reattach

Stop a reattach from badging every restored agent tab

src/terminal/remote.rs

@@ -16,7 +16,7 @@ use crate::terminal::parked_cursor::{CursorCut, ParkedCursorRepair, ParkedCursor
 
 use std::collections::VecDeque;
 
-use crate::core::cli_agent::{AgentSessionState, CLIAgent};
+use crate::core::cli_agent::{AgentSessionState, AgentStatus, CLIAgent};
 use crate::core::config::CursorStyle as ConfigCursorStyle;
 use crate::core::osc::OscTokenizer;
 use crate::daemon::protocol::{
@@ -64,12 +64,35 @@ struct ShellState {
     cycle: u64,
 }
 
+/// The pane's agent status as the client last heard it, plus how it heard it.
+///
+/// A reattach — the app restarting onto panes the daemon kept alive, or a
+/// dropped link coming back — has the daemon replay the pane's *stored* status
+/// as an ordinary `AgentStatus` frame ([`crate::daemon`]'s `replay_state`).
+/// Nothing on the wire distinguishes it from a live transition, and a client
+/// that reads it as one concludes that every restored agent finished its turn
+/// in the instant the window opened. `replayed` is that distinction, kept in
+/// the same lock as the value it describes so a reader can never observe the
+/// status without also learning where it came from.
+#[derive(Default)]
+struct AgentSlot {
+    state: Option<AgentSessionState>,
+    /// `state` arrived as an attach replay and no one has adopted it yet.
+    /// Cleared by the first taker — the view adopts it as a baseline rather
+    /// than as an edge.
+    replayed: bool,
+}
+
 struct ReaderSignals {
     cwd: Arc<Mutex<Option<PathBuf>>>,
     shell: Arc<Mutex<ShellState>>,
     remote: Arc<Mutex<Option<RemoteContext>>>,
     agent: Arc<Mutex<Option<CLIAgent>>>,
-    agent_session: Arc<Mutex<Option<AgentSessionState>>>,
+    agent_session: Arc<Mutex<AgentSlot>>,
+    /// Whether this link still owes us the attach replay. The reader keeps it
+    /// as a plain local: a link's replay is a property of that link's stream
+    /// position, and nothing outside the reader thread ever needs to read it.
+    awaiting_replay: bool,
     exited: Arc<AtomicBool>,
     child_exited: Arc<AtomicBool>,
     zle_reading: Arc<AtomicBool>,
@@ -571,7 +594,7 @@ pub struct RemoteTerminal {
     ssh_user: Option<String>,
     auto_supplied_password: bool,
     agent: Arc<Mutex<Option<CLIAgent>>>,
-    agent_session: Arc<Mutex<Option<AgentSessionState>>>,
+    agent_session: Arc<Mutex<AgentSlot>>,
     /// Kitty-graphics images placed on this pane's grid (issue #213).
     /// Written by the reader thread from out-of-band `Image`/`DeleteImage`
     /// frames, read by the paint path — only the client holds the grid the
@@ -819,7 +842,8 @@ impl RemoteTerminal {
             }
             Err(e) => return Err(e),
         };
-        let mut term = Self::from_stream_with(stream, size, buffered, PtySource::for_route(route))?;
+        let mut term =
+            Self::from_stream_parts(stream, size, buffered, PtySource::for_route(route), true)?;
         term.route = route.clone();
         Ok(term)
     }
@@ -866,6 +890,17 @@ impl RemoteTerminal {
 
         let read_half = stream.try_clone()?;
 
+        // A relink keeps the view, and the view keeps the status it last saw
+        // before the link dropped. That is already a baseline, and the better
+        // one: if the agent finished its turn while the link was down, the
+        // daemon's replayed `Done` against the view's `Working` is exactly the
+        // edge the reader must be told about. So the relink's replay is read
+        // as a live report, and a cold-attach mark nobody took yet is dropped
+        // rather than left to swallow that edge.
+        if let Ok(mut guard) = self.agent_session.lock() {
+            guard.replayed = false;
+        }
+
         self.exited_flag.store(false, Ordering::SeqCst);
         self.exited = false;
         {
@@ -891,6 +926,10 @@ impl RemoteTerminal {
                 remote: self.remote_context.clone(),
                 agent: self.agent.clone(),
                 agent_session: self.agent_session.clone(),
+                // The daemon does replay the stored status down this link, but
+                // the view already has a baseline from before the drop — see
+                // above.
+                awaiting_replay: false,
                 exited: self.exited_flag.clone(),
                 child_exited: self.child_exited.clone(),
                 zle_reading: self.zle_reading.clone(),
@@ -920,6 +959,20 @@ impl RemoteTerminal {
         Ok(())
     }
 
+    /// A pane the client *reattached* to rather than spawned — the shape the
+    /// app restores last session's tabs in, where the head of the stream is the
+    /// daemon replaying state the pane already had.
+    #[cfg(test)]
+    pub(super) fn from_stream_reattached(stream: Stream, size: TermSize) -> anyhow::Result<Self> {
+        Self::from_stream_parts(
+            stream,
+            size,
+            Vec::new(),
+            PtySource::for_route(&PaneRoute::Local),
+            true,
+        )
+    }
+
     /// A pane on a pty of this machine's own — what the tests build, and what
     /// `spawn_on` narrows with the route it dialled.
     pub(super) fn from_stream(stream: Stream, size: TermSize) -> anyhow::Result<Self> {
@@ -936,6 +989,22 @@ impl RemoteTerminal {
         size: TermSize,
         buffered: Vec<u8>,
         pty: PtySource,
+    ) -> anyhow::Result<Self> {
+        Self::from_stream_parts(stream, size, buffered, pty, false)
+    }
+
+    /// `awaiting_replay` says this link is an attach rather than a spawn, and
+    /// so that the frames at the head of its stream describe a pane that was
+    /// already running — see [`AgentSlot`]. It has to be decided here rather
+    /// than set on the returned terminal: the reader starts inside this
+    /// function, and against a daemon that answers promptly the replay can be
+    /// parsed before the caller gets its value back.
+    fn from_stream_parts(
+        stream: Stream,
+        size: TermSize,
+        buffered: Vec<u8>,
+        pty: PtySource,
+        awaiting_replay: bool,
     ) -> anyhow::Result<Self> {
         let read_half = stream.try_clone()?;
         let write_half = stream;
@@ -955,7 +1024,7 @@ impl RemoteTerminal {
         let shell_state: Arc<Mutex<ShellState>> = Arc::new(Mutex::new(ShellState::default()));
         let remote_context: Arc<Mutex<Option<RemoteContext>>> = Arc::new(Mutex::new(None));
         let agent: Arc<Mutex<Option<CLIAgent>>> = Arc::new(Mutex::new(None));
-        let agent_session: Arc<Mutex<Option<AgentSessionState>>> = Arc::new(Mutex::new(None));
+        let agent_session: Arc<Mutex<AgentSlot>> = Arc::new(Mutex::new(AgentSlot::default()));
         let exited_flag = Arc::new(AtomicBool::new(false));
         let child_exited = Arc::new(AtomicBool::new(false));
         let zle_reading = Arc::new(AtomicBool::new(false));
@@ -982,6 +1051,7 @@ impl RemoteTerminal {
                 remote: remote_context.clone(),
                 agent: agent.clone(),
                 agent_session: agent_session.clone(),
+                awaiting_replay,
                 exited: exited_flag.clone(),
                 child_exited: child_exited.clone(),
                 zle_reading: zle_reading.clone(),
@@ -1092,6 +1162,7 @@ impl RemoteTerminal {
                     remote,
                     agent,
                     agent_session,
+                    awaiting_replay,
                     exited: exited_flag,
                     child_exited,
                     zle_reading,
@@ -1104,6 +1175,7 @@ impl RemoteTerminal {
                     clipboard_write_busy,
                     repair_cursor,
                 } = signals;
+                let mut awaiting_replay = awaiting_replay;
                 crate::core::threads::promote_to_user_interactive();
                 let mut stream = read_half;
                 let mut processor: ansi::Processor = ansi::Processor::new();
@@ -1327,6 +1399,16 @@ impl RemoteTerminal {
                                 proxy.send_event(AlacEvent::Wakeup);
                             }
                             DaemonMsg::Output(bytes) => {
+                                // Live output only ever follows the whole
+                                // replay (the daemon sends the stored status
+                                // last, and the stream keeps that order), so
+                                // the first frame here ends the window in which
+                                // a status can still be a replayed one. Without
+                                // this, a pane that had no agent session to
+                                // replay would keep the window open until some
+                                // agent it ran *later* reported for the first
+                                // time, and that report would be discounted.
+                                awaiting_replay = false;
                                 out_batch.extend_from_slice(&bytes);
                                 tr_frames += 1;
                             }
@@ -1508,7 +1590,11 @@ impl RemoteTerminal {
                             DaemonMsg::AgentStatus(state) => {
                                 flush_batch!();
                                 if let Ok(mut guard) = agent_session.lock() {
-                                    *guard = state;
+                                    guard.state = state;
+                                    // The first such frame on an attached link
+                                    // is the pane's stored status being
+                                    // replayed, not a turn changing state now.
+                                    guard.replayed = std::mem::take(&mut awaiting_replay);
                                 }
                                 proxy.send_event(AlacEvent::Wakeup);
                             }
@@ -1729,7 +1815,22 @@ impl RemoteTerminal {
     }
 
     pub fn agent_session(&self) -> Option<AgentSessionState> {
-        self.agent_session.lock().ok().and_then(|g| g.clone())
+        self.agent_session.lock().ok().and_then(|g| g.state.clone())
+    }
+
+    /// The status the daemon replayed when this link attached, handed out once.
+    ///
+    /// `Some(status)` means what [`Self::agent_session`] reports right now is
+    /// stored state from before this client existed — the caller should take it
+    /// as its starting point, not as something that just happened. Answering
+    /// only once is what keeps the very next live transition an edge again.
+    pub fn take_replayed_agent_status(&self) -> Option<Option<AgentStatus>> {
+        let mut guard = self.agent_session.lock().ok()?;
+        if !guard.replayed {
+            return None;
+        }
+        guard.replayed = false;
+        Some(guard.state.as_ref().map(|s| s.status))
     }
 
     pub fn zle_reading(&self) -> bool {
@@ -5604,6 +5705,118 @@ mod tests {
         assert!(poll(None), "agent exit should clear it");
     }
 
+    /// The daemon replays a reattached pane's stored agent status as an
+    /// ordinary report. Nothing on the wire says so, so the link has to
+    /// remember that the first report it hears is that replay — and that
+    /// everything after it is live.
+    #[test]
+    fn a_reattached_link_marks_only_its_first_status_report_as_replayed() {
+        use crate::core::cli_agent::{AgentSessionState, AgentStatus};
+
+        crate::core::config::pin_test_config_dir();
+        let (client_side, mut daemon_side) = UnixStream::pair().unwrap();
+        let term =
+            RemoteTerminal::from_stream_reattached(client_side, TermSize::new(80, 24)).unwrap();
+        assert_eq!(
+            term.take_replayed_agent_status(),
+            None,
+            "nothing replayed until the frame actually arrives"
+        );
+
+        let report = |status, daemon: &mut UnixStream| {
+            DaemonMsg::AgentStatus(Some(AgentSessionState {
+                status,
+                message: None,
+                session_id: Some("sid-1".into()),
+                launch_argv: None,
+                rich: true,
+                cwd: None,
+                activity: 0,
+                turns: 0,
+            }))
+            .encode(daemon)
+            .unwrap();
+            daemon.flush().unwrap();
+        };
+        let poll = |want: AgentStatus| {
+            for _ in 0..200 {
+                if term.agent_session().map(|s| s.status) == Some(want) {
+                    return true;
+                }
+                std::thread::sleep(std::time::Duration::from_millis(5));
+            }
+            false
+        };
+
+        report(AgentStatus::Done, &mut daemon_side);
+        assert!(poll(AgentStatus::Done), "the replayed status should land");
+        assert_eq!(
+            term.take_replayed_agent_status(),
+            Some(Some(AgentStatus::Done)),
+            "the first report on a reattached link is stored state"
+        );
+        assert_eq!(
+            term.take_replayed_agent_status(),
+            None,
+            "only one taker gets it"
+        );
+
+        report(AgentStatus::Working, &mut daemon_side);
+        assert!(poll(AgentStatus::Working), "the live status should land");
+        assert_eq!(
+            term.take_replayed_agent_status(),
+            None,
+            "everything after the replay is something the client watched happen"
+        );
+    }
+
+    /// A pane with no agent session to replay sends no status frame at all, so
+    /// the replay window has to close on its own — otherwise the first report
+    /// from an agent launched *later* would be discounted as stored state.
+    #[test]
+    fn live_output_closes_the_replay_window() {
+        use crate::core::cli_agent::{AgentSessionState, AgentStatus};
+
+        crate::core::config::pin_test_config_dir();
+        let (client_side, mut daemon_side) = UnixStream::pair().unwrap();
+        let term =
+            RemoteTerminal::from_stream_reattached(client_side, TermSize::new(80, 24)).unwrap();
+
+        DaemonMsg::Output(b"$ claude\r\n".to_vec())
+            .encode(&mut daemon_side)
+            .unwrap();
+        DaemonMsg::AgentStatus(Some(AgentSessionState {
+            status: AgentStatus::Done,
+            message: None,
+            session_id: Some("sid-1".into()),
+            launch_argv: None,
+            rich: true,
+            cwd: None,
+            activity: 0,
+            turns: 0,
+        }))
+        .encode(&mut daemon_side)
+        .unwrap();
+        daemon_side.flush().unwrap();
+
+        for _ in 0..200 {
+            if term.agent_session().is_some() {
+                break;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(5));
+        }
+        assert_eq!(
+            term.agent_session().map(|s| s.status),
+            Some(AgentStatus::Done),
+            "the status still lands"
+        );
+        assert_eq!(
+            term.take_replayed_agent_status(),
+            None,
+            "a report that follows live output is live"
+        );
+    }
+
     #[test]
     fn agent_session_follows_daemon_status_reports() {
         use crate::core::cli_agent::{AgentSessionState, AgentStatus};

src/terminal/view.rs

@@ -207,6 +207,11 @@ impl gpui::Global for AgentReadMarks {}
 #[derive(Clone)]
 struct AgentReadMark {
     session: (Option<String>, Option<Vec<String>>),
+    /// The status the pane's last view saw. Kept for every status, not only
+    /// `Done`, so that a pane with no mark at all is one this app never watched
+    /// an agent in — the only case where a reattach's replayed status may be
+    /// taken as a baseline (see `poll_agent_status`).
+    status: Option<crate::core::cli_agent::AgentStatus>,
     turns: u64,
     unread: bool,
 }
@@ -1903,23 +1908,20 @@ impl TerminalView {
     }
 
     /// Leave what the reader has seen of this pane's agent where the pane's
-    /// next view will look for it — see [`AgentReadMarks`]. Anything but a
-    /// finished turn drops the mark: whatever finishes next is news.
+    /// next view will look for it — see [`AgentReadMarks`]. A mark left at any
+    /// status other than `Done` never vouches for a finished turn, but it still
+    /// records that this app was watching: a turn that was running when the
+    /// view went and finished before the next one came must badge.
     fn record_agent_read_mark(&self, turns: u64, cx: &mut App) {
-        let key = (self.host_id, self.pane_id);
-        let marks = &mut cx.default_global::<AgentReadMarks>().0;
-        if self.last_agent_status == Some(crate::core::cli_agent::AgentStatus::Done) {
-            marks.insert(
-                key,
-                AgentReadMark {
-                    session: self.last_agent_session.clone(),
-                    turns,
-                    unread: self.agent_result_unread,
-                },
-            );
-        } else {
-            marks.remove(&key);
-        }
+        cx.default_global::<AgentReadMarks>().0.insert(
+            (self.host_id, self.pane_id),
+            AgentReadMark {
+                session: self.last_agent_session.clone(),
+                status: self.last_agent_status,
+                turns,
+                unread: self.agent_result_unread,
+            },
+        );
     }
 
     /// Carry a change to the badge alone into the mark the last status left.
@@ -3956,6 +3958,33 @@ impl TerminalView {
     ) -> bool {
         use crate::core::cli_agent::AgentStatus;
 
+        // Attaching to a pane the daemon kept alive — the app restarting onto
+        // last session's tabs above all — has the daemon replay the pane's
+        // stored agent status as an ordinary report. When this app has never
+        // watched the pane, that is a baseline, not an edge: the turn it
+        // describes ended before this view existed, often before this process
+        // did, and reading it as "a result just landed" is what used to bring
+        // every restored agent tab up wearing an unread badge for output its
+        // reader had long since read.
+        //
+        // When an earlier view of this app did watch the pane (a workspace
+        // switched out and back, a window reopened from the tray), its read
+        // mark knows more than the replay does, so the replay stays an edge and
+        // the mark decides below: the same finished turn takes its badge back
+        // as the reader left it, anything else is news (#870).
+        let adopted_baseline = match self.terminal.take_replayed_agent_status() {
+            Some(restored)
+                if !cx
+                    .try_global::<AgentReadMarks>()
+                    .is_some_and(|marks| marks.0.contains_key(&(self.host_id, self.pane_id))) =>
+            {
+                self.last_agent_status = restored;
+                self.agent_status_seen = true;
+                true
+            }
+            _ => false,
+        };
+
         let session = self.terminal.agent_session();
         if session.as_ref().is_some_and(|s| s.rich) {
             self.agent_was_rich = true;
@@ -3974,12 +4003,17 @@ impl TerminalView {
         }
 
         let status = session.as_ref().map(|s| s.status);
+        let turns = session.as_ref().map_or(0, |s| s.turns);
+        if adopted_baseline {
+            // From here on this app is watching the pane, so a later rebuild
+            // must find a mark rather than adopt its own replay.
+            self.record_agent_read_mark(turns, cx);
+        }
         if status == self.last_agent_status {
             return false;
         }
         let prev = std::mem::replace(&mut self.last_agent_status, status);
         let first_sight = !std::mem::replace(&mut self.agent_status_seen, true);
-        let turns = session.as_ref().map_or(0, |s| s.turns);
 
         // A view built over a pane that already holds a finished turn sees
         // `Done` arrive from nothing, the same as a turn finishing now. If the
@@ -3992,7 +4026,11 @@ impl TerminalView {
             && let Some(mark) = cx
                 .try_global::<AgentReadMarks>()
                 .and_then(|marks| marks.0.get(&(self.host_id, self.pane_id)))
-                .filter(|mark| mark.session == self.last_agent_session && mark.turns == turns)
+                .filter(|mark| {
+                    mark.status == Some(AgentStatus::Done)
+                        && mark.session == self.last_agent_session
+                        && mark.turns == turns
+                })
                 .cloned()
         {
             self.agent_result_unread = mark.unread && !self.focus_handle.is_focused(window);
@@ -9789,6 +9827,22 @@ pub(crate) fn quiet_test_pane(
     (view, daemon_side)
 }
 
+/// The same pane, but reattached rather than spawned — what restoring last
+/// session's tabs builds, and the only shape in which the daemon replays state
+/// the pane already had.
+#[cfg(test)]
+pub(crate) fn quiet_reattached_test_pane(
+    pane_id: u64,
+    window: &mut Window,
+    cx: &mut gpui::App,
+) -> (gpui::Entity<TerminalView>, crate::daemon::transport::Stream) {
+    let (client_side, daemon_side) = test_stream_pair();
+    let terminal = RemoteTerminal::from_stream_reattached(client_side, TermSize::new(80, 24))
+        .expect("quiet reattached test terminal");
+    let view = cx.new(|cx| TerminalView::with_terminal(terminal, pane_id, window, cx));
+    (view, daemon_side)
+}
+
 /// A quiet pane that was dialled by hand, with no saved host behind it.
 ///
 /// Ungated on purpose: the transport this hands back is already
@@ -10301,6 +10355,197 @@ mod gpui_tests {
             .unwrap();
     }
 
+    /// Reinstalling or restarting the app leaves the daemon — and every agent
+    /// in it — running, so each restored tab reattaches to a pane whose agent
+    /// finished its turn long ago. The daemon replays that status, and reading
+    /// it as a turn that just landed put an unread badge on every agent tab in
+    /// the window the moment it opened.
+    #[gpui::test]
+    fn a_restored_pane_does_not_badge_the_turn_it_reattached_to(cx: &mut TestAppContext) {
+        use crate::core::cli_agent::AgentStatus;
+
+        crate::core::config::pin_test_config_dir();
+        let (window, _root_daemon) = harness(cx);
+        let (pane, mut daemon) = window
+            .update(cx, |_, window, cx| {
+                super::quiet_reattached_test_pane(2, window, cx)
+            })
+            .unwrap();
+        window
+            .update(cx, |view, window, cx| {
+                view.focus_handle.clone().focus(window, cx);
+            })
+            .unwrap();
+        cx.run_until_parked();
+
+        report_agent_status(AgentStatus::Done, &pane, cx, &mut daemon);
+        window
+            .update(cx, |_, window, cx| {
+                pane.update(cx, |pane, cx| {
+                    pane.poll_agent_status(false, window, cx);
+                    assert!(
+                        !pane.agent_result_unread(),
+                        "the replayed status is where this pane starts, not a result that \
+                         just arrived"
+                    );
+                });
+            })
+            .unwrap();
+
+        // And the pane is still armed: the next turn it actually watches finish
+        // badges exactly as it would have without the reattach.
+        report_agent_status(AgentStatus::Working, &pane, cx, &mut daemon);
+        window
+            .update(cx, |_, window, cx| {
+                pane.update(cx, |pane, cx| {
+                    pane.poll_agent_status(false, window, cx);
+                });
+            })
+            .unwrap();
+        report_agent_status(AgentStatus::Done, &pane, cx, &mut daemon);
+        window
+            .update(cx, |_, window, cx| {
+                pane.update(cx, |pane, cx| {
+                    pane.poll_agent_status(false, window, cx);
+                    assert!(
+                        pane.agent_result_unread(),
+                        "a turn that finished while the reader was elsewhere is unread"
+                    );
+                });
+            })
+            .unwrap();
+    }
+
+    /// Build `pane_id`'s first view (unfocused), let it watch `status` at
+    /// `turns`, then rebuild the pane the way restoring a workspace does — a
+    /// reattach, whose head is the daemon replaying `replayed` — and read the
+    /// rebuilt view's badge.
+    fn rebuild_by_reattach(
+        watched: (crate::core::cli_agent::AgentStatus, u64),
+        replayed: (crate::core::cli_agent::AgentStatus, u64),
+        cx: &mut TestAppContext,
+    ) -> bool {
+        let (window, _root_daemon) = harness(cx);
+        let focus_root = |cx: &mut TestAppContext| {
+            window
+                .update(cx, |view, window, cx| {
+                    view.focus_handle.clone().focus(window, cx)
+                })
+                .unwrap();
+            cx.run_until_parked();
+        };
+        let (before, mut before_daemon) = window
+            .update(cx, |_, window, cx| super::quiet_test_pane(2, window, cx))
+            .unwrap();
+        focus_root(cx);
+        report_agent_turn(watched.0, watched.1, &before, cx, &mut before_daemon);
+        poll_unread(window, &before, cx);
+        drop(before);
+
+        let (after, mut daemon) = window
+            .update(cx, |_, window, cx| {
+                super::quiet_reattached_test_pane(2, window, cx)
+            })
+            .unwrap();
+        focus_root(cx);
+        report_agent_turn(replayed.0, replayed.1, &after, cx, &mut daemon);
+        poll_unread(window, &after, cx)
+    }
+
+    /// A reattach whose pane this app already watched is a rebuild, not a
+    /// restart: the read mark, not the replay, says whether the turn is news.
+    #[gpui::test]
+    fn a_reattached_pane_takes_back_the_badge_its_reader_left(cx: &mut TestAppContext) {
+        use crate::core::cli_agent::AgentStatus;
+        assert!(
+            rebuild_by_reattach((AgentStatus::Done, 1), (AgentStatus::Done, 1), cx),
+            "the reader never cleared that badge; rebuilding the pane is not reading it"
+        );
+    }
+
+    #[gpui::test]
+    fn a_reattached_pane_badges_a_turn_that_was_running_when_its_view_went(
+        cx: &mut TestAppContext,
+    ) {
+        use crate::core::cli_agent::AgentStatus;
+        assert!(
+            rebuild_by_reattach((AgentStatus::Working, 0), (AgentStatus::Done, 1), cx),
+            "nobody saw this turn finish"
+        );
+    }
+
+    #[gpui::test]
+    fn a_reattached_pane_badges_a_later_turn_that_finished_while_away(cx: &mut TestAppContext) {
+        use crate::core::cli_agent::AgentStatus;
+        // Turn one left a mark; turn two finishing before the reattach is a
+        // different count, so the mark does not vouch for it.
+        assert!(
+            rebuild_by_reattach((AgentStatus::Done, 1), (AgentStatus::Done, 2), cx),
+            "the second turn finished unseen"
+        );
+    }
+
+    /// A relink keeps the view and what it last saw. A turn that was running
+    /// when the link dropped and finished before it came back reaches the view
+    /// only as the daemon's replay, and that replay has to badge: nobody saw
+    /// the turn finish.
+    #[gpui::test]
+    fn a_turn_that_finished_while_the_link_was_down_still_badges(cx: &mut TestAppContext) {
+        use crate::core::cli_agent::AgentStatus;
+
+        crate::core::config::pin_test_config_dir();
+        let (window, _root_daemon) = harness(cx);
+        let (pane, mut daemon) = window
+            .update(cx, |_, window, cx| {
+                super::quiet_reattached_test_pane(2, window, cx)
+            })
+            .unwrap();
+        window
+            .update(cx, |view, window, cx| {
+                view.focus_handle.clone().focus(window, cx);
+            })
+            .unwrap();
+        cx.run_until_parked();
+
+        report_agent_status(AgentStatus::Working, &pane, cx, &mut daemon);
+        window
+            .update(cx, |_, window, cx| {
+                pane.update(cx, |pane, cx| {
+                    pane.poll_agent_status(false, window, cx);
+                    assert!(!pane.agent_result_unread(), "the turn is still running");
+                });
+            })
+            .unwrap();
+
+        let (new_client, mut new_daemon) = super::test_stream_pair();
+        pane.update(cx, |pane, cx| {
+            pane.adopt_relink(
+                new_client,
+                Vec::new(),
+                &crate::terminal::PaneRoute::Local,
+                TermSize::new(80, 24),
+                8,
+                17,
+                cx,
+            )
+            .expect("the swap itself cannot fail");
+        });
+        drop(daemon);
+
+        report_agent_status(AgentStatus::Done, &pane, cx, &mut new_daemon);
+        window
+            .update(cx, |_, window, cx| {
+                pane.update(cx, |pane, cx| {
+                    pane.poll_agent_status(false, window, cx);
+                    assert!(
+                        pane.agent_result_unread(),
+                        "the turn finished while the link was down, so nobody read it"
+                    );
+                });
+            })
+            .unwrap();
+    }
+
     #[gpui::test]
     fn a_finished_turn_on_the_focused_pane_is_already_read(cx: &mut TestAppContext) {
         use crate::core::cli_agent::AgentStatus;