feat: manage multiple ssh machines from one client (#3670)

16 files changed, 1717 insertions, 214 deletions

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<bool> {
     }
 }
 
-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<String>)
     })
 }
 
+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<crate::protocol::ServerMessage> {
+    (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::<Option<Vec<_>>>()?;
+    // 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<u8>) {
         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<u8>) -> Result<(), TrySendError<Vec<u8>>> {
         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<ServerEvent>,
+    should_quit: &Arc<AtomicBool>,
+) -> 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<ServerEvent>,
     should_quit: &Arc<AtomicBool>,
+    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<bool>,
     /// 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<u64>,
     /// Request id and buffered response for a deferred worktree-created navigation.
     pub(crate) shell_deferred_navigation_request_id: Option<String>,
     pub(crate) shell_deferred_navigation_response: Option<Vec<u8>>,
@@ -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<u64, ClientConnection>) -> Option<u64> {
     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<crate::server::clients::ClientShellHeldInput>,
+    ) {
+        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<ShellFocusTarget>)> {
         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::<String, Vec<u64>>::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<api::schema::Request>,
+    ) -> 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<crate::kitty_graphics::surface::DeliveryCache>,
 }
 
 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<crate::protocol::ClientShellSnapshot> {
     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::<api::schema::ErrorResponse>(&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,6 +1419,11 @@ 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)
@@ -1394,7 +1433,17 @@ async fn deferred_worktree_response_moves_only_its_source_client() {
         .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,
         })
     );
@@ -2664,6 +2715,19 @@ fn retained_test_server(
     std::sync::mpsc::Receiver<Vec<u8>>,
     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<Vec<u8>>,
+    std::sync::mpsc::Receiver<Vec<u8>>,
+    crate::layout::PaneId,
+) {
     let mut server = test_headless_server();
     let mut workspace = crate::workspace::Workspace::test_new("test");
     let pane_id = workspace.focused_pane_id().expect("focused pane");
@@ -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<Vec<u8>>, 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]
@@ -77,6 +91,237 @@ 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 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) =
         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\\");
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<Mutex<Vec<crate::protocol::ClientMessage>>>);
+
+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::<api::schema::ErrorResponse>(&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::<api::schema::SuccessResponse>(&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::<api::schema::SuccessResponse>(&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::<Vec<_>>();
+    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<crate::protocol::ClientMessage>,
+) {
+    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::<api::schema::Request>(&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::<api::schema::Request>(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::<crate::protocol::ClientShellSnapshot>(&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;