From 28360107c1bfab57fb0de99d74bfa48e74423ca8 Mon Sep 17 00:00:00 2001 From: Can Celik Date: Mon, 21 Sep 2026 20:47:14 +0300 Subject: fix: distinguish agent completion from startup and session changes (#4457) refs #3338 refs #3990 refs #3916 --- src/server/client_shell.rs | 28 +- src/server/clients.rs | 2 + src/server/headless.rs | 49 ++-- src/server/headless/notifications.rs | 80 ++---- src/server/headless/render.rs | 19 +- src/server/headless/tests/mod.rs | 376 ++++++++++++++++++++++---- src/server/headless/tests/pane_graphics.rs | 25 +- src/server/headless/tests/pane_move.rs | 4 +- src/server/headless/tests/surface_interest.rs | 23 +- src/server/notifications.rs | 6 +- src/server/render_scale_benchmark.rs | 29 +- 11 files changed, 467 insertions(+), 174 deletions(-) (limited to 'src/server') 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, /// Last coherent shell replacement sent to this client. pub(crate) shell_snapshot: Option, + pub(crate) shell_agent_completions: Option, /// View policy paired with the last coherent shell replacement. pub(crate) shell_agent_view: Option, /// 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 { - 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, ) -> 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 { - let ServerMessage::EndpointControl { kind, data } = message else { - panic!("expected client shell snapshot"); +fn client_shell_projection( + receiver: &std::sync::mpsc::Receiver>, +) -> ( + Box, + 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 = + 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>, +) -> Box { + client_shell_projection(receiver).0 } fn client_agent_view_projection( @@ -714,6 +738,64 @@ async fn client_shell_attach_seeds_workspace() { shutdown_test_runtimes(&mut server); } +#[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(); @@ -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::(&response).expect("successful report"); +} + +fn completion_guard_notifications( + server: &mut HeadlessServer, + receiver: &std::sync::mpsc::Receiver>, +) -> Vec { + 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 { 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) { - 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 { -- cgit v1.2.3