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

3 files changed, 1037 insertions, 3 deletions

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);
+}