From 9e9bc8a144667a6b7debfd4f0c7a31ff0e11493a Mon Sep 17 00:00:00 2001 From: Can Celik Date: Sun, 6 Sep 2026 02:27:06 +0300 Subject: feat: manage multiple ssh machines from one client (#3670) --- src/server/autodetect.rs | 46 +- src/server/client_commands.rs | 30 ++ src/server/client_endpoint_control.rs | 24 + src/server/client_shell_graphics.rs | 42 ++ src/server/client_transport.rs | 123 +++-- src/server/clients.rs | 13 +- src/server/headless.rs | 217 ++++---- src/server/headless/client_views.rs | 36 +- src/server/headless/endpoint_requests.rs | 130 +++++ src/server/headless/render.rs | 30 +- src/server/headless/retained_surface.rs | 113 +++- src/server/headless/surface_interest.rs | 86 +++ src/server/headless/tests/mod.rs | 68 ++- src/server/headless/tests/pane_graphics.rs | 247 ++++++++- src/server/headless/tests/surface_interest.rs | 725 ++++++++++++++++++++++++++ src/server/mod.rs | 1 + 16 files changed, 1717 insertions(+), 214 deletions(-) create mode 100644 src/server/client_endpoint_control.rs create mode 100644 src/server/headless/endpoint_requests.rs create mode 100644 src/server/headless/surface_interest.rs create mode 100644 src/server/headless/tests/surface_interest.rs (limited to 'src/server') diff --git a/src/server/autodetect.rs b/src/server/autodetect.rs index bd364a70..e6fdaaa4 100644 --- a/src/server/autodetect.rs +++ b/src/server/autodetect.rs @@ -142,7 +142,7 @@ fn client_protocol_accepts_hello(socket_path: &Path) -> io::Result { } } -fn validate_running_server_compatibility() -> io::Result<()> { +fn validate_running_server_compatibility(saved_federation: bool) -> io::Result<()> { let Some(status) = read_server_status()? else { return Err(io::Error::other(format!( "a herdr server is listening, but its status API is unavailable.\n\n{}\nIf that fails, stop the old server process manually.", @@ -150,16 +150,23 @@ fn validate_running_server_compatibility() -> io::Result<()> { ))); }; - let endpoint_generation = status - .capabilities - .as_ref() - .and_then(|capabilities| capabilities.endpoint_protocol_generation); - if endpoint_generation == Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION) { + let capabilities = status.capabilities.as_ref(); + let endpoint_generation = + capabilities.and_then(|capabilities| capabilities.endpoint_protocol_generation); + let surface_interest = capabilities.is_some_and(|capabilities| capabilities.surface_interest); + if endpoint_generation == Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION) + && (!saved_federation || surface_interest) + { return Ok(()); } + let requirement = if saved_federation && !surface_interest { + "saved SSH machines require surface lifecycle support" + } else { + "the stable endpoint generation is incompatible" + }; Err(io::Error::other(format!( - "This session predates Herdr's stable endpoint protocol and needs one final server update.\n\nserver: v{} endpoint generation {}\nclient: v{} endpoint generation {}\n\n{}", + "This session needs one final server update before Herdr can attach ({requirement}).\n\nserver: v{} endpoint generation {}\nclient: v{} endpoint generation {}\n\n{}", status.version.as_deref().unwrap_or("unknown"), endpoint_generation .map(|value| value.to_string()) @@ -285,18 +292,27 @@ pub fn wait_for_server_socket(socket_path: &Path, timeout: Duration) -> io::Resu /// 1. Check if a server is listening on the client socket /// 2. If no server → spawn server daemon → wait for socket readiness /// 3. Run the thin client (which connects to the server) -pub fn auto_detect_launch() -> io::Result<()> { +pub fn auto_detect_launch(saved_federation: bool) -> io::Result<()> { let socket_path = client_socket_path(); info!(path = %socket_path.display(), "auto-detect launch starting"); - if is_server_listening_at(&socket_path) { - validate_running_server_compatibility()?; + let startup = if is_server_listening_at(&socket_path) { info!("server already running, attaching as client"); + if saved_federation { + Ok(()) + } else { + validate_running_server_compatibility(false) + } } else { info!("no server running, spawning server daemon"); - spawn_server_daemon()?; - wait_for_server_socket(&socket_path, SERVER_READY_TIMEOUT)?; - info!("server ready, attaching as client"); + spawn_server_daemon() + .and_then(|_| wait_for_server_socket(&socket_path, SERVER_READY_TIMEOUT)) + }; + if let Err(error) = startup { + if !saved_federation { + return Err(error); + } + tracing::warn!(%error, "Local startup failed; keeping saved machines available"); } // Now attach as a thin client. @@ -522,7 +538,7 @@ test "$sid" = "$$" let path = dir.join("api.sock"); std::env::set_var(crate::api::SOCKET_PATH_ENV_VAR, &path); - let err = validate_running_server_compatibility().unwrap_err(); + let err = validate_running_server_compatibility(false).unwrap_err(); assert!( err.to_string().contains("status API is unavailable"), @@ -558,7 +574,7 @@ test "$sid" = "$$" stream.flush().unwrap(); }); - let err = validate_running_server_compatibility().unwrap_err(); + let err = validate_running_server_compatibility(false).unwrap_err(); let message = err.to_string(); let _ = handle.join(); diff --git a/src/server/client_commands.rs b/src/server/client_commands.rs index 34be0519..0af1fad1 100644 --- a/src/server/client_commands.rs +++ b/src/server/client_commands.rs @@ -13,6 +13,7 @@ pub(crate) const MAX_ENDPOINT_REQUEST_ID_BYTES: usize = 128; const ENDPOINT_RESPONSE_CHUNK_BYTES: usize = 512 * 1024; const CLIENT_SHELL_METHODS: &[&str] = &[ + "client_shell.surface.set", "command.invoke", "integration.install", "integration.list", @@ -77,6 +78,30 @@ pub(crate) fn error_response(id: String, code: &str, message: impl Into) }) } +pub(crate) fn success_message_with_result( + boot_id: String, + request_id: String, + result: crate::api::schema::ResponseResult, +) -> crate::protocol::ServerMessage { + let response = serde_json::to_string(&crate::api::schema::SuccessResponse { + id: request_id.clone(), + result, + }) + .unwrap_or_else(|_| { + error_response( + request_id.clone(), + "serialization_error", + "failed to serialize endpoint response", + ) + }); + crate::protocol::ServerMessage::ClientShellEndpointResponseChunk { + boot_id, + request_id, + final_chunk: true, + data: response.into_bytes(), + } +} + pub(crate) fn error_message( boot_id: String, request_id: String, @@ -312,6 +337,11 @@ mod tests { #[test] fn client_shell_lane_excludes_api_front_door_and_lifecycle_methods() { + assert!(supports_client_shell_method( + &Method::ClientShellSurfaceSet(crate::api::schema::ClientShellSurfaceSetParams { + active: false, + }) + )); assert!(supports_client_shell_method(&Method::ServerReloadConfig( crate::api::schema::EmptyParams::default(), ))); diff --git a/src/server/client_endpoint_control.rs b/src/server/client_endpoint_control.rs new file mode 100644 index 00000000..09970004 --- /dev/null +++ b/src/server/client_endpoint_control.rs @@ -0,0 +1,24 @@ +pub(crate) fn response(kind: &str, data: String) -> Option { + (kind == crate::protocol::endpoint::HEALTH_PING_KIND).then(|| { + crate::protocol::ServerMessage::EndpointControl { + kind: crate::protocol::endpoint::HEALTH_PONG_KIND.into(), + data, + } + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn health_ping_echoes_a_pong_without_entering_server_state() { + let pong = response(crate::protocol::endpoint::HEALTH_PING_KIND, "probe".into()).unwrap(); + assert!(matches!( + pong, + crate::protocol::ServerMessage::EndpointControl { kind, data } + if kind == crate::protocol::endpoint::HEALTH_PONG_KIND && data == "probe" + )); + assert!(response("future.control", String::new()).is_none()); + } +} diff --git a/src/server/client_shell_graphics.rs b/src/server/client_shell_graphics.rs index 578350b2..00355230 100644 --- a/src/server/client_shell_graphics.rs +++ b/src/server/client_shell_graphics.rs @@ -1,6 +1,48 @@ use crate::kitty_graphics::surface::DeliveryCache; use crate::protocol::{ClientShellPopupSurface, SurfaceGraphicsScene}; +pub(crate) fn collect_retained( + app: &crate::app::App, + surface: &crate::protocol::PaneSurfaceFrame, + target: crate::ui::TabSurfaceTarget, + cell_size: crate::kitty_graphics::HostCellSize, + delivered: &DeliveryCache, + client_id: u64, +) -> Option<(SurfaceGraphicsScene, DeliveryCache)> { + let rect = |rect: crate::protocol::SurfaceRect| { + ratatui::layout::Rect::new(rect.x, rect.y, rect.width, rect.height) + }; + let pane_infos = surface + .panes + .iter() + .map(|pane| { + let (workspace_index, id) = app.parse_pane_id(&pane.pane_id)?; + if workspace_index != target.workspace_index { + return None; + } + Some(crate::layout::PaneInfo { + id, + rect: rect(pane.rect), + inner_rect: rect(pane.inner_rect), + scrollbar_rect: pane.scrollbar_rect.map(rect), + borders: ratatui::widgets::Borders::NONE, + is_focused: pane.focused, + }) + }) + .collect::>>()?; + // Image clipping uses the retained content rectangles, not borders or split handles. + Some(collect( + app, + &pane_infos, + &[], + None, + Some(target), + cell_size, + delivered, + client_id, + )) +} + pub(crate) fn collect( app: &crate::app::App, pane_infos: &[crate::layout::PaneInfo], diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs index 25a963e7..c57d0f86 100644 --- a/src/server/client_transport.rs +++ b/src/server/client_transport.rs @@ -124,6 +124,11 @@ pub(crate) struct ClientWriter { } impl ClientWriter { + /// Drops render-lane work that has not yet been claimed by the writer. + pub(crate) fn discard_pending_render(&self) { + self.render.queue.discard_pending_render(); + } + #[cfg(test)] pub(crate) fn test_fill_render(&self, data: Vec) { self.render.try_send(data).unwrap(); @@ -304,6 +309,13 @@ impl ClientWriterQueue { Ok(()) } + fn discard_pending_render(&self) { + let mut state = self.lock_state(); + state.render = None; + state.ordered.clear(); + self.ready.notify_all(); + } + fn send_ordered(&self, data: Vec) -> Result<(), TrySendError>> { let mut state = self.lock_state(); if !state.writer_alive { @@ -382,6 +394,7 @@ pub(crate) enum ServerEvent { direct_graphics: bool, endpoint_keybindings: bool, mouse_capture: bool, + surface_active: bool, writer: ClientWriter, }, /// A client sent an input message. @@ -483,6 +496,8 @@ pub(crate) enum ServerEvent { ClientShellFocus { client_id: u64, focused: bool }, /// A client-owned shell updated its local mouse-capture preference. ClientShellMouseCapture { client_id: u64, enabled: bool }, + /// The committed shell asks the server to replay presentation effects before input resumes. + ClientShellPresentationSync { client_id: u64, token: String }, /// A client-owned shell invoked one endpoint operation through this connection. ClientShellEndpointRequest { client_id: u64, @@ -744,6 +759,7 @@ pub(crate) fn handle_client_handshake( hello.direct_graphics, hello.endpoint_keybindings, hello.mouse_capture, + hello.surface_active, )), ) } @@ -831,33 +847,39 @@ pub(crate) fn handle_client_handshake( } // Notify the main loop about the new client. - let connected = - if let Some((pixel_mouse, direct_graphics, endpoint_keybindings, mouse_capture)) = - shell_options - { - ServerEvent::ClientShellConnected { - client_id, - surface_cols: client_cols, - surface_rows: client_rows, - cell_width_px, - cell_height_px, - pixel_mouse, - direct_graphics, - endpoint_keybindings, - mouse_capture, - writer, - } - } else { - ServerEvent::ClientConnected { - client_id, - cols: client_cols, - rows: client_rows, - cell_width_px, - cell_height_px, - pixel_mouse: terminal_pixel_mouse, - writer, - } - }; + let endpoint_control_writer = shell_options.as_ref().map(|_| writer.control.clone()); + let connected = if let Some(( + pixel_mouse, + direct_graphics, + endpoint_keybindings, + mouse_capture, + surface_active, + )) = shell_options + { + ServerEvent::ClientShellConnected { + client_id, + surface_cols: client_cols, + surface_rows: client_rows, + cell_width_px, + cell_height_px, + pixel_mouse, + direct_graphics, + endpoint_keybindings, + mouse_capture, + surface_active, + writer, + } + } else { + ServerEvent::ClientConnected { + client_id, + cols: client_cols, + rows: client_rows, + cell_width_px, + cell_height_px, + pixel_mouse: terminal_pixel_mouse, + writer, + } + }; if let Err(err) = server_event_tx.blocking_send(connected) { match err.0 { ServerEvent::ClientConnected { writer, .. } @@ -869,7 +891,13 @@ pub(crate) fn handle_client_handshake( } // Enter read loop — read client messages and forward to main loop. - client_read_loop(stream, client_id, server_event_tx, should_quit) + client_read_loop_with_endpoint_controls( + stream, + client_id, + server_event_tx, + should_quit, + endpoint_control_writer.as_ref(), + ) } fn send_shutdown_to_unregistered_client(writer: &ClientWriter) { @@ -924,11 +952,22 @@ fn write_framed_bytes(stream: &mut LocalStream, data: &[u8]) -> bool { } /// The client read loop — reads messages from the client and forwards to the server event channel. +#[cfg(test)] fn client_read_loop( + stream: LocalStream, + client_id: u64, + server_event_tx: &mpsc::Sender, + should_quit: &Arc, +) -> io::Result<()> { + client_read_loop_with_endpoint_controls(stream, client_id, server_event_tx, should_quit, None) +} + +fn client_read_loop_with_endpoint_controls( mut stream: LocalStream, client_id: u64, server_event_tx: &mpsc::Sender, should_quit: &Arc, + endpoint_control_writer: Option<&ClientControlWriter>, ) -> io::Result<()> { while !should_quit.load(Ordering::Acquire) { let msg: ClientMessage = match protocol::read_message(&mut stream, MAX_GRAPHICS_FRAME_SIZE) @@ -1242,8 +1281,29 @@ fn client_read_loop( }, } } - ClientMessage::EndpointControl { kind, .. } => { - debug!(client_id, %kind, "ignoring unknown endpoint control message"); + ClientMessage::EndpointControl { kind, data } + if kind == crate::protocol::endpoint::PRESENTATION_EFFECTS_SYNC_KIND => + { + ServerEvent::ClientShellPresentationSync { + client_id, + token: data, + } + } + ClientMessage::EndpointControl { kind, data } => { + let Some(response) = crate::server::client_endpoint_control::response(&kind, data) + else { + debug!(client_id, %kind, "ignoring unknown endpoint control message"); + continue; + }; + let Some(writer) = endpoint_control_writer else { + continue; + }; + let mut framed = Vec::new(); + if protocol::write_message(&mut framed, &response).is_err() + || writer.send(framed).is_err() + { + break; + } continue; } ClientMessage::Detach => { @@ -1356,6 +1416,7 @@ mod tests { direct_graphics: true, endpoint_keybindings: true, mouse_capture: true, + surface_active: true, snapshot_codecs: vec![crate::protocol::endpoint::SNAPSHOT_CODEC_V1.into()], surface_codecs: vec![crate::protocol::endpoint::SURFACE_CODEC_V1.into()], input_codecs: vec![crate::protocol::endpoint::INPUT_CODEC_V1.into()], @@ -1767,6 +1828,7 @@ mod tests { direct_graphics, endpoint_keybindings, mouse_capture, + surface_active, writer, } => { assert_eq!(client_id, 43); @@ -1776,6 +1838,7 @@ mod tests { assert!(direct_graphics); assert!(endpoint_keybindings); assert!(mouse_capture); + assert!(surface_active); drop(writer); } other => panic!("expected ClientShellConnected, got {other:?}"), diff --git a/src/server/clients.rs b/src/server/clients.rs index 4f1baadb..972342ac 100644 --- a/src/server/clients.rs +++ b/src/server/clients.rs @@ -159,6 +159,8 @@ pub(crate) struct ClientConnection { pub(crate) host_keyboard_report_all_active: Option, /// Whether an ordinary render was skipped because the render channel was full. pub(crate) render_pending: bool, + /// Whether this connection receives pane surfaces and may affect presentation state. + pub(crate) shell_surface_active: bool, /// Whether this shell wants host mouse capture without pane demand. pub(crate) shell_mouse_capture: bool, /// Last host mouse capture mode sent to this client. @@ -179,6 +181,9 @@ pub(crate) struct ClientConnection { pub(crate) shell_projection_revision: u64, /// Whether this shell is waiting for one ordered endpoint command response. pub(crate) shell_endpoint_command_in_flight: bool, + /// Surface projection epoch that owned the in-flight command. Deferred navigation may run + /// only if this exact presentation lease is still active when its response arrives. + pub(crate) shell_endpoint_command_surface_revision: Option, /// Request id and buffered response for a deferred worktree-created navigation. pub(crate) shell_deferred_navigation_request_id: Option, pub(crate) shell_deferred_navigation_response: Option>, @@ -230,6 +235,7 @@ impl ClientConnection { outer_terminal_focus: None, host_keyboard_report_all_active: None, render_pending: false, + shell_surface_active: true, shell_mouse_capture: false, host_mouse_capture_active: None, host_sgr_pixels_active: None, @@ -240,6 +246,7 @@ impl ClientConnection { shell_snapshot: None, shell_projection_revision: 0, shell_endpoint_command_in_flight: false, + shell_endpoint_command_surface_revision: None, shell_deferred_navigation_request_id: None, shell_deferred_navigation_response: None, shell_uses_endpoint_keybindings: false, @@ -451,12 +458,16 @@ impl ClientConnection { pub(crate) fn is_shell_client(&self) -> bool { matches!(self.mode, ClientConnectionMode::ClientShell) } + + pub(crate) fn is_active_shell_client(&self) -> bool { + self.is_shell_client() && self.shell_surface_active + } } pub(crate) fn latest_shell_client(clients: &HashMap) -> Option { clients .iter() - .filter(|(_, client)| client.is_shell_client()) + .filter(|(_, client)| client.is_active_shell_client()) .max_by_key(|(_, client)| client.last_activity) .map(|(&client_id, _)| client_id) } diff --git a/src/server/headless.rs b/src/server/headless.rs index 04a1b17b..abaca0aa 100644 --- a/src/server/headless.rs +++ b/src/server/headless.rs @@ -73,11 +73,13 @@ use crate::server::terminal_attach::paste_payload_for_runtime; mod bootstrap; mod client_views; +mod endpoint_requests; mod lifecycle; mod notifications; mod pane_graphics; mod render; mod retained_surface; +mod surface_interest; pub use bootstrap::run_server; use lifecycle::wait_for_live_handoff_response_write; @@ -910,6 +912,9 @@ impl HeadlessServer { let Some(client) = self.clients.get_mut(&client_id) else { return false; }; + if !client.shell_surface_active { + return false; + } client.last_activity = stamp; let changed = self.foreground_client_id != Some(client_id); @@ -929,13 +934,13 @@ impl HeadlessServer { fn app_client_count(&self) -> usize { self.clients .values() - .filter(|client| client.is_shell_client() && client.writer.is_some()) + .filter(|client| client.is_active_shell_client() && client.writer.is_some()) .count() } fn client_supports_direct_graphics(&self, client_id: u64) -> bool { self.clients.get(&client_id).is_some_and(|client| { - client.is_shell_client() + client.is_active_shell_client() && client.writer.is_some() && client.direct_graphics && client.pixel_mouse @@ -992,12 +997,14 @@ impl HeadlessServer { let disconnected_focus = self .clients .get(&client_id) - .filter(|client| client.is_shell_client() && client.outer_terminal_focus == Some(true)) + .filter(|client| { + client.is_active_shell_client() && client.outer_terminal_focus == Some(true) + }) .and_then(|_| self.shell_focus_target(client_id)); let should_release_focus = disconnected_focus.as_ref().is_some_and(|target| { !self.clients.iter().any(|(&other_id, client)| { other_id != client_id - && client.is_shell_client() + && client.is_active_shell_client() && client.outer_terminal_focus == Some(true) && self.shell_tab_id_for_client(other_id).as_deref() == Some(target.tab_id.as_str()) @@ -1008,7 +1015,8 @@ impl HeadlessServer { self.tab_geometry_controllers .retain(|_, controller_id| *controller_id != client_id); if let Some(mut removed) = removed { - self.release_client_shell_inputs(client_id, &mut removed); + let held_inputs = removed.drain_shell_held_inputs(); + self.release_client_shell_inputs(client_id, held_inputs); crate::server::clipboard_image::remove_files(removed.staged_clipboard_files); if let ClientConnectionMode::TerminalAttach { terminal_id } = removed.mode { self.terminal_attach_owners.remove(&terminal_id); @@ -1033,8 +1041,12 @@ impl HeadlessServer { } } - fn release_client_shell_inputs(&mut self, client_id: u64, client: &mut ClientConnection) { - for held in client.drain_shell_held_inputs() { + fn release_client_shell_inputs( + &mut self, + client_id: u64, + held_inputs: Vec, + ) { + for held in held_inputs { let result = match held.target { ClientShellInputTarget::Pane(pane_id) => { let Some((workspace_index, runtime_pane_id)) = self.app.parse_pane_id(&pane_id) @@ -1181,16 +1193,18 @@ impl HeadlessServer { protocol::ClientClipboardImageTarget::Pane(pane_id) => { !self.handoff_in_progress && self.app.state.popup_pane.is_none() - && self.clients.get(&client_id).is_some_and(|client| { - matches!(client.mode, ClientConnectionMode::ClientShell) - }) + && self + .clients + .get(&client_id) + .is_some_and(ClientConnection::is_active_shell_client) && self.app.parse_pane_id(pane_id).is_some() } protocol::ClientClipboardImageTarget::Popup(terminal_id) => { !self.handoff_in_progress - && self.clients.get(&client_id).is_some_and(|client| { - matches!(client.mode, ClientConnectionMode::ClientShell) - }) + && self + .clients + .get(&client_id) + .is_some_and(ClientConnection::is_active_shell_client) && self .app .state @@ -1237,9 +1251,10 @@ impl HeadlessServer { } protocol::ClientClipboardImageTarget::Pane(pane_id) => { if self.handoff_in_progress - || !self.clients.get(&client_id).is_some_and(|client| { - matches!(client.mode, ClientConnectionMode::ClientShell) - }) + || !self + .clients + .get(&client_id) + .is_some_and(ClientConnection::is_active_shell_client) { return false; } @@ -1273,9 +1288,10 @@ impl HeadlessServer { } protocol::ClientClipboardImageTarget::Popup(terminal_id) => { if self.handoff_in_progress - || !self.clients.get(&client_id).is_some_and(|client| { - matches!(client.mode, ClientConnectionMode::ClientShell) - }) + || !self + .clients + .get(&client_id) + .is_some_and(ClientConnection::is_active_shell_client) { return false; } @@ -1929,6 +1945,7 @@ impl HeadlessServer { direct_graphics, endpoint_keybindings, mouse_capture, + surface_active, writer, } => { if self.handoff_in_progress { @@ -1950,6 +1967,7 @@ impl HeadlessServer { rows = surface_rows, cell_width_px, cell_height_px, + surface_active, render_encoding = ?protocol::RenderEncoding::SemanticFrame, "client connected" ); @@ -1972,6 +1990,7 @@ impl HeadlessServer { connection.direct_graphics = direct_graphics; connection.shell_uses_endpoint_keybindings = endpoint_keybindings; connection.shell_mouse_capture = mouse_capture; + connection.shell_surface_active = surface_active; connection.shell_projection_revision = 1; let config_diagnostic = if endpoint_keybindings { self.server_config_diagnostic.as_deref() @@ -2002,7 +2021,9 @@ impl HeadlessServer { self.popup_owner_tab_id = self.shell_tab_id_for_client(client_id); } self.send_to_client(client_id, snapshot_message); - self.foreground_client_id = Some(client_id); + if surface_active { + self.foreground_client_id = Some(client_id); + } if first_app_client { self.app.mark_git_status_refresh_due(Instant::now()); } @@ -2224,6 +2245,9 @@ impl HeadlessServer { client.cell_size = observed; } client.pixel_mouse = pixel_mouse && observed.is_known(); + if !client.shell_surface_active { + return false; + } client.request_repaint(); self.promote_client_to_foreground(client_id); self.resize_shell_tab_if_controller(client_id, true); @@ -2239,7 +2263,7 @@ impl HeadlessServer { if !client.update_host_theme(&update) { return false; } - if self.foreground_client_id != Some(client_id) { + if !client.shell_surface_active || self.foreground_client_id != Some(client_id) { return false; } let mut changed = self.app.set_host_terminal_appearance_state( @@ -2256,15 +2280,14 @@ impl HeadlessServer { let Some(client) = self.clients.get(&client_id) else { return false; }; - if !matches!(client.mode, ClientConnectionMode::ClientShell) - || client.outer_terminal_focus == Some(focused) + if !client.is_active_shell_client() || client.outer_terminal_focus == Some(focused) { return false; } let tab_id = self.shell_tab_id_for_client(client_id); let another_focused_viewer = self.clients.iter().any(|(&other_id, client)| { other_id != client_id - && client.is_shell_client() + && client.is_active_shell_client() && client.outer_terminal_focus == Some(true) && self.shell_tab_id_for_client(other_id) == tab_id }); @@ -2308,15 +2331,38 @@ impl HeadlessServer { client.host_mouse_capture_active = None; true } + ServerEvent::ClientShellPresentationSync { client_id, token } => { + let Some(client) = self.clients.get_mut(&client_id) else { + return false; + }; + if !client.is_active_shell_client() { + return false; + } + client.host_mouse_capture_active = None; + client.host_sgr_pixels_active = None; + client.host_keyboard_report_all_active = None; + self.sent_window_title = None; + self.stream_host_mouse_capture_mode(); + self.stream_direct_terminal_keyboard_mode(); + self.sync_window_title(); + self.send_to_client( + client_id, + ServerMessage::EndpointControl { + kind: crate::protocol::endpoint::PRESENTATION_EFFECTS_READY_KIND.into(), + data: token, + }, + ) + } ServerEvent::ClientShellPaneInput { client_id, pane_id, events, } => { if self.handoff_in_progress - || !self.clients.get(&client_id).is_some_and(|client| { - matches!(client.mode, ClientConnectionMode::ClientShell) - }) + || !self + .clients + .get(&client_id) + .is_some_and(ClientConnection::is_active_shell_client) { return false; } @@ -2400,9 +2446,10 @@ impl HeadlessServer { events, } => { if self.handoff_in_progress - || !self.clients.get(&client_id).is_some_and(|client| { - matches!(client.mode, ClientConnectionMode::ClientShell) - }) + || !self + .clients + .get(&client_id) + .is_some_and(ClientConnection::is_active_shell_client) { return false; } @@ -2494,96 +2541,8 @@ impl HeadlessServer { ServerEvent::ClientShellEndpointRequest { client_id, boot_id, - mut request, - } => { - let Some(client) = self.clients.get(&client_id) else { - return false; - }; - if !matches!(client.mode, ClientConnectionMode::ClientShell) { - self.remove_client_and_resize_if_needed(client_id); - return true; - } - let request_id = request.id.clone(); - if !crate::server::client_commands::supports_client_shell_method(&request.method) { - let message = crate::server::client_commands::error_message( - boot_id, - request_id, - "unsupported_endpoint_command", - "this method is not available through the client shell command lane", - ); - self.send_to_client(client_id, message); - return false; - } - if boot_id != self.client_shell_boot_id { - let message = crate::server::client_commands::error_message( - boot_id, - request_id, - "stale_boot", - "endpoint command targeted an earlier server boot", - ); - self.send_to_client(client_id, message); - return false; - } - if client.shell_endpoint_command_in_flight { - self.send_to_client( - client_id, - ServerMessage::ServerShutdown { - reason: Some("client sent concurrent endpoint commands".into()), - }, - ); - self.remove_client_and_resize_if_needed(client_id); - return true; - } - - let api_request_id = format!( - "endpoint:{}:{client_id}:{request_id}", - self.client_shell_boot_id - ); - request.id = api_request_id.clone(); - let (respond_to, response_rx) = std::sync::mpsc::channel(); - if let Err(err) = crate::server::client_commands::spawn_response_waiter( - client_id, - boot_id.clone(), - request_id.clone(), - response_rx, - self.server_event_tx.clone(), - ) { - let message = crate::server::client_commands::error_message( - boot_id, - request_id, - "server_unavailable", - format!("failed to start endpoint response bridge: {err}"), - ); - self.send_to_client(client_id, message); - return false; - } - if let Some(client) = self.clients.get_mut(&client_id) { - client.shell_endpoint_command_in_flight = true; - let deferred_worktree = matches!( - &request.method, - api::schema::Method::WorktreeCreate(_) - | api::schema::Method::WorktreeRemove(_) - ); - let deferred_navigation = matches!( - &request.method, - api::schema::Method::WorktreeCreate(params) if params.focus - ); - client.shell_deferred_navigation_request_id = - deferred_worktree.then(|| api_request_id.clone()); - client.shell_deferred_navigation_response = deferred_navigation.then(Vec::new); - } - let foreground_changed = self.promote_client_to_foreground(client_id); - foreground_changed - | self.handle_client_shell_api_request( - client_id, - api::ApiRequestMessage { - request: *request, - respond_to, - response_write_complete: None, - stream_active: None, - }, - ) - } + request, + } => self.handle_client_shell_endpoint_request(client_id, boot_id, request), ServerEvent::ClientShellEndpointResponseChunkReady { client_id, boot_id, @@ -2591,14 +2550,16 @@ impl HeadlessServer { final_chunk, data, } => { - let valid = self.clients.get(&client_id).is_some_and(|client| { - matches!(client.mode, ClientConnectionMode::ClientShell) + let command_surface_revision = self.clients.get(&client_id).and_then(|client| { + (matches!(client.mode, ClientConnectionMode::ClientShell) && client.shell_endpoint_command_in_flight - && boot_id == self.client_shell_boot_id + && boot_id == self.client_shell_boot_id) + .then_some(client.shell_endpoint_command_surface_revision) + .flatten() }); - if !valid { + let Some(command_surface_revision) = command_surface_revision else { return false; - } + }; let completed_deferred_response = self.clients.get_mut(&client_id).and_then(|client| { let response = client.shell_deferred_navigation_response.as_mut()?; @@ -2610,6 +2571,7 @@ impl HeadlessServer { if final_chunk { if let Some(client) = self.clients.get_mut(&client_id) { client.shell_endpoint_command_in_flight = false; + client.shell_endpoint_command_surface_revision = None; client.shell_deferred_navigation_request_id = None; } } @@ -2618,7 +2580,10 @@ impl HeadlessServer { .and_then(Self::deferred_endpoint_navigation_tab_id); let focus_before = self.shell_focus_target(client_id); let focused_tabs_before = self.focused_shell_tabs(); - let navigation_changed = deferred_tab_id + let navigation_changed = self.clients.get(&client_id).is_some_and(|client| { + client.is_active_shell_client() + && client.shell_projection_revision == command_surface_revision + }) && deferred_tab_id .as_deref() .is_some_and(|tab_id| self.focus_shell_client_on_tab(client_id, tab_id)); let geometry_changed = diff --git a/src/server/headless/client_views.rs b/src/server/headless/client_views.rs index 2d49a50a..2257a05e 100644 --- a/src/server/headless/client_views.rs +++ b/src/server/headless/client_views.rs @@ -388,7 +388,7 @@ impl HeadlessServer { self.clients .iter() .filter(|(_, client)| { - client.is_shell_client() && client.outer_terminal_focus == Some(true) + client.is_active_shell_client() && client.outer_terminal_focus == Some(true) }) .filter_map(|(&client_id, _)| self.shell_tab_id_for_client(client_id)) .collect() @@ -397,7 +397,7 @@ impl HeadlessServer { pub(super) fn shell_focus_targets(&self) -> Vec<(u64, Option)> { self.clients .iter() - .filter(|(_, client)| client.is_shell_client()) + .filter(|(_, client)| client.is_active_shell_client()) .map(|(&client_id, _)| (client_id, self.shell_focus_target(client_id))) .collect() } @@ -608,11 +608,16 @@ impl HeadlessServer { &mut self, start_pending_agent_resumes: bool, ) -> bool { - if self.app_client_count() != 1 { + let active_shell_count = self + .clients + .values() + .filter(|client| client.is_active_shell_client() && client.writer.is_some()) + .count(); + if active_shell_count != 1 { return false; } let Some(client_id) = self.clients.iter().find_map(|(&client_id, client)| { - (client.is_shell_client() && client.writer.is_some()).then_some(client_id) + (client.is_active_shell_client() && client.writer.is_some()).then_some(client_id) }) else { return false; }; @@ -625,7 +630,7 @@ impl HeadlessServer { ) -> bool { let mut viewed_tabs = HashMap::>::new(); for (&client_id, client) in &self.clients { - if !client.is_shell_client() || client.writer.is_none() { + if !client.is_active_shell_client() || client.writer.is_none() { continue; } let Some(tab_id) = self.shell_tab_id_for_client(client_id) else { @@ -684,6 +689,13 @@ impl HeadlessServer { client_id: u64, start_pending_agent_resumes: bool, ) -> bool { + if !self + .clients + .get(&client_id) + .is_some_and(|client| client.shell_surface_active) + { + return false; + } let Some(tab_id) = self.shell_tab_id_for_client(client_id) else { return false; }; @@ -698,6 +710,13 @@ impl HeadlessServer { client_id: u64, start_pending_agent_resumes: bool, ) -> bool { + if !self + .clients + .get(&client_id) + .is_some_and(|client| client.shell_surface_active) + { + return false; + } let Some(tab_id) = self.shell_tab_id_for_client(client_id) else { return false; }; @@ -713,6 +732,13 @@ impl HeadlessServer { client_id: u64, start_pending_agent_resumes: bool, ) -> bool { + if !self + .clients + .get(&client_id) + .is_some_and(|client| client.shell_surface_active) + { + return false; + } let Some(tab_id) = self.shell_tab_id_for_client(client_id) else { return false; }; diff --git a/src/server/headless/endpoint_requests.rs b/src/server/headless/endpoint_requests.rs new file mode 100644 index 00000000..4f02b676 --- /dev/null +++ b/src/server/headless/endpoint_requests.rs @@ -0,0 +1,130 @@ +use super::*; + +impl HeadlessServer { + pub(super) fn handle_client_shell_endpoint_request( + &mut self, + client_id: u64, + boot_id: String, + mut request: Box, + ) -> bool { + let Some(client) = self.clients.get(&client_id) else { + return false; + }; + if !matches!(client.mode, ClientConnectionMode::ClientShell) { + self.remove_client_and_resize_if_needed(client_id); + return true; + } + let request_id = request.id.clone(); + if !crate::server::client_commands::supports_client_shell_method(&request.method) { + let message = crate::server::client_commands::error_message( + boot_id, + request_id, + "unsupported_endpoint_command", + "this method is not available through the client shell command lane", + ); + self.send_to_client(client_id, message); + return false; + } + if boot_id != self.client_shell_boot_id { + let message = crate::server::client_commands::error_message( + boot_id, + request_id, + "stale_boot", + "endpoint command targeted an earlier server boot", + ); + self.send_to_client(client_id, message); + return false; + } + let surface_active = client.shell_surface_active; + if let api::schema::Method::ClientShellSurfaceSet(params) = &request.method { + let Some((changed, projection_revision)) = + self.set_client_shell_surface_active(client_id, params.active) + else { + return false; + }; + self.send_to_client( + client_id, + crate::server::client_commands::success_message_with_result( + boot_id, + request_id, + api::schema::ResponseResult::ClientShellSurfaceSet { + active: params.active, + projection_revision, + }, + ), + ); + return changed; + } + if client.shell_endpoint_command_in_flight { + let message = crate::server::client_commands::error_message( + boot_id, + request_id, + "endpoint_busy", + "this endpoint is still processing another command", + ); + self.send_to_client(client_id, message); + return false; + } + if !surface_active { + let message = crate::server::client_commands::error_message( + boot_id, + request_id, + "surface_inactive", + "this method requires an active client shell surface", + ); + self.send_to_client(client_id, message); + return false; + } + + let api_request_id = format!( + "endpoint:{}:{client_id}:{request_id}", + self.client_shell_boot_id + ); + request.id = api_request_id.clone(); + let (respond_to, response_rx) = std::sync::mpsc::channel(); + if let Err(err) = crate::server::client_commands::spawn_response_waiter( + client_id, + boot_id.clone(), + request_id.clone(), + response_rx, + self.server_event_tx.clone(), + ) { + let message = crate::server::client_commands::error_message( + boot_id, + request_id, + "server_unavailable", + format!("failed to start endpoint response bridge: {err}"), + ); + self.send_to_client(client_id, message); + return false; + } + if let Some(client) = self.clients.get_mut(&client_id) { + client.shell_endpoint_command_in_flight = true; + // A later source restore has a new projection revision. Keep this request's lease + // so a delayed worktree response cannot focus a pane after endpoint switching. + client.shell_endpoint_command_surface_revision = Some(client.shell_projection_revision); + let deferred_worktree = matches!( + &request.method, + api::schema::Method::WorktreeCreate(_) | api::schema::Method::WorktreeRemove(_) + ); + let deferred_navigation = matches!( + &request.method, + api::schema::Method::WorktreeCreate(params) if params.focus + ); + client.shell_deferred_navigation_request_id = + deferred_worktree.then(|| api_request_id.clone()); + client.shell_deferred_navigation_response = deferred_navigation.then(Vec::new); + } + let foreground_changed = self.promote_client_to_foreground(client_id); + foreground_changed + | self.handle_client_shell_api_request( + client_id, + api::ApiRequestMessage { + request: *request, + respond_to, + response_write_complete: None, + stream_active: None, + }, + ) + } +} diff --git a/src/server/headless/render.rs b/src/server/headless/render.rs index 991d1861..5e36be01 100644 --- a/src/server/headless/render.rs +++ b/src/server/headless/render.rs @@ -39,7 +39,10 @@ impl HeadlessServer { .iter() .filter_map(|(&client_id, client)| match &client.mode { ClientConnectionMode::ClientShell => { - let focused = self.shell_focused_runtime(client_id); + let focused = client + .shell_surface_active + .then(|| self.shell_focused_runtime(client_id)) + .flatten(); let child_requests_mouse = focused.is_some_and(|(runtime, _)| runtime.mouse_reporting_enabled()); let sgr_pixels = client.pixel_mouse @@ -49,8 +52,9 @@ impl HeadlessServer { }); Some(( client_id, - client.shell_mouse_capture || child_requests_mouse, - sgr_pixels, + client.shell_surface_active + && (client.shell_mouse_capture || child_requests_mouse), + client.shell_surface_active && sgr_pixels, )) } ClientConnectionMode::TerminalAttach { terminal_id } => { @@ -113,9 +117,10 @@ impl HeadlessServer { .clients .iter() .filter(|(_, client)| client.is_shell_client()) - .map(|(&client_id, _)| { - let report_all = - self.shell_focused_runtime(client_id) + .map(|(&client_id, client)| { + let report_all = client.shell_surface_active + && self + .shell_focused_runtime(client_id) .is_some_and(|(runtime, _)| { let protocol = runtime.keyboard_protocol(); protocol.reports_all_keys() @@ -224,7 +229,7 @@ impl HeadlessServer { let mut pane_ids = HashSet::new(); if has_app_target { for (&client_id, client) in &self.clients { - if !client.is_shell_client() || client.writer.is_none() { + if !client.is_active_shell_client() || client.writer.is_none() { continue; } let Some(target) = self.shell_target_for_client(client_id) else { @@ -279,7 +284,10 @@ impl HeadlessServer { .filter(|client| client.writer.is_some()) { match &client.mode { - ClientConnectionMode::ClientShell => has_app_target = true, + ClientConnectionMode::ClientShell if client.shell_surface_active => { + has_app_target = true; + } + ClientConnectionMode::ClientShell => {} ClientConnectionMode::TerminalAttach { terminal_id } | ClientConnectionMode::TerminalObserve { terminal_id } => { direct_terminal_targets.insert(terminal_id.as_str()); @@ -339,7 +347,7 @@ impl HeadlessServer { fn any_shell_surface_contains_pane(&self, pane_id: crate::layout::PaneId) -> bool { self.clients.iter().any(|(&client_id, client)| { - if !client.is_shell_client() || client.writer.is_none() { + if !client.is_active_shell_client() || client.writer.is_none() { return false; } if self @@ -456,6 +464,10 @@ impl HeadlessServer { client.shell_snapshot = Some(candidate); } shell_projection_revision = client.shell_projection_revision; + if !client.shell_surface_active { + client.clear_deferred_render(); + continue; + } } let shell_graphics_delivery = self .clients diff --git a/src/server/headless/retained_surface.rs b/src/server/headless/retained_surface.rs index 230704a6..a62a43f6 100644 --- a/src/server/headless/retained_surface.rs +++ b/src/server/headless/retained_surface.rs @@ -199,12 +199,14 @@ struct CollectedPanePatch { mouse_reporting: bool, sgr_pixel_mouse: bool, alternate_screen_active: bool, + graphics_may_have_placements: bool, } struct RetainedRecipientUpdate { client_id: u64, surface: protocol::PaneSurfaceFrame, patch: protocol::PaneSurfacePatch, + graphics_delivery: Option, } impl HeadlessServer { @@ -239,11 +241,20 @@ impl HeadlessServer { { fallback!("unsafe_state"); } - let targets = render_targets(&self.clients, self.foreground_client_id); - if targets.is_empty() - || targets - .iter() - .any(|target| !matches!(target.4, ClientConnectionMode::ClientShell)) + let mut targets = render_targets(&self.clients, self.foreground_client_id); + targets.retain(|(client_id, _, _, _, mode)| { + !matches!(mode, ClientConnectionMode::ClientShell) + || self + .clients + .get(client_id) + .is_some_and(|client| client.shell_surface_active) + }); + if targets.is_empty() { + success!("no_active_surface"); + } + if targets + .iter() + .any(|target| !matches!(target.4, ClientConnectionMode::ClientShell)) { fallback!("non_shell_target"); } @@ -266,8 +277,6 @@ impl HeadlessServer { || surface.frame.height != *rows || surface.popup.is_some() || !surface.graphics.assets.is_empty() - || !surface.graphics.placements.is_empty() - || !surface.graphics.retained_assets.is_empty() || !surface.frame.graphics.is_empty() { fallback!("baseline_mismatch"); @@ -325,6 +334,8 @@ impl HeadlessServer { fallback!("terminal_patch"); } }; + let graphics_may_have_placements = + crate::kitty_graphics::is_enabled() && runtime.kitty_graphics_may_have_placements(); let revision = runtime.content_seq(); if revision != revision_before || !revision.is_multiple_of(2) { fallback!("content_changed"); @@ -337,6 +348,7 @@ impl HeadlessServer { mouse_reporting: runtime.mouse_reporting_enabled(), sgr_pixel_mouse: runtime.sgr_pixel_mouse_enabled(), alternate_screen_active: runtime.alternate_screen_active(), + graphics_may_have_placements, }); } @@ -349,6 +361,8 @@ impl HeadlessServer { let mut changed_panes = Vec::with_capacity(collected.len()); let mut patch_rows = Vec::new(); let mut metadata_changed = false; + let mut refresh_graphics = !surface.graphics.placements.is_empty() + || !surface.graphics.retained_assets.is_empty(); for collected_pane in &collected { let Some(pane) = surface .panes @@ -370,6 +384,7 @@ impl HeadlessServer { ) { fallback!("hyperlink"); } + refresh_graphics |= collected_pane.graphics_may_have_placements; let previous_pane = pane.clone(); let Some(rows) = apply_rows(&mut surface.frame, pane.inner_rect, &collected_pane.patch) @@ -405,7 +420,31 @@ impl HeadlessServer { let cursor = retained_cursor(&self.app, &surface.panes); let cursor_changed = cursor != surface.frame.cursor; surface.frame.cursor = cursor.clone(); - if patch_rows.is_empty() && !cursor_changed && !metadata_changed { + let mut graphics_changed = false; + let graphics_delivery = if refresh_graphics { + let Some(target) = self.shell_target_for_client(client_id) else { + fallback!("graphics_target"); + }; + let client = &self.clients[&client_id]; + let Some((graphics, delivery)) = + crate::server::client_shell_graphics::collect_retained( + &self.app, + &surface, + target, + client.cell_size, + &client.shell_graphics_delivery, + client_id, + ) + else { + fallback!("graphics_geometry"); + }; + graphics_changed = graphics != surface.graphics; + surface.graphics = graphics; + Some(delivery) + } else { + None + }; + if patch_rows.is_empty() && !cursor_changed && !metadata_changed && !graphics_changed { continue; } let patch = protocol::PaneSurfacePatch { @@ -421,6 +460,7 @@ impl HeadlessServer { client_id, surface, patch, + graphics_delivery, }); } if updates.is_empty() { @@ -435,6 +475,7 @@ impl HeadlessServer { client_id, surface, patch, + graphics_delivery, } = update; let Some(client) = self.clients.get_mut(&client_id) else { continue; @@ -444,31 +485,53 @@ impl HeadlessServer { deferred += 1; continue; }; - let Some(prepared) = client - .render_state - .prepare_pane_surface_patch(patch, surface) - else { + // The published row patch cannot carry images. Reuse the retained text/layout + // in a graphics-capable surface message rather than invoking the full renderer. + let prepared = if graphics_delivery.is_some() { + client.render_state.prepare_pane_surface(surface) + } else { + client + .render_state + .prepare_pane_surface_patch(patch, surface) + }; + let Some(prepared) = prepared else { client.defer_full_render(); deferred += 1; continue; }; - let serialized = match Self::frame_server_message(prepared.message()) { - Ok(serialized) => serialized, - Err(error) => { - warn!( - client_id, - %error, - "failed to serialize retained pane surface patch" - ); - client.defer_full_render(); - deferred += 1; - continue; - } + let max_frame_size = if graphics_delivery.is_some() { + MAX_GRAPHICS_FRAME_SIZE + } else { + protocol::MAX_FRAME_SIZE }; + let serialized = + match Self::frame_server_message_with_max(prepared.message(), max_frame_size) { + Ok(serialized) => serialized, + Err(error) => { + warn!( + client_id, + %error, + "failed to serialize retained pane surface patch" + ); + client.defer_full_render(); + deferred += 1; + continue; + } + }; crate::render_prof::counter("retained_surface.bytes", serialized.len() as u64); match writer.render.try_send(serialized) { Ok(()) => { - client.clear_deferred_render(); + let graphics_pending = graphics_delivery + .as_ref() + .is_some_and(crate::kitty_graphics::surface::DeliveryCache::has_pending); + if let Some(delivery) = graphics_delivery { + client.shell_graphics_delivery = delivery; + } + if graphics_pending { + client.defer_full_render(); + } else { + client.clear_deferred_render(); + } client.render_state.commit_sent_frame(prepared); sent += 1; } diff --git a/src/server/headless/surface_interest.rs b/src/server/headless/surface_interest.rs new file mode 100644 index 00000000..7270be0d --- /dev/null +++ b/src/server/headless/surface_interest.rs @@ -0,0 +1,86 @@ +use super::*; + +impl HeadlessServer { + /// Apply a client-shell surface lease and return the resulting projection floor. + /// + /// Every `active: true` request advances the floor, even if the server already considered + /// the connection active. That makes a new client activation epoch distinguishable from a + /// delayed same-boot PaneSurface that was prepared for an earlier epoch. + pub(super) fn set_client_shell_surface_active( + &mut self, + client_id: u64, + active: bool, + ) -> Option<(bool, u64)> { + let focus_before = self.shell_focus_targets(); + let focused_tabs_before = self.focused_shell_tabs(); + let (changed, projection_revision) = { + let client = self.clients.get_mut(&client_id)?; + if !client.is_shell_client() { + return None; + } + let changed = client.shell_surface_active != active; + if active { + client.shell_projection_revision = + client.shell_projection_revision.saturating_add(1); + // Force the next control snapshot to carry this new floor instead of reusing a + // same-boot cached snapshot from the prior surface epoch. + client.shell_snapshot = None; + } + if !changed && !active { + return Some((false, client.shell_projection_revision)); + } + client.shell_surface_active = active; + client.request_repaint(); + client.shell_graphics_delivery = Default::default(); + // The client drops target effects while its old source frame is frozen. Reset the + // dedupe state whenever a viewer is (re)activated so the post-commit replay can + // produce mouse/keyboard modes and graphics even when runtime demand is unchanged. + if active { + client.host_mouse_capture_active = None; + client.host_sgr_pixels_active = None; + client.host_keyboard_report_all_active = None; + } + client.clear_deferred_render(); + if !active { + if let Some(writer) = &client.writer { + writer.discard_pending_render(); + } + } + (changed, client.shell_projection_revision) + }; + let held_inputs = (!active && changed).then(|| { + self.clients + .get_mut(&client_id) + .expect("checked client") + .drain_shell_held_inputs() + }); + + if let Some(held_inputs) = held_inputs { + self.release_client_shell_inputs(client_id, held_inputs); + self.retire_direct_graphics_for_client(client_id); + } + if changed { + self.finish_shell_location_reconciliation(focus_before, &focused_tabs_before); + } + + if active { + self.promote_client_to_foreground(client_id); + // Do not emit/cache a title during a frozen target activation. A committed client + // explicitly requests the bounded replay after its coherent frame is visible. + self.sent_window_title = None; + self.resize_shared_runtime_to_effective_size_with_pending_agent_resumes(true); + self.claim_shell_tab_geometry(client_id, true); + } else { + self.tab_geometry_controllers + .retain(|_, controller_id| *controller_id != client_id); + if self.foreground_client_id == Some(client_id) { + self.promote_latest_remaining_client(); + self.resize_shared_runtime_to_effective_size_with_pending_agent_resumes(true); + } else { + self.sync_foreground_client_state(); + } + self.reapply_controlled_shell_tab_geometry(true); + } + Some((changed || active, projection_revision)) + } +} diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs index e7e73e8b..5e0e5d01 100644 --- a/src/server/headless/tests/mod.rs +++ b/src/server/headless/tests/mod.rs @@ -2,6 +2,8 @@ use super::*; #[path = "pane_graphics.rs"] mod pane_graphics_tests; +#[path = "surface_interest.rs"] +mod surface_interest_tests; fn client_shell_snapshot(message: ServerMessage) -> Box { let ServerMessage::EndpointControl { kind, data } = message else { @@ -607,6 +609,7 @@ async fn client_shell_attach_seeds_workspace() { direct_graphics: false, endpoint_keybindings: false, mouse_capture: false, + surface_active: true, writer, }) ); @@ -636,6 +639,7 @@ async fn client_shell_endpoint_request_uses_the_selected_connection() { direct_graphics: false, endpoint_keybindings: false, mouse_capture: false, + surface_active: true, writer, }) ); @@ -654,6 +658,26 @@ async fn client_shell_endpoint_request_uses_the_selected_connection() { ); assert!(server.clients[&client_id].shell_endpoint_command_in_flight); + assert!( + !server.handle_server_event(ServerEvent::ClientShellEndpointRequest { + client_id, + boot_id: boot_id.clone(), + request: Box::new(api::schema::Request { + id: "client-shell:busy".into(), + method: api::schema::Method::IntegrationList(api::schema::EmptyParams::default()), + }), + }) + ); + assert!(server.clients.contains_key(&client_id)); + let ServerMessage::ClientShellEndpointResponseChunk { data, .. } = + read_server_message(control_rx.recv().expect("busy endpoint response")) + else { + panic!("expected busy endpoint response"); + }; + let response = + serde_json::from_slice::(&data).expect("typed busy response"); + assert_eq!(response.error.code, "endpoint_busy"); + let response_ready = server .server_event_rx .recv() @@ -751,6 +775,7 @@ async fn client_shell_receives_metadata_then_shell_free_pane_surface() { direct_graphics: false, endpoint_keybindings: false, mouse_capture: false, + surface_active: true, writer, }) ); @@ -894,6 +919,7 @@ fn connect_test_shell( direct_graphics: false, endpoint_keybindings: false, mouse_capture: false, + surface_active: true, writer, }) ); @@ -1219,6 +1245,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() { direct_graphics: false, endpoint_keybindings: false, mouse_capture: false, + surface_active: true, writer: local_writer, }) ); @@ -1242,6 +1269,7 @@ async fn client_shell_config_diagnostics_follow_keybinding_ownership() { direct_graphics: false, endpoint_keybindings: true, mouse_capture: false, + surface_active: true, writer: endpoint_writer, }) ); @@ -1351,6 +1379,12 @@ async fn deferred_worktree_response_moves_only_its_source_client() { .get_mut(&51) .unwrap() .shell_endpoint_command_in_flight = true; + let source_surface_revision = server.clients[&51].shell_projection_revision; + server + .clients + .get_mut(&51) + .unwrap() + .shell_endpoint_command_surface_revision = Some(source_surface_revision); server .clients .get_mut(&51) @@ -1385,16 +1419,31 @@ async fn deferred_worktree_response_moves_only_its_source_client() { ); assert!(server.focus_shell_client_on_tab(51, &original_tab_id)); + // A source command begun in the old presentation epoch may finish after source-off and + // source-on rollback. Its response remains endpoint-local, but it must not apply deferred + // client navigation to the restored source. + assert!(server.set_client_shell_surface_active(51, false).is_some()); + assert!(server.set_client_shell_surface_active(51, true).is_some()); server .clients .get_mut(&51) .unwrap() .shell_endpoint_command_in_flight = true; + server + .clients + .get_mut(&51) + .unwrap() + .shell_endpoint_command_surface_revision = Some(source_surface_revision); server .clients .get_mut(&51) .unwrap() .shell_deferred_navigation_request_id = Some("background-worktree".into()); + server + .clients + .get_mut(&51) + .unwrap() + .shell_deferred_navigation_response = Some(Vec::new()); assert!( !server.handle_server_event(ServerEvent::ClientShellEndpointResponseChunkReady { client_id: 51, @@ -1919,6 +1968,7 @@ async fn public_api_focus_replaces_every_client_shell_projection() { direct_graphics: false, endpoint_keybindings: false, mouse_capture: false, + surface_active: true, writer, }) ); @@ -2168,6 +2218,7 @@ async fn client_shell_streams_and_targets_popup_terminal_content() { direct_graphics: false, endpoint_keybindings: false, mouse_capture: false, + surface_active: true, writer, }) ); @@ -2663,6 +2714,19 @@ fn retained_test_server( HeadlessServer, std::sync::mpsc::Receiver>, crate::layout::PaneId, +) { + let (server, _control_rx, render_rx, pane_id) = + retained_test_server_with_control(initial_screen); + (server, render_rx, pane_id) +} + +fn retained_test_server_with_control( + initial_screen: &[u8], +) -> ( + HeadlessServer, + std::sync::mpsc::Receiver>, + std::sync::mpsc::Receiver>, + crate::layout::PaneId, ) { let mut server = test_headless_server(); let mut workspace = crate::workspace::Workspace::test_new("test"); @@ -2676,7 +2740,7 @@ fn retained_test_server( server.app.state.selected = 0; server.app.state.mode = crate::app::Mode::Terminal; - let (client_tx, _client_control_rx, client_rx) = test_client_writer(); + let (client_tx, client_control_rx, client_rx) = test_client_writer(); server.clients.insert( 1, ClientConnection::new( @@ -2691,7 +2755,7 @@ fn retained_test_server( server.sync_foreground_client_state(); assert!(server.claim_unowned_shell_tab_geometry(1, true)); - (server, client_rx, pane_id) + (server, client_control_rx, client_rx, pane_id) } #[test] diff --git a/src/server/headless/tests/pane_graphics.rs b/src/server/headless/tests/pane_graphics.rs index d310003c..f8637e8f 100644 --- a/src/server/headless/tests/pane_graphics.rs +++ b/src/server/headless/tests/pane_graphics.rs @@ -6,7 +6,8 @@ fn receive_render(receiver: &std::sync::mpsc::Receiver>, timeout: Durati #[tokio::test] async fn client_shell_surface_sends_complete_placements_and_each_live_asset_once() { - 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 = @@ -38,6 +39,19 @@ async fn client_shell_surface_sends_complete_placements_and_each_live_asset_once }; assert_eq!(second.graphics.placements, first.graphics.placements); assert!(second.graphics.assets.is_empty()); + + // A post-commit typed surface.set(true) resets only this viewer's delivery cache, so the + // already selected target receives the asset again without relying on graphics that may have + // arrived during the frozen source frame. + assert!(server.set_client_shell_surface_active(1, true).is_some()); + server.render_and_stream(); + let replay = read_server_message(receive_render(&client_rx, Duration::from_millis(100))); + let ServerMessage::PaneSurface(replay) = replay else { + panic!("expected post-commit graphics replay"); + }; + assert_eq!(replay.graphics.placements, first.graphics.placements); + assert_eq!(replay.graphics.assets.len(), 1); + assert_eq!(replay.graphics.assets[0].data, vec![1, 2, 3, 4]); } #[tokio::test] @@ -76,6 +90,237 @@ async fn client_shell_asset_delivery_is_bounded_to_the_current_live_scene() { assert_eq!(restored.graphics.assets[0].data, vec![1, 2, 3, 4]); } +#[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 client = server.clients.get_mut(&1).unwrap(); + client.mode = ClientConnectionMode::ClientShell; + client.render_state = + crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame); + client.cell_size = crate::kitty_graphics::HostCellSize { + width_px: 10, + height_px: 20, + }; + server.render_and_stream(); + let ServerMessage::PaneSurface(initial) = + read_server_message(receive_render(&client_rx, Duration::from_millis(100))) + else { + panic!("expected text-only baseline"); + }; + assert!(initial.graphics.placements.is_empty()); + + write_shared_test_pane( + &mut server, + pane_id, + 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\\", + ); + assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id]))); + let ServerMessage::PaneSurface(repainted) = + read_server_message(receive_render(&client_rx, Duration::from_millis(100))) + else { + panic!("expected image update without a full redraw"); + }; + assert_eq!(repainted.graphics.placements.len(), 1); + assert_eq!(repainted.graphics.assets[0].data, [255, 0, 0, 255]); + assert_eq!(repainted.panes[0].inner_rect, initial.panes[0].inner_rect); + + // Text changes while an image is visible must reuse its uploaded pixels. + write_shared_test_pane(&mut server, pane_id, b"\rupdated text"); + assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id]))); + let ServerMessage::PaneSurface(text_update) = + read_server_message(receive_render(&client_rx, Duration::from_millis(100))) + else { + panic!("expected retained text and image scene"); + }; + assert_eq!( + text_update.graphics.placements, + repainted.graphics.placements + ); + assert!(text_update.graphics.assets.is_empty()); + assert!(frame_text(&text_update.frame).contains("updated text")); + + write_shared_test_pane(&mut server, pane_id, b"\x1b_Ga=d,d=A\x1b\\"); + assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id]))); + let ServerMessage::PaneSurface(deleted) = + read_server_message(receive_render(&client_rx, Duration::from_millis(100))) + else { + panic!("expected image removal"); + }; + assert!(deleted.graphics.placements.is_empty()); + assert!(deleted.graphics.assets.is_empty()); + + write_shared_test_pane(&mut server, pane_id, b"\rtext only again"); + assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id]))); + assert!(matches!( + read_server_message(receive_render(&client_rx, Duration::from_millis(100))), + ServerMessage::PaneSurfacePatch(_) + )); + + // A full output queue must not mark unsent pixels as delivered. + fill_render_lane(&server); + write_shared_test_pane( + &mut server, + pane_id, + 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\\", + ); + assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id]))); + assert!(server.clients[&1] + .render_state + .last_pane_surface() + .unwrap() + .graphics + .placements + .is_empty()); + let _ = receive_render(&client_rx, Duration::from_millis(100)); + server.render_and_stream(); + let ServerMessage::PaneSurface(recovered) = + read_server_message(receive_render(&client_rx, Duration::from_millis(100))) + else { + panic!("expected deferred graphics recovery"); + }; + assert_eq!(recovered.graphics.assets[0].data, [255, 0, 0, 255]); +} + +#[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 client = server.clients.get_mut(&1).unwrap(); + client.mode = ClientConnectionMode::ClientShell; + client.cell_size = crate::kitty_graphics::HostCellSize { + width_px: 10, + height_px: 20, + }; + server.render_and_stream(); + let _ = receive_render(&client_rx, Duration::from_millis(100)); + let sources = HashSet::from([pane_id]); + // Yazi-style virtual placement: uploading the image and drawing its Unicode cell + // can happen in separate PTY reads, with no text dirty rows when upload completes. + for bytes in [ + b"\x1b_Ga=t,f=32,t=d,i=1193046,s=1,v=1,q=2;/wAA/w".as_slice(), + b"==\x1b\\", + b"\x1b_Ga=p,U=1,i=1193046,c=1,r=1,q=2\x1b\\", + ] { + write_shared_test_pane(&mut server, pane_id, bytes); + assert!(server.render_retained_pane_surface_and_stream(&sources)); + for frame in client_rx.try_iter() { + if let ServerMessage::PaneSurface(surface) = read_server_message(frame) { + assert!(surface.graphics.placements.is_empty()); + } + } + } + write_shared_test_pane( + &mut server, + pane_id, + "\x1b[2;3H\x1b[38;2;18;52;86m\u{10eeee}\u{0305}\u{0305}\x1b[0m".as_bytes(), + ); + assert!(server.render_retained_pane_surface_and_stream(&sources)); + let ServerMessage::PaneSurface(surface) = + read_server_message(receive_render(&client_rx, Duration::from_millis(100))) + else { + panic!("virtual image must arrive without a tab switch"); + }; + assert_eq!(surface.graphics.placements.len(), 1); + assert_eq!(surface.graphics.assets[0].data, [255, 0, 0, 255]); + // Replacing pixels under the same image ID must invalidate the delivered asset. + write_shared_test_pane( + &mut server, + pane_id, + b"\x1b_Ga=t,f=32,t=d,i=1193046,s=1,v=1,q=2;AP8A/w==\x1b\\", + ); + assert!(server.render_retained_pane_surface_and_stream(&sources)); + let ServerMessage::PaneSurface(replaced) = + read_server_message(receive_render(&client_rx, Duration::from_millis(100))) + else { + panic!("updated image pixels must arrive"); + }; + assert_eq!(replaced.graphics.assets[0].data, [0, 255, 0, 255]); + assert_ne!( + surface.graphics.assets[0].key, + replaced.graphics.assets[0].key + ); +} + +#[tokio::test] +#[ignore = "manual retained text/image scaling profile"] +async fn render_scale_profile_retained_graphics() { + use ratatui::layout::Direction; + 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 pane_ids = vec![root]; + for index in 1..count { + let workspace = &mut server.app.state.workspaces[0]; + workspace.tabs[0] + .layout + .focus_pane(pane_ids[(index - 1) / 2]); + let id = workspace.test_split(if index % 2 == 0 { + Direction::Vertical + } else { + Direction::Horizontal + }); + workspace.insert_test_runtime( + id, + crate::terminal::TerminalRuntime::test_with_screen_bytes( + 80, + 24, + b"populated terminal\r\n", + ), + ); + pane_ids.push(id); + } + let client = server.clients.get_mut(&1).unwrap(); + client.mode = ClientConnectionMode::ClientShell; + client.cell_size = crate::kitty_graphics::HostCellSize { + width_px: 10, + height_px: 20, + }; + if with_image { + write_shared_test_pane( + &mut server, + root, + 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\\", + ); + } + server.render_and_stream(); + let _ = receive_render(&client_rx, Duration::from_millis(100)); + let sources = pane_ids.iter().copied().collect(); + let mut samples = Vec::new(); + for sample in 0..110 { + for id in &pane_ids { + write_shared_test_pane( + &mut server, + *id, + format!("\x1b[H{sample:03}").as_bytes(), + ); + } + let started = Instant::now(); + if retained { + assert!(server.render_retained_pane_surface_and_stream(&sources)); + } else { + server.render_and_stream(); + } + let elapsed = started.elapsed(); + for frame in client_rx.try_iter() { + if let ServerMessage::PaneSurface(surface) = read_server_message(frame) { + assert!(surface.graphics.assets.is_empty()); + } + } + if sample >= 10 { + samples.push(elapsed); + } + } + samples.sort_unstable(); + println!( + "retained={retained} 80x24 panes={count} image={with_image} median_us={} p95_us={}", + samples[50].as_micros(), + samples[94].as_micros() + ); + } + } + } +} + #[tokio::test] async fn client_shell_surface_projects_terminal_kitty_images_from_authoritative_runtime() { let (mut server, client_rx, _pane_id) = diff --git a/src/server/headless/tests/surface_interest.rs b/src/server/headless/tests/surface_interest.rs new file mode 100644 index 00000000..591f6a8f --- /dev/null +++ b/src/server/headless/tests/surface_interest.rs @@ -0,0 +1,725 @@ +use super::*; + +use std::sync::{Arc, Mutex}; + +use crate::client::endpoint::{ + ClientEndpointId, ClientEndpointStatus, EndpointNegotiation, EndpointRegistry, + EndpointTransport, ProfileId, SavedSshEndpoint, +}; + +#[derive(Clone)] +struct CapturingEndpointTransport(Arc>>); + +impl EndpointTransport for CapturingEndpointTransport { + fn send(&mut self, message: &crate::protocol::ClientMessage) -> std::io::Result<()> { + self.0.lock().unwrap().push(message.clone()); + Ok(()) + } +} + +fn lifecycle_negotiation() -> EndpointNegotiation { + EndpointNegotiation::new( + vec!["client_shell.surface.set".into()], + vec![ + crate::protocol::endpoint::SURFACE_INTEREST_CAPABILITY.into(), + crate::protocol::endpoint::PRESENTATION_EFFECTS_FENCE_CAPABILITY.into(), + ], + ) +} + +fn lifecycle_resize() -> crate::protocol::ClientMessage { + crate::protocol::ClientMessage::ClientShellResize { + cell_width_px: 8, + cell_height_px: 16, + surface_size: crate::protocol::ClientSurfaceSize { cols: 80, rows: 24 }, + pixel_mouse: false, + } +} + +#[tokio::test] +async fn metadata_only_shell_is_isolated_until_surface_activation() { + let mut server = test_headless_server(); + let mut input_rx = install_focused_test_runtime(&mut server, b""); + let pane_id = server.app.session_snapshot().focused_pane_id.unwrap(); + let workspace_id = server.app.session_snapshot().focused_workspace_id.unwrap(); + let original_size = server.effective_size; + let (writer, control_rx, render_rx) = test_client_writer(); + let client_id = 52; + + assert!( + server.handle_server_event(ServerEvent::ClientShellConnected { + client_id, + surface_cols: 101, + surface_rows: 37, + cell_width_px: 9, + cell_height_px: 18, + pixel_mouse: true, + direct_graphics: false, + endpoint_keybindings: true, + mouse_capture: true, + surface_active: false, + writer, + }) + ); + assert!(matches!( + read_server_message(control_rx.recv().expect("metadata snapshot")), + ServerMessage::EndpointControl { .. } + )); + assert_eq!(server.foreground_client_id, None); + assert_eq!(server.effective_size, original_size); + + server.render_and_stream(); + assert!(render_rx.try_recv().is_err()); + assert!(server.clients[&client_id] + .render_state + .last_pane_surface() + .is_none()); + + assert!( + !server.handle_server_event(ServerEvent::ClientShellPaneInput { + client_id, + pane_id, + events: vec![crate::protocol::ClientPaneInputEvent::Paste( + "blocked".into() + )], + }) + ); + assert!(input_rx.try_recv().is_err()); + + let boot_id = server.client_shell_boot_id.clone(); + assert!( + !server.handle_server_event(ServerEvent::ClientShellEndpointRequest { + client_id, + boot_id: boot_id.clone(), + request: Box::new(api::schema::Request { + id: "inactive-mutation".into(), + method: api::schema::Method::WorkspaceFocus(api::schema::WorkspaceTarget { + workspace_id, + }), + }), + }) + ); + let ServerMessage::ClientShellEndpointResponseChunk { data, .. } = + read_server_message(control_rx.recv().expect("inactive mutation response")) + else { + panic!("expected endpoint response"); + }; + let error = serde_json::from_slice::(&data).unwrap(); + assert_eq!(error.error.code, "surface_inactive"); + + assert!( + server.send_to_client_shells(ServerMessage::SemanticNotification( + crate::protocol::SemanticNotification { + kind: crate::protocol::SemanticNotificationKind::Custom, + title: "metadata event".into(), + body: None, + sound: None, + agent: None, + workspace_id: None, + tab_id: None, + pane_id: None, + position: None, + }, + )) + ); + assert!(matches!( + read_server_message(control_rx.recv().expect("metadata notification")), + ServerMessage::SemanticNotification(_) + )); + + assert!( + server.handle_server_event(ServerEvent::ClientShellEndpointRequest { + client_id, + boot_id: boot_id.clone(), + request: Box::new(api::schema::Request { + id: "activate-surface".into(), + method: api::schema::Method::ClientShellSurfaceSet( + api::schema::ClientShellSurfaceSetParams { active: true }, + ), + }), + }) + ); + let ServerMessage::ClientShellEndpointResponseChunk { data, .. } = + read_server_message(control_rx.recv().expect("surface activation response")) + else { + panic!("expected typed surface activation response"); + }; + let activation_ack = serde_json::from_slice::(&data).unwrap(); + let api::schema::ResponseResult::ClientShellSurfaceSet { + active: true, + projection_revision: activation_floor, + } = activation_ack.result + else { + panic!("expected typed surface activation result"); + }; + assert_eq!(server.foreground_client_id, Some(client_id)); + assert_eq!(server.effective_size, (101, 37)); + + server.render_and_stream(); + let ServerMessage::PaneSurface(surface) = + read_server_message(render_rx.recv().expect("activated surface")) + else { + panic!("expected pane surface"); + }; + assert_eq!((surface.frame.width, surface.frame.height), (101, 37)); + assert!(surface.projection_revision >= activation_floor); + assert_eq!(surface.surface_revision, 1); + server + .clients + .get_mut(&client_id) + .expect("surface client") + .shell_endpoint_command_in_flight = true; + + assert!( + server.handle_server_event(ServerEvent::ClientShellEndpointRequest { + client_id, + boot_id: boot_id.clone(), + request: Box::new(api::schema::Request { + id: "deactivate-surface".into(), + method: api::schema::Method::ClientShellSurfaceSet( + api::schema::ClientShellSurfaceSetParams { active: false }, + ), + }), + }) + ); + let _ = control_rx.recv().expect("surface deactivation response"); + assert!(server.clients.contains_key(&client_id)); + let (_, runtime_pane_id) = server.app.parse_pane_id(&surface.panes[0].pane_id).unwrap(); + server + .app + .state + .runtime_for_pane_in_workspace(&server.app.terminal_runtimes, 0, runtime_pane_id) + .unwrap() + .test_process_pty_bytes(b"REACTIVATED"); + assert!( + server.render_retained_pane_surface_and_stream(&std::collections::HashSet::from([ + runtime_pane_id + ])) + ); + assert!(render_rx.try_recv().is_err()); + + assert!( + server.handle_server_event(ServerEvent::ClientShellEndpointRequest { + client_id, + boot_id, + request: Box::new(api::schema::Request { + id: "reactivate-surface".into(), + method: api::schema::Method::ClientShellSurfaceSet( + api::schema::ClientShellSurfaceSetParams { active: true }, + ), + }), + }) + ); + let data = loop { + let message = + read_server_message(control_rx.recv().expect("surface reactivation response")); + match message { + ServerMessage::ClientShellEndpointResponseChunk { + request_id, data, .. + } if request_id == "reactivate-surface" => break data, + ServerMessage::EndpointControl { .. } + | ServerMessage::ClientShellEndpointResponseChunk { .. } => continue, + other => panic!("unexpected surface reactivation message: {other:?}"), + } + }; + let reactivation_ack = serde_json::from_slice::(&data).unwrap(); + let api::schema::ResponseResult::ClientShellSurfaceSet { + active: true, + projection_revision: reactivation_floor, + } = reactivation_ack.result + else { + panic!("expected typed surface reactivation result"); + }; + assert!(reactivation_floor > activation_floor); + server.render_and_stream(); + let ServerMessage::PaneSurface(surface) = + read_server_message(render_rx.recv().expect("reactivated surface")) + else { + panic!("expected replacement pane surface"); + }; + assert!(frame_text(&surface.frame).contains("REACTIVATED")); + assert!(surface.projection_revision >= reactivation_floor); + assert_eq!(surface.surface_revision, 2); + shutdown_test_runtimes(&mut server); +} + +#[tokio::test] +async fn presentation_sync_epoch_replays_modes_and_title() { + let mut server = test_headless_server(); + let (writer, control_rx, _render_rx) = test_client_writer(); + let client_id = 63; + assert!( + server.handle_server_event(ServerEvent::ClientShellConnected { + client_id, + surface_cols: 80, + surface_rows: 24, + cell_width_px: 8, + cell_height_px: 16, + pixel_mouse: false, + direct_graphics: false, + endpoint_keybindings: true, + mouse_capture: true, + surface_active: true, + writer, + }) + ); + let _ = control_rx.recv().expect("initial snapshot"); + server.api_window_title = Some("target title".into()); + { + let client = server.clients.get_mut(&client_id).unwrap(); + client.host_mouse_capture_active = Some(false); + client.host_sgr_pixels_active = Some(false); + client.host_keyboard_report_all_active = Some(false); + } + + let boot_id = server.client_shell_boot_id.clone(); + assert!( + server.handle_server_event(ServerEvent::ClientShellEndpointRequest { + client_id, + boot_id, + request: Box::new(api::schema::Request { + id: "post-commit-reassert".into(), + method: api::schema::Method::ClientShellSurfaceSet( + api::schema::ClientShellSurfaceSetParams { active: true }, + ), + }), + }) + ); + let _ = control_rx + .recv() + .expect("typed surface reassertion acknowledgement"); + server.stream_host_mouse_capture_mode(); + server.stream_direct_terminal_keyboard_mode(); + server.sync_window_title(); + assert_eq!( + server.clients[&client_id].host_mouse_capture_active, + Some(true), + "the target mode is sent after, not during, the frozen handoff" + ); + assert_eq!( + server.clients[&client_id].host_keyboard_report_all_active, + Some(false) + ); + let messages = (0..3) + .map(|_| read_server_message(control_rx.recv().expect("reassertion effect"))) + .collect::>(); + assert!(messages + .iter() + .any(|message| matches!(message, ServerMessage::MouseCapture { enabled: true, .. }))); + assert!(messages.iter().any(|message| matches!( + message, + ServerMessage::ClientShellKeyboardReportAll { enabled: false } + ))); + assert!(messages.iter().any(|message| matches!( + message, + ServerMessage::WindowTitle { title: Some(title) } if title == "target title" + ))); + shutdown_test_runtimes(&mut server); +} + +fn dispatch_lifecycle_messages( + server: &mut HeadlessServer, + client_id: u64, + messages: Vec, +) { + for message in messages { + let event = match message { + crate::protocol::ClientMessage::ClientShellResize { + cell_width_px, + cell_height_px, + surface_size, + pixel_mouse, + } => ServerEvent::ClientShellResize { + client_id, + cell_width_px, + cell_height_px, + surface_cols: surface_size.cols, + surface_rows: surface_size.rows, + pixel_mouse, + }, + crate::protocol::ClientMessage::ClientShellFocus { focused } => { + ServerEvent::ClientShellFocus { client_id, focused } + } + crate::protocol::ClientMessage::ClientShellEndpointRequest { boot_id, request } => { + ServerEvent::ClientShellEndpointRequest { + client_id, + boot_id, + request: Box::new(serde_json::from_str(&request).unwrap()), + } + } + other => panic!("unhandled lifecycle message: {other:?}"), + }; + server.handle_server_event(event); + } +} + +/// A real two-server/client lifecycle harness. Both source-off and target-on traverse the +/// production HeadlessServer endpoint request path; the client test only routes its emitted wire +/// messages and never authors an acknowledgement, snapshot, or surface response. +#[tokio::test] +async fn two_headless_servers_drive_atomic_endpoint_handoff() { + let mut source_server = test_headless_server(); + let _source_input = install_focused_test_runtime(&mut source_server, b"local source"); + let (source_writer, source_control, _source_render) = test_client_writer(); + let source_client_id = 78; + assert!( + source_server.handle_server_event(ServerEvent::ClientShellConnected { + client_id: source_client_id, + surface_cols: 80, + surface_rows: 24, + cell_width_px: 8, + cell_height_px: 16, + pixel_mouse: false, + direct_graphics: false, + endpoint_keybindings: true, + mouse_capture: true, + surface_active: true, + writer: source_writer, + }) + ); + let source_snapshot = client_shell_snapshot(read_server_message( + source_control + .recv() + .expect("real source metadata snapshot"), + )); + + let mut target_server = test_headless_server(); + let _target_input = install_focused_test_runtime(&mut target_server, b"remote target"); + let (target_writer, target_control, target_render) = test_client_writer(); + let target_client_id = 79; + assert!( + target_server.handle_server_event(ServerEvent::ClientShellConnected { + client_id: target_client_id, + surface_cols: 80, + surface_rows: 24, + cell_width_px: 8, + cell_height_px: 16, + pixel_mouse: false, + direct_graphics: false, + endpoint_keybindings: true, + mouse_capture: true, + surface_active: false, + writer: target_writer, + }) + ); + let remote_snapshot = client_shell_snapshot(read_server_message( + target_control + .recv() + .expect("real target metadata snapshot"), + )); + + let profile = SavedSshEndpoint { + id: ProfileId::parse("0123456789abcdef0123456789abcdef").unwrap(), + label: "Remote".into(), + target: "dev@example.com".into(), + session: "main".into(), + enabled: true, + }; + let target_id = ClientEndpointId::Ssh(profile.id.clone()); + let mut shell = crate::client::ClientShellState::new( + crate::client::ClientShellConfig::from_config(&crate::config::Config::default()), + ); + shell.set_endpoint_catalog(&[profile]); + shell.set_snapshot(source_snapshot); + shell.set_endpoint_status(&target_id, ClientEndpointStatus::Online); + shell.set_endpoint_snapshot(&target_id, remote_snapshot); + + let source_sent = Arc::new(Mutex::new(Vec::new())); + let target_sent = Arc::new(Mutex::new(Vec::new())); + let mut endpoints = EndpointRegistry::new( + CapturingEndpointTransport(source_sent.clone()), + 1, + lifecycle_negotiation(), + ); + endpoints.insert( + target_id.clone(), + CapturingEndpointTransport(target_sent.clone()), + 7, + lifecycle_negotiation(), + false, + ); + let mut activation = crate::client::endpoint::PendingEndpointActivation::begin( + &shell, + &mut endpoints, + target_id.clone(), + None, + lifecycle_resize(), + 41, + std::time::Instant::now(), + ) + .unwrap(); + + // Route the source-off-first client messages through a second real HeadlessServer. Its + // typed response is the only source acknowledgement supplied to the activation state. + let mut source_release_request_id = None; + for message in std::mem::take(&mut *source_sent.lock().unwrap()) { + match message { + crate::protocol::ClientMessage::ClientShellFocus { focused } => { + assert!( + source_server.handle_server_event(ServerEvent::ClientShellFocus { + client_id: source_client_id, + focused, + }) + ); + } + crate::protocol::ClientMessage::ClientShellEndpointRequest { boot_id, request } => { + let request = serde_json::from_str::(&request).unwrap(); + if matches!( + request.method, + api::schema::Method::ClientShellSurfaceSet( + api::schema::ClientShellSurfaceSetParams { active: false } + ) + ) { + source_release_request_id = Some(request.id.clone()); + } + assert!(source_server.handle_server_event( + ServerEvent::ClientShellEndpointRequest { + client_id: source_client_id, + boot_id, + request: Box::new(request), + } + )); + } + other => panic!("unexpected source lifecycle message: {other:?}"), + } + } + let source_release_request_id = source_release_request_id.expect("client source-off request"); + let source_release_data = loop { + let message = read_server_message(source_control.recv().expect("source typed release ack")); + match message { + ServerMessage::ClientShellEndpointResponseChunk { + request_id, data, .. + } if request_id == source_release_request_id => break data, + ServerMessage::EndpointControl { .. } + | ServerMessage::MouseCapture { .. } + | ServerMessage::ClientShellKeyboardReportAll { .. } + | ServerMessage::WindowTitle { .. } + | ServerMessage::ClientShellEndpointResponseChunk { .. } => continue, + other => panic!("unexpected source release message: {other:?}"), + } + }; + assert_eq!( + activation.receive_response( + &ClientEndpointId::Local, + 1, + &source_release_request_id, + &source_release_data, + &mut endpoints, + ), + crate::client::endpoint::SurfaceActivationProgress::Pending + ); + + dispatch_lifecycle_messages( + &mut target_server, + target_client_id, + std::mem::take(&mut *target_sent.lock().unwrap()), + ); + assert_eq!( + target_server.clients[&target_client_id].outer_terminal_focus, + Some(true) + ); + assert_eq!( + source_server.clients[&source_client_id].outer_terminal_focus, + Some(false) + ); + let ServerMessage::ClientShellEndpointResponseChunk { + request_id, data, .. + } = read_server_message(target_control.recv().expect("target typed activation ack")) + else { + panic!("expected target activation acknowledgement"); + }; + assert_eq!( + activation.receive_response(&target_id, 7, &request_id, &data, &mut endpoints), + crate::client::endpoint::SurfaceActivationProgress::Pending + ); + + target_server.render_and_stream(); + let coherent_snapshot = client_shell_snapshot(read_server_message( + target_control.recv().expect("target replacement snapshot"), + )); + let snapshot_progress = activation.receive_snapshot(&target_id, 7, &coherent_snapshot); + shell.set_endpoint_snapshot(&target_id, coherent_snapshot); + assert_eq!( + snapshot_progress, + crate::client::endpoint::SurfaceActivationProgress::Pending + ); + let ServerMessage::PaneSurface(coherent_surface) = + read_server_message(target_render.recv().expect("target replacement surface")) + else { + panic!("expected target pane surface"); + }; + assert_eq!( + activation.receive_surface(&target_id, 7, coherent_surface), + crate::client::endpoint::SurfaceActivationProgress::Ready + ); + + assert!(matches!( + activation.complete(&mut shell, &mut endpoints), + Ok(crate::client::endpoint::ActivationCompletion::AwaitingPresentationSync { + endpoint, + .. + }) if endpoint == target_id + )); + assert!( + !endpoints.active_surface_available(), + "target input remains fenced while presentation effects resynchronize" + ); + + let sync_request = target_sent + .lock() + .unwrap() + .iter() + .find_map(|message| match message { + crate::protocol::ClientMessage::ClientShellEndpointRequest { boot_id, request } => { + let request = serde_json::from_str::(request).ok()?; + request + .id + .ends_with(":presentation-sync") + .then(|| (boot_id.clone(), Box::new(request))) + } + _ => None, + }) + .expect("client presentation synchronization request"); + assert!( + target_server.handle_server_event(ServerEvent::ClientShellEndpointRequest { + client_id: target_client_id, + boot_id: sync_request.0, + request: sync_request.1, + }) + ); + let (sync_request_id, sync_data) = loop { + let message = read_server_message(target_control.recv().expect("presentation sync ack")); + if let ServerMessage::ClientShellEndpointResponseChunk { + request_id, data, .. + } = message + { + if request_id.ends_with(":presentation-sync") { + break (request_id, data); + } + } + }; + assert_eq!( + activation.receive_response(&target_id, 7, &sync_request_id, &sync_data, &mut endpoints,), + crate::client::endpoint::SurfaceActivationProgress::Pending + ); + target_server.render_and_stream(); + let sync_snapshot = loop { + let message = + read_server_message(target_control.recv().expect("presentation sync snapshot")); + if let ServerMessage::EndpointControl { kind, data } = message { + if kind == crate::protocol::endpoint::ENDPOINT_SNAPSHOT_KIND { + break serde_json::from_str::(&data).unwrap(); + } + } + }; + let sync_progress = activation.receive_snapshot(&target_id, 7, &sync_snapshot); + shell.set_endpoint_snapshot_for_generation(&target_id, 7, Box::new(sync_snapshot)); + assert_eq!( + sync_progress, + crate::client::endpoint::SurfaceActivationProgress::Pending + ); + let ServerMessage::PaneSurface(sync_surface) = + read_server_message(target_render.recv().expect("presentation sync surface")) + else { + panic!("expected synchronized target surface"); + }; + assert_eq!( + activation.receive_surface(&target_id, 7, sync_surface), + crate::client::endpoint::SurfaceActivationProgress::Ready + ); + assert_eq!( + activation.complete(&mut shell, &mut endpoints), + Ok(crate::client::endpoint::ActivationCompletion::AwaitingPresentationEffects) + ); + let effects_token = target_sent + .lock() + .unwrap() + .iter() + .find_map(|message| match message { + crate::protocol::ClientMessage::EndpointControl { kind, data } + if kind == crate::protocol::endpoint::PRESENTATION_EFFECTS_SYNC_KIND => + { + Some(data.clone()) + } + _ => None, + }) + .expect("client presentation effects fence"); + assert!( + target_server.handle_server_event(ServerEvent::ClientShellPresentationSync { + client_id: target_client_id, + token: effects_token.clone(), + }) + ); + let mut replayed_mouse = false; + let mut replayed_keyboard = false; + loop { + match read_server_message(target_control.recv().expect("presentation effect or fence")) { + ServerMessage::MouseCapture { .. } => replayed_mouse = true, + ServerMessage::ClientShellKeyboardReportAll { .. } => replayed_keyboard = true, + ServerMessage::EndpointControl { kind, data } + if kind == crate::protocol::endpoint::PRESENTATION_EFFECTS_READY_KIND => + { + assert_eq!(data, effects_token); + assert_eq!( + activation.receive_presentation_effects_ready(&target_id, 7, &data), + crate::client::endpoint::SurfaceActivationProgress::Ready + ); + break; + } + ServerMessage::WindowTitle { .. } => {} + other => panic!("unexpected presentation fence message: {other:?}"), + } + } + assert!(replayed_mouse); + assert!(replayed_keyboard); + assert_eq!( + activation.complete(&mut shell, &mut endpoints), + Ok(crate::client::endpoint::ActivationCompletion::Activated) + ); + endpoints.unfreeze_input(); + assert_eq!(endpoints.active_id(), &target_id); + assert!(endpoints.active_surface_available()); + assert!(shell.endpoint_is_active(&target_id)); + + target_sent.lock().unwrap().clear(); + let mut returning = crate::client::endpoint::PendingEndpointActivation::begin( + &shell, + &mut endpoints, + ClientEndpointId::Local, + None, + lifecycle_resize(), + 42, + std::time::Instant::now(), + ) + .unwrap(); + dispatch_lifecycle_messages( + &mut target_server, + target_client_id, + std::mem::take(&mut *target_sent.lock().unwrap()), + ); + loop { + if let ServerMessage::ClientShellEndpointResponseChunk { + request_id, data, .. + } = read_server_message(target_control.recv().unwrap()) + { + returning.receive_response(&target_id, 7, &request_id, &data, &mut endpoints); + break; + } + } + dispatch_lifecycle_messages( + &mut source_server, + source_client_id, + std::mem::take(&mut *source_sent.lock().unwrap()), + ); + assert_eq!( + source_server.clients[&source_client_id].outer_terminal_focus, + Some(true), + "returning to Local must restore focus without a host focus event" + ); + assert_eq!( + target_server.clients[&target_client_id].outer_terminal_focus, + Some(false) + ); + shutdown_test_runtimes(&mut source_server); + shutdown_test_runtimes(&mut target_server); +} diff --git a/src/server/mod.rs b/src/server/mod.rs index 8b155df1..5d249587 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -3,6 +3,7 @@ pub mod autodetect; #[cfg(unix)] pub(crate) mod client_accept; pub(crate) mod client_commands; +mod client_endpoint_control; pub(crate) mod client_shell; pub(crate) mod client_shell_graphics; pub(crate) mod client_transport; -- cgit v1.2.3