diff --git a/src/terminal/remote.rs b/src/terminal/remote.rs index 4e561763..ec225a46 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) } @@ -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::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 +989,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 +1024,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 +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 { - 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 +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}; diff --git a/src/terminal/view.rs b/src/terminal/view.rs index e783728e..1c82402e 100644 --- a/src/terminal/view.rs +++ b/src/terminal/view.rs @@ -207,6 +207,11 @@ impl gpui::Global for AgentReadMarks {} #[derive(Clone)] struct AgentReadMark { session: (Option, Option>), + /// 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, 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::().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::().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::() + .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::() .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, 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;