From eb5aae172d953ab53671aa869bc408181727deb6 Mon Sep 17 00:00:00 2001 From: l0ng-ai <24760907+l0ng-ai@users.noreply.github.com> Date: Tue, 15 Sep 2026 09:01:04 +0800 Subject: [PATCH] fix(agents): stop a reattach from badging every restored agent tab MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restarting the app leaves the daemon and every agent in it running, so each restored tab attaches to a pane whose turn ended long ago. The daemon replays that pane's stored agent status as an ordinary `AgentStatus` frame, and nothing on the wire said it was stored state — so the view read `None -> Done` as a turn finishing right then and put an unread badge on every agent tab that was not the focused one, for results the reader had already read. A relink did the same on every dropped link. Keep the distinction where the value lives: the shared slot now carries `replayed` alongside the status, under the same lock, so a reader cannot observe one without the other. The reader thread marks the first status frame on an attached link as the replay; the first frame of live output closes that window, which is what a pane with no session to replay needs (its next agent's first report is live, and must stay an edge). The view takes that mark once and adopts the status as its baseline instead of an edge. The notification path already required the previous status to be Working/Waiting, so only the badge was wrong. Claude-Session: https://claude.ai/code/session_01FG2s9mbZu6LbjjmU54X7kt --- src/terminal/remote.rs | 214 +++++++++++++++++++++++++++++++++++++++-- src/terminal/view.rs | 88 +++++++++++++++++ 2 files changed, 295 insertions(+), 7 deletions(-) diff --git a/src/terminal/remote.rs b/src/terminal/remote.rs index be4cee3e..3ba287ae 100644 --- a/src/terminal/remote.rs +++ b/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, + /// `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>>, shell: Arc>, remote: Arc>>, agent: Arc>>, - agent_session: Arc>>, + agent_session: Arc>, + /// 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, child_exited: Arc, zle_reading: Arc, @@ -571,7 +594,7 @@ pub struct RemoteTerminal { ssh_user: Option, auto_supplied_password: bool, agent: Arc>>, - agent_session: Arc>>, + agent_session: Arc>, /// 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) } @@ -891,6 +915,10 @@ impl RemoteTerminal { remote: self.remote_context.clone(), agent: self.agent.clone(), agent_session: self.agent_session.clone(), + // A relink attaches to the pane all over again, so the daemon + // replays its stored agent status down the new link just as it + // does on a cold attach. + awaiting_replay: true, exited: self.exited_flag.clone(), child_exited: self.child_exited.clone(), zle_reading: self.zle_reading.clone(), @@ -920,6 +948,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::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 { @@ -936,6 +978,22 @@ impl RemoteTerminal { size: TermSize, buffered: Vec, pty: PtySource, + ) -> anyhow::Result { + 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, + pty: PtySource, + awaiting_replay: bool, ) -> anyhow::Result { let read_half = stream.try_clone()?; let write_half = stream; @@ -955,7 +1013,7 @@ impl RemoteTerminal { let shell_state: Arc> = Arc::new(Mutex::new(ShellState::default())); let remote_context: Arc>> = Arc::new(Mutex::new(None)); let agent: Arc>> = Arc::new(Mutex::new(None)); - let agent_session: Arc>> = Arc::new(Mutex::new(None)); + let agent_session: Arc> = 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 +1040,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 +1151,7 @@ impl RemoteTerminal { remote, agent, agent_session, + awaiting_replay, exited: exited_flag, child_exited, zle_reading, @@ -1104,6 +1164,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 +1388,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 +1579,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 +1804,22 @@ impl RemoteTerminal { } pub fn agent_session(&self) -> Option { - 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> { + 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 +5694,116 @@ 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, + })) + .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, + })) + .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}; diff --git a/src/terminal/view.rs b/src/terminal/view.rs index 3d77f506..a35b3d97 100644 --- a/src/terminal/view.rs +++ b/src/terminal/view.rs @@ -3887,6 +3887,17 @@ 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, a dropped link coming back — has the daemon + // replay the pane's stored agent status as an ordinary report. It 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. + if let Some(restored) = self.terminal.take_replayed_agent_status() { + self.last_agent_status = restored; + } + let session = self.terminal.agent_session(); if session.as_ref().is_some_and(|s| s.rich) { self.agent_was_rich = true; @@ -9695,6 +9706,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, 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 @@ -10039,6 +10066,67 @@ 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(); + } + #[gpui::test] fn a_finished_turn_on_the_focused_pane_is_already_read(cx: &mut TestAppContext) { use crate::core::cli_agent::AgentStatus;