fix: keep local usable when remote machines stall (#4234)

refs #3903

11 files changed, 713 insertions, 163 deletions

diff --git a/src/client/catalog_reload.rs b/src/client/catalog_reload.rs
index 21f33bea..a20b74dc 100644
--- a/src/client/catalog_reload.rs
+++ b/src/client/catalog_reload.rs
@@ -111,42 +111,7 @@ mod tests {
     }
 
     fn state() -> ClientState {
-        ClientState {
-            blit_encoder: render_ansi::BlitEncoder::new(),
-            mouse_capture_active: false,
-            endpoint_mouse_capture_requested: false,
-            endpoint_sgr_pixels_requested: false,
-            host_theme_updates: Vec::new(),
-            direct_mouse_capture_preference: false,
-            shell_mouse_capture_preference: false,
-            direct_keyboard_protocol: Default::default(),
-            pane_keyboard_report_all: false,
-            keyboard_report_all_active: false,
-            reported_size: (100, 30),
-            reported_cell_size: (0, 0),
-            sound_config: Default::default(),
-            kitty_graphics_enabled: false,
-            pixel_geometry_enabled: false,
-            pixel_geometry_exact: false,
-            #[cfg(unix)]
-            direct_graphics_response: Default::default(),
-            #[cfg(unix)]
-            retired_direct_graphics: None,
-            #[cfg(unix)]
-            pending_surface_graphics: HashMap::new(),
-            attach_escape: None,
-            #[cfg(unix)]
-            mouse_scroll_lines: 3,
-            remote_image_paste_key: None,
-            redraw_on_focus_gained: false,
-            repaint_pending: false,
-            presentation_frozen: false,
-            draw_host_cursor: false,
-            detached_process_children: Vec::new(),
-            shell: Some(shell::ClientShellState::new(
-                shell::ClientShellConfig::from_config(&crate::config::Config::default()),
-            )),
-        }
+        ClientState::test_new()
     }
 
     #[test]
diff --git a/src/client/endpoint/activation.rs b/src/client/endpoint/activation.rs
index a5b63344..5ab9dfa8 100644
--- a/src/client/endpoint/activation.rs
+++ b/src/client/endpoint/activation.rs
@@ -13,11 +13,31 @@ use protocol::*;
 
 const ACTIVATION_TIMEOUT: Duration = Duration::from_secs(5);
 
+fn release_surface_best_effort(
+    lease: &EndpointLease,
+    endpoints: &mut EndpointRegistry,
+    request_id: String,
+) {
+    if !endpoints.accepts(&lease.endpoint_id, lease.generation) {
+        return;
+    }
+    endpoints.set_surface_active(&lease.endpoint_id, false);
+    let _ = endpoints.send_to(
+        &lease.endpoint_id,
+        &crate::protocol::ClientMessage::ClientShellFocus { focused: false },
+    );
+    match surface_interest_request(&lease.boot_id, request_id, false) {
+        Ok(request) => {
+            let _ = endpoints.send_to(&lease.endpoint_id, &request);
+        }
+        Err(error) => tracing::warn!(%error, "could not request abandoned surface cleanup"),
+    }
+}
+
 impl PendingEndpointActivation {
-    #[allow(clippy::too_many_arguments)]
-    pub(crate) fn begin(
+    pub(crate) fn prepare(
         shell: &crate::client::shell::ClientShellState,
-        endpoints: &mut EndpointRegistry,
+        endpoints: &EndpointRegistry,
         target: ClientEndpointId,
         focus: Option<crate::client::shell::ClientEndpointFocusTarget>,
         resize: crate::protocol::ClientMessage,
@@ -60,16 +80,14 @@ impl PendingEndpointActivation {
         let source_is_target = source.endpoint_id == target_lease.endpoint_id;
         // Validate every typed lifecycle and optional focus envelope before the first transport
         // write. Any error above this line is guaranteed not to have changed either endpoint.
-        let source_release_request = (source_available && !source_is_target)
-            .then(|| {
-                surface_interest_request(
-                    &source.boot_id,
-                    format!("client-shell-surface:{serial}:off"),
-                    false,
-                )
-            })
-            .transpose()
+        if source_available && !source_is_target {
+            surface_interest_request(
+                &source.boot_id,
+                format!("client-shell-surface:{serial}:off"),
+                false,
+            )
             .map_err(|error| ActivationBeginError::Preflight(error.to_string()))?;
+        }
         surface_interest_request(
             &target_lease.boot_id,
             format!("client-shell-surface:{serial}:on"),
@@ -85,14 +103,13 @@ impl PendingEndpointActivation {
             .map_err(|error| ActivationBeginError::Preflight(error.to_string()))?;
         }
 
-        endpoints.freeze_input();
-        let mut activation = Self {
+        Ok(Self {
             source,
             source_available,
             target: target_lease,
             focus,
             host_focused: shell.host_focus_baseline(),
-            resize: resize.clone(),
+            resize,
             phase: ActivationPhase::ReleasingSource {
                 request_id: format!("client-shell-surface:{serial}:off"),
             },
@@ -101,44 +118,82 @@ impl PendingEndpointActivation {
             next_focus_serial: 0,
             rollback_error: None,
             successor: None,
-        };
+        })
+    }
 
-        // Reconnecting the selected endpoint has no live source surface to release. All normal
-        // handoffs must make the source locally inactive before a target request is even sent.
-        if source_is_target || !source_available {
-            if let Err(error) = activation.start_target(endpoints, resize) {
-                return Err(ActivationBeginError::Partial {
-                    activation: Box::new(activation),
-                    error,
-                });
-            }
-        } else {
-            // `surface.set(false)` removes the viewer, but old servers only emit the PTY focus
-            // loss while the viewer is still active. Revoke it explicitly before source-off.
-            if endpoints.send_to(
-                &activation.source.endpoint_id,
-                &crate::protocol::ClientMessage::ClientShellFocus { focused: false },
-            ) != EndpointSendOutcome::Sent
-            {
-                return Err(ActivationBeginError::Partial {
-                    activation: Box::new(activation),
-                    error: "source endpoint focus revoke could not be sent".into(),
-                });
+    pub(crate) fn start(
+        mut self,
+        endpoints: &mut EndpointRegistry,
+    ) -> Result<Self, ActivationBeginError> {
+        endpoints.freeze_input();
+        match self.start_prepared(endpoints) {
+            Ok(()) => Ok(self),
+            Err(error) => Err(ActivationBeginError::Partial {
+                activation: Box::new(self),
+                error,
+            }),
+        }
+    }
+
+    fn start_prepared(&mut self, endpoints: &mut EndpointRegistry) -> Result<(), String> {
+        let source_is_target = self.source.endpoint_id == self.target.endpoint_id;
+        // Local must not depend on a remote acknowledgement to become usable.
+        if source_is_target || !self.source_available || self.target.endpoint_id.is_local() {
+            if self.source_available && !source_is_target {
+                release_surface_best_effort(
+                    &self.source,
+                    endpoints,
+                    format!("client-shell-surface:{}:off", self.epoch),
+                );
             }
-            let request = source_release_request.expect("validated source release request");
-            if endpoints.send_to(&activation.source.endpoint_id, &request)
-                != EndpointSendOutcome::Sent
-            {
-                return Err(ActivationBeginError::Partial {
-                    activation: Box::new(activation),
-                    error: "source endpoint release could not be sent".into(),
-                });
+            return self.start_target(endpoints, self.resize.clone());
+        }
+        // Old servers emit PTY focus loss only while the viewer is still active.
+        if endpoints.send_to(
+            &self.source.endpoint_id,
+            &crate::protocol::ClientMessage::ClientShellFocus { focused: false },
+        ) != EndpointSendOutcome::Sent
+        {
+            return Err("source endpoint focus revoke could not be sent".into());
+        }
+        let request = surface_interest_request(
+            &self.source.boot_id,
+            format!("client-shell-surface:{}:off", self.epoch),
+            false,
+        )
+        .map_err(|error| error.to_string())?;
+        if endpoints.send_to(&self.source.endpoint_id, &request) != EndpointSendOutcome::Sent {
+            return Err("source endpoint release could not be sent".into());
+        }
+        endpoints.set_surface_active(&self.source.endpoint_id, false);
+        Ok(())
+    }
+
+    #[cfg(test)]
+    pub(crate) fn begin(
+        shell: &crate::client::shell::ClientShellState,
+        endpoints: &mut EndpointRegistry,
+        target: ClientEndpointId,
+        focus: Option<crate::client::shell::ClientEndpointFocusTarget>,
+        resize: crate::protocol::ClientMessage,
+        serial: u64,
+        now: Instant,
+    ) -> Result<Self, ActivationBeginError> {
+        Self::prepare(shell, endpoints, target, focus, resize, serial, now)?.start(endpoints)
+    }
+
+    pub(crate) fn abandon(&self, endpoints: &mut EndpointRegistry) {
+        endpoints.freeze_input();
+        for lease in [&self.source, &self.target] {
+            release_surface_best_effort(
+                lease,
+                endpoints,
+                format!("client-shell-surface:{}:abandon", self.epoch),
+            );
+            if self.source.endpoint_id == self.target.endpoint_id {
+                break;
             }
-            // This is deliberately before the acknowledgement: the source is no longer viewed
-            // locally while its release is in flight, and pane input is consequently blocked.
-            endpoints.set_surface_active(&activation.source.endpoint_id, false);
         }
-        Ok(activation)
     }
 
     pub(crate) fn target(&self) -> &ClientEndpointId {
diff --git a/src/client/endpoint/activation_tests.rs b/src/client/endpoint/activation_tests.rs
index b66cb98b..7201940a 100644
--- a/src/client/endpoint/activation_tests.rs
+++ b/src/client/endpoint/activation_tests.rs
@@ -976,14 +976,33 @@ fn rapid_a_to_b_to_a_restores_source_before_a_fresh_latest_epoch() {
 }
 
 #[test]
-fn disconnected_committed_source_does_not_block_switching_to_a_live_endpoint() {
-    let (mut shell, mut endpoints, local_sent, _remote_sent) = shell_and_registry();
+fn local_activation_does_not_wait_for_a_disconnected_stalled_or_failed_remote() {
+    for source_state in ["disconnected", "stalled", "write-failed"] {
+        local_escape(source_state);
+    }
+}
+
+fn local_escape(source_state: &str) {
+    let (mut shell, mut endpoints, local_sent, remote_sent) = shell_and_registry();
     let disconnected = endpoint();
     endpoints.set_surface_active(&ClientEndpointId::Local, false);
     endpoints.set_surface_active(&disconnected, true);
     assert!(endpoints.set_active(&disconnected));
     assert!(shell.activate_endpoint_projection(&disconnected));
-    endpoints.disconnect(&disconnected);
+    match source_state {
+        "disconnected" => endpoints.disconnect(&disconnected),
+        "write-failed" => endpoints.insert(
+            disconnected.clone(),
+            FakeTransport {
+                sent: remote_sent,
+                fail_after_write: true,
+            },
+            7,
+            negotiation(),
+            true,
+        ),
+        _ => {}
+    }
 
     let mut activation = PendingEndpointActivation::begin(
         &shell,
@@ -995,16 +1014,28 @@ fn disconnected_committed_source_does_not_block_switching_to_a_live_endpoint() {
         Instant::now(),
     )
     .unwrap();
-    assert_eq!(activation.source_command_lane(), None);
+    let local_activations = local_sent
+        .lock()
+        .unwrap()
+        .iter()
+        .filter_map(surface_set_active)
+        .collect::<Vec<_>>();
+    assert_eq!(local_activations, vec![true], "Local must not wait for SSH");
+    assert!(!endpoints.active_surface_available());
+    assert_eq!(endpoints.active_id(), &disconnected);
     assert_eq!(
-        local_sent
-            .lock()
-            .unwrap()
-            .iter()
-            .filter_map(surface_set_active)
-            .collect::<Vec<_>>(),
-        vec![true],
-        "there is no unavailable source release to await"
+        activation.receive_response(
+            &disconnected,
+            7,
+            "client-shell-surface:22:off",
+            &surface_success("client-shell-surface:22:off", false, 1),
+            &mut endpoints,
+        ),
+        SurfaceActivationProgress::Stale
+    );
+    assert_eq!(
+        activation.receive_surface(&disconnected, 7, surface("remote-boot", 2, "stale")),
+        SurfaceActivationProgress::Stale
     );
 
     assert_eq!(
@@ -1082,6 +1113,261 @@ fn disconnected_committed_source_does_not_block_switching_to_a_live_endpoint() {
     );
     assert_eq!(endpoints.active_id(), &ClientEndpointId::Local);
     assert_ne!(endpoints.active_id(), &disconnected);
+    assert!(
+        !endpoints.active_surface_available(),
+        "runtime opens input only after the effects fence"
+    );
+}
+
+#[test]
+fn local_selection_abandons_every_unfinished_remote_handoff_phase() {
+    use crate::client::{
+        endpoint_commands::EndpointCommands, shell_runtime::begin_endpoint_activation, ClientState,
+    };
+    for phase in ["release", "target", "rollback", "restore"] {
+        let (shell, mut endpoints, local_sent, _remote_sent) = shell_and_registry();
+        let mut abandoned = PendingEndpointActivation::begin(
+            &shell,
+            &mut endpoints,
+            endpoint(),
+            None,
+            resize(),
+            30,
+            Instant::now(),
+        )
+        .unwrap();
+        if phase != "release" {
+            abandoned.receive_response(
+                &ClientEndpointId::Local,
+                1,
+                "client-shell-surface:30:off",
+                &surface_success("client-shell-surface:30:off", false, 1),
+                &mut endpoints,
+            );
+        }
+        if matches!(phase, "rollback" | "restore") {
+            assert_eq!(
+                abandoned.rollback(&mut endpoints, "cancel".into(), false),
+                ActivationRollback::Pending
+            );
+        }
+        if phase == "restore" {
+            abandoned.receive_response(
+                &endpoint(),
+                7,
+                "client-shell-surface:30:rollback-target-off",
+                &surface_success("client-shell-surface:30:rollback-target-off", false, 1),
+                &mut endpoints,
+            );
+        }
+        local_sent.lock().unwrap().clear();
+        let mut state = ClientState::test_new();
+        state.shell = Some(shell);
+        let mut commands = EndpointCommands::default();
+        let mut pending = Some(abandoned);
+        let mut serial = 31;
+        let mut scheduled = None;
+        for _ in 0..2 {
+            begin_endpoint_activation(
+                &mut state,
+                &mut endpoints,
+                &mut commands,
+                &mut pending,
+                &mut serial,
+                ClientEndpointId::Local,
+                None,
+                false,
+                Instant::now(),
+                &mut scheduled,
+            )
+            .unwrap();
+        }
+        assert_eq!(serial, 32, "repeated Local selection must coalesce");
+        let local = pending.as_mut().unwrap();
+        assert_eq!(local.target(), &ClientEndpointId::Local);
+        assert!(!endpoints.active_surface_available());
+        assert!(state.presentation_frozen);
+        let activations = local_sent
+            .lock()
+            .unwrap()
+            .iter()
+            .filter_map(surface_set_active)
+            .collect::<Vec<_>>();
+        assert_eq!(activations, vec![false, true], "phase: {phase}");
+        assert!(!local.accepts_response(
+            &ClientEndpointId::Local,
+            1,
+            "local-boot",
+            "client-shell-surface:30:off"
+        ));
+        assert_eq!(
+            local.receive_surface(&endpoint(), 7, surface("remote-boot", 2, "stale")),
+            SurfaceActivationProgress::Stale
+        );
+    }
+}
+
+#[test]
+fn local_selection_waits_for_fresh_metadata_without_abandoning_remote() {
+    use crate::client::{
+        endpoint_commands::EndpointCommands,
+        shell_runtime::{begin_endpoint_activation, take_ready_local_activation},
+        ClientLoopEvent, ClientState,
+    };
+    for replaced_generation in [false, true] {
+        let (mut shell, mut endpoints, local_sent, remote_sent) = shell_and_registry();
+        shell.set_endpoint_snapshot_for_generation(
+            &ClientEndpointId::Local,
+            1,
+            Box::new(test_snapshot("local-boot", 1)),
+        );
+        let mut state = ClientState::test_new();
+        state.shell = Some(shell);
+        let mut commands = EndpointCommands::default();
+        let mut pending = None;
+        let mut serial = 40;
+        let mut scheduled = None;
+        begin_endpoint_activation(
+            &mut state,
+            &mut endpoints,
+            &mut commands,
+            &mut pending,
+            &mut serial,
+            endpoint(),
+            None,
+            false,
+            Instant::now(),
+            &mut scheduled,
+        )
+        .unwrap();
+        if replaced_generation {
+            endpoints.insert(
+                ClientEndpointId::Local,
+                FakeTransport {
+                    sent: local_sent.clone(),
+                    fail_after_write: false,
+                },
+                2,
+                negotiation(),
+                false,
+            );
+        } else {
+            endpoints.disconnect(&ClientEndpointId::Local);
+        }
+        begin_endpoint_activation(
+            &mut state,
+            &mut endpoints,
+            &mut commands,
+            &mut pending,
+            &mut serial,
+            ClientEndpointId::Local,
+            Some(crate::client::shell::ClientEndpointFocusTarget::Workspace(
+                "selected-local".into(),
+            )),
+            false,
+            Instant::now(),
+            &mut scheduled,
+        )
+        .unwrap();
+        assert_eq!(
+            pending.as_ref().map(PendingEndpointActivation::target),
+            Some(&endpoint())
+        );
+        assert!(remote_sent.lock().unwrap().is_empty());
+        assert_eq!(serial, 41);
+        assert!(state.deferred_local_activation.is_some());
+        assert!(take_ready_local_activation(&mut state, &endpoints).is_none());
+        if !replaced_generation {
+            endpoints.insert(
+                ClientEndpointId::Local,
+                FakeTransport {
+                    sent: local_sent,
+                    fail_after_write: false,
+                },
+                2,
+                negotiation(),
+                false,
+            );
+        }
+        assert!(take_ready_local_activation(&mut state, &endpoints).is_none());
+        state
+            .shell
+            .as_mut()
+            .unwrap()
+            .cache_endpoint_snapshot_inactive_for_generation(
+                &ClientEndpointId::Local,
+                2,
+                Box::new(test_snapshot("local-boot", 1)),
+            );
+        let event = take_ready_local_activation(&mut state, &endpoints).unwrap();
+        let ClientLoopEvent::ActivateEndpoint {
+            endpoint_id,
+            target,
+            force,
+        } = event
+        else {
+            panic!("expected retained Local selection");
+        };
+        assert_eq!(endpoint_id, ClientEndpointId::Local);
+        assert_eq!(
+            target,
+            Some(crate::client::shell::ClientEndpointFocusTarget::Workspace(
+                "selected-local".into()
+            ))
+        );
+        begin_endpoint_activation(
+            &mut state,
+            &mut endpoints,
+            &mut commands,
+            &mut pending,
+            &mut serial,
+            endpoint_id,
+            target,
+            force,
+            Instant::now(),
+            &mut scheduled,
+        )
+        .unwrap();
+        assert_eq!(pending.as_ref().unwrap().target(), &ClientEndpointId::Local);
+        assert_eq!(serial, 42);
+        assert!(!endpoints.active_surface_available());
+        assert!(state.deferred_local_activation.is_none());
+    }
+}
+
+#[test]
+fn newer_remote_selection_cancels_deferred_local_selection() {
+    use crate::client::{
+        endpoint_commands::EndpointCommands, shell_runtime::begin_endpoint_activation, ClientState,
+    };
+    let (shell, mut endpoints, _, _) = shell_and_registry();
+    let mut state = ClientState::test_new();
+    state.shell = Some(shell);
+    let mut commands = EndpointCommands::default();
+    let mut pending = None;
+    let mut serial = 50;
+    let mut scheduled = None;
+    endpoints.disconnect(&ClientEndpointId::Local);
+    for endpoint_id in [ClientEndpointId::Local, endpoint()] {
+        begin_endpoint_activation(
+            &mut state,
+            &mut endpoints,
+            &mut commands,
+            &mut pending,
+            &mut serial,
+            endpoint_id.clone(),
+            None,
+            false,
+            Instant::now(),
+            &mut scheduled,
+        )
+        .unwrap();
+        assert_eq!(
+            state.deferred_local_activation.is_some(),
+            endpoint_id.is_local()
+        );
+    }
+    assert_eq!(pending.as_ref().unwrap().target(), &endpoint());
 }
 
 #[test]
diff --git a/src/client/mod.rs b/src/client/mod.rs
index 9fff9719..cb465096 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -414,6 +414,7 @@ async fn run_client_loop(
         redraw_on_focus_gained: config.redraw_on_focus_gained,
         repaint_pending: false,
         presentation_frozen: false,
+        deferred_local_activation: None,
         draw_host_cursor,
         detached_process_children: Vec::new(),
         shell: config.shell_config.map(shell::ClientShellState::new),
@@ -819,7 +820,7 @@ async fn run_client_loop(
                         &mut pending_activation,
                         &mut endpoint_commands,
                         &mut prefix_input_source,
-                        &event_tx,
+                        &mut scheduled_activation,
                     )? {
                         return Ok(());
                     }
@@ -983,7 +984,7 @@ async fn run_client_loop(
                         &mut pending_activation,
                         &mut endpoint_commands,
                         &mut prefix_input_source,
-                        &event_tx,
+                        &mut scheduled_activation,
                     )? {
                         return Ok(());
                     }
@@ -1087,7 +1088,7 @@ async fn run_client_loop(
                         &mut pending_activation,
                         &mut endpoint_commands,
                         &mut prefix_input_source,
-                        &event_tx,
+                        &mut scheduled_activation,
                     )? {
                         return Ok(());
                     }
@@ -1263,7 +1264,7 @@ async fn run_client_loop(
                     target,
                     force,
                     now,
-                    &event_tx,
+                    &mut scheduled_activation,
                 )?;
             }
             ClientLoopEvent::ServerMessage {
@@ -1744,7 +1745,7 @@ async fn run_client_loop(
                             &mut write_stream,
                             state.shell.as_mut(),
                             &mut state.detached_process_children,
-                            &event_tx,
+                            &mut scheduled_activation,
                         )?;
                         let repaint = repaint || dispatch_repaint;
                         if replay_mouse.is_empty() {
@@ -1776,7 +1777,7 @@ async fn run_client_loop(
                                 &mut pending_activation,
                                 &mut endpoint_commands,
                                 &mut prefix_input_source,
-                                &event_tx,
+                                &mut scheduled_activation,
                             )? {
                                 return Ok(());
                             }
@@ -1962,6 +1963,14 @@ async fn run_client_loop(
                             }
                         }
                         write_stream.mark_ready(&endpoint_id, generation);
+                        if endpoint_id.is_local() {
+                            if let Some(event) =
+                                take_ready_local_activation(&mut state, &write_stream)
+                            {
+                                scheduled_activation = Some(event);
+                                continue;
+                            }
+                        }
                         let selected_endpoint = endpoint_catalog
                             .selected_profile
                             .as_ref()
@@ -1978,7 +1987,11 @@ async fn run_client_loop(
                         let needs_surface = write_stream
                             .connection(&selected_endpoint)
                             .is_some_and(|connection| !connection.surface_active);
-                        if activation_ready && needs_surface && pending_activation.is_none() {
+                        if activation_ready
+                            && needs_surface
+                            && pending_activation.is_none()
+                            && state.deferred_local_activation.is_none()
+                        {
                             scheduled_activation = Some(ClientLoopEvent::ActivateEndpoint {
                                 endpoint_id: selected_endpoint,
                                 target: None,
@@ -2113,7 +2126,7 @@ async fn run_client_loop(
                         &mut pending_activation,
                         &mut endpoint_commands,
                         &mut prefix_input_source,
-                        &event_tx,
+                        &mut scheduled_activation,
                     )? {
                         return Ok(());
                     }
diff --git a/src/client/shell/endpoint_navigation.rs b/src/client/shell/endpoint_navigation.rs
index 38ad0621..542bf3f9 100644
--- a/src/client/shell/endpoint_navigation.rs
+++ b/src/client/shell/endpoint_navigation.rs
@@ -35,23 +35,11 @@ impl ClientShellState {
         press: ClientWorkspacePress,
         outcome: &mut ClientShellInput,
     ) {
-        if press.endpoint_id == self.active_endpoint_id {
-            self.push_endpoint_method(
-                crate::api::schema::Method::WorkspaceFocus(crate::api::schema::WorkspaceTarget {
-                    workspace_id: press.workspace_id,
-                }),
-                outcome,
-            );
-        } else if self.endpoint_is_online(&press.endpoint_id) {
-            outcome.actions.push(ClientShellAction::ActivateEndpoint {
-                endpoint_id: press.endpoint_id,
-                target: Some(ClientEndpointFocusTarget::Workspace(press.workspace_id)),
-            });
-        } else {
-            let label = self.endpoint_label(&press.endpoint_id).to_owned();
-            self.receive_endpoint_unavailable(format!("{label} is not ready"));
-            outcome.repaint = true;
-        }
+        self.focus_or_activate(
+            press.endpoint_id,
+            ClientEndpointFocusTarget::Workspace(press.workspace_id),
+            outcome,
+        );
     }
 
     pub(super) fn handle_endpoint_machine_click(
@@ -68,12 +56,16 @@ impl ClientShellState {
             return false;
         };
         let endpoint_id = hit.endpoint_id.clone();
-        if super::contains(hit.collapse_toggle, point) || endpoint_id == self.active_endpoint_id {
+        let collapse_toggle = super::contains(hit.collapse_toggle, point);
+        if collapse_toggle || endpoint_id == self.active_endpoint_id {
             if !self.collapsed_endpoints.remove(&endpoint_id) {
-                self.collapsed_endpoints.insert(endpoint_id);
+                self.collapsed_endpoints.insert(endpoint_id.clone());
             }
             outcome.repaint = true;
-        } else if self.endpoint_is_online(&endpoint_id) {
+            if !collapse_toggle && endpoint_id.is_local() {
+                self.activate_endpoint(endpoint_id, outcome);
+            }
+        } else if endpoint_id.is_local() || self.endpoint_is_online(&endpoint_id) {
             outcome.actions.push(ClientShellAction::ActivateEndpoint {
                 endpoint_id,
                 target: None,
@@ -236,13 +228,16 @@ impl ClientShellState {
         endpoint_id: ClientEndpointId,
         outcome: &mut ClientShellInput,
     ) -> bool {
-        if !self.endpoint_is_online(&endpoint_id) {
+        let online = self.endpoint_is_online(&endpoint_id);
+        if !online && !endpoint_id.is_local() {
             let label = self.endpoint_label(&endpoint_id).to_owned();
             self.receive_endpoint_unavailable(format!("{label} is not ready"));
             outcome.repaint = true;
             return false;
         }
-        if endpoint_id != self.active_endpoint_id {
+        if (endpoint_id.is_local() && (self.multi_endpoint_active() || !online))
+            || endpoint_id != self.active_endpoint_id
+        {
             outcome.actions.push(ClientShellAction::ActivateEndpoint {
                 endpoint_id,
                 target: None,
@@ -257,13 +252,18 @@ impl ClientShellState {
         target: ClientEndpointFocusTarget,
         outcome: &mut ClientShellInput,
     ) -> bool {
-        if !self.endpoint_is_online(&endpoint_id) {
+        let online = self.endpoint_is_online(&endpoint_id);
+        if !online && !endpoint_id.is_local() {
             let label = self.endpoint_label(&endpoint_id).to_owned();
             self.receive_endpoint_unavailable(format!("{label} is not ready"));
             outcome.repaint = true;
             return false;
         }
-        if endpoint_id == self.active_endpoint_id {
+        // Local can still be displayed while a remote activation is pending.
+        // Route explicit selections through the runtime so they can cancel that handoff.
+        if endpoint_id == self.active_endpoint_id
+            && !(endpoint_id.is_local() && (self.multi_endpoint_active() || !online))
+        {
             let method = match target {
                 ClientEndpointFocusTarget::Workspace(workspace_id) => {
                     crate::api::schema::Method::WorkspaceFocus(
diff --git a/src/client/shell/mobile.rs b/src/client/shell/mobile.rs
index 185bef51..8f06f54b 100644
--- a/src/client/shell/mobile.rs
+++ b/src/client/shell/mobile.rs
@@ -1000,16 +1000,12 @@ impl ClientShellState {
                 if endpoint_id == self.active_endpoint_id {
                     self.mode = ClientShellMode::Terminal;
                     self.navigate_workspace_id = None;
-                } else if self.endpoint_is_online(&endpoint_id) {
+                    if endpoint_id.is_local() {
+                        self.activate_endpoint(endpoint_id, outcome);
+                    }
+                } else if self.activate_endpoint(endpoint_id, outcome) {
                     self.mode = ClientShellMode::Terminal;
                     self.navigate_workspace_id = None;
-                    outcome.actions.push(ClientShellAction::ActivateEndpoint {
-                        endpoint_id,
-                        target: None,
-                    });
-                } else {
-                    let label = self.endpoint_label(&endpoint_id).to_owned();
-                    self.receive_endpoint_unavailable(format!("{label} is not ready"));
                 }
             }
             Some(ClientMobileTarget::NewWorkspace) => {
diff --git a/src/client/shell/tests/endpoint_requests.rs b/src/client/shell/tests/endpoint_requests.rs
index b8805596..1f51f06c 100644
--- a/src/client/shell/tests/endpoint_requests.rs
+++ b/src/client/shell/tests/endpoint_requests.rs
@@ -259,6 +259,42 @@ impl crate::client::endpoint::EndpointTransport for TestTransport {
 }
 
 #[test]
+fn local_selection_is_scheduled_ahead_of_a_full_event_queue() {
+    use crate::client::{
+        endpoint::{EndpointNegotiation, EndpointRegistry},
+        endpoint_commands::EndpointCommands,
+        ClientLoopEvent,
+    };
+    let mut endpoints = EndpointRegistry::new(
+        TestTransport { fail: false },
+        1,
+        EndpointNegotiation::default(),
+    );
+    let mut commands = EndpointCommands::default();
+    let (tx, mut rx) = tokio::sync::mpsc::channel(1);
+    tx.try_send(ClientLoopEvent::Timer).unwrap();
+    let mut scheduled = None;
+    crate::client::shell_runtime::dispatch_client_shell_actions(
+        vec![ClientShellAction::ActivateEndpoint {
+            endpoint_id: ClientEndpointId::Local,
+            target: Some(ClientEndpointFocusTarget::Workspace("ws_1".into())),
+        }],
+        &mut commands,
+        &mut endpoints,
+        None,
+        &mut Vec::new(),
+        &mut scheduled,
+    )
+    .unwrap();
+    let next = scheduled.take().or_else(|| rx.try_recv().ok());
+    assert!(matches!(next, Some(ClientLoopEvent::ActivateEndpoint {
+        endpoint_id: ClientEndpointId::Local,
+        target: Some(ClientEndpointFocusTarget::Workspace(id)), ..
+    }) if id == "ws_1"));
+    assert!(matches!(rx.try_recv(), Ok(ClientLoopEvent::Timer)));
+}
+
+#[test]
 fn dispatcher_cancels_worktree_requests_on_frozen_surface_or_failed_send() {
     use crate::client::endpoint::{EndpointNegotiation, EndpointRegistry};
     use crate::client::endpoint_commands::EndpointCommands;
@@ -272,14 +308,14 @@ fn dispatcher_cancels_worktree_requests_on_frozen_surface_or_failed_send() {
         );
         endpoints.set_surface_active(&ClientEndpointId::Local, fail_send);
         let mut commands = EndpointCommands::default();
-        let (tx, _rx) = tokio::sync::mpsc::channel(16);
+        let mut scheduled = None;
         let (replay, repaint) = crate::client::shell_runtime::dispatch_client_shell_actions(
             actions,
             &mut commands,
             &mut endpoints,
             Some(&mut state),
             &mut Vec::new(),
-            &tx,
+            &mut scheduled,
         )
         .unwrap();
         assert!(repaint);
diff --git a/src/client/shell/tests/endpoints.rs b/src/client/shell/tests/endpoints.rs
index de7abb3d..5c00e645 100644
--- a/src/client/shell/tests/endpoints.rs
+++ b/src/client/shell/tests/endpoints.rs
@@ -1398,6 +1398,71 @@ fn clicking_remote_machine_name_requests_activation_without_mutating_projection(
 }
 
 #[test]
+fn clicking_local_can_cancel_a_remote_switch_while_local_is_still_displayed() {
+    for workspace in [false, true] {
+        let (mut state, remote) = state_with_remote();
+        state.compose(100, 28).unwrap();
+        let mut pending = ClientShellInput::default();
+        assert!(state.activate_endpoint(remote, &mut pending));
+        assert_eq!(state.active_endpoint_id, ClientEndpointId::Local);
+        let rect = if workspace {
+            state
+                .hits
+                .workspaces
+                .iter()
+                .find(|hit| hit.endpoint_id.is_local())
+                .unwrap()
+                .rect
+        } else {
+            state
+                .hits
+                .machines
+                .iter()
+                .find(|hit| hit.endpoint_id.is_local())
+                .unwrap()
+                .rect
+        };
+        let outcome = state.handle_raw_events(vec![
+            RawInputEvent::Mouse(MouseEvent {
+                kind: MouseEventKind::Down(MouseButton::Left),
+                column: rect.x + 5,
+                row: rect.y,
+                modifiers: KeyModifiers::empty(),
+            }),
+            RawInputEvent::Mouse(MouseEvent {
+                kind: MouseEventKind::Up(MouseButton::Left),
+                column: rect.x + 5,
+                row: rect.y,
+                modifiers: KeyModifiers::empty(),
+            }),
+        ]);
+        assert!(
+            matches!(outcome.actions.as_slice(), [ClientShellAction::ActivateEndpoint {
+            endpoint_id: ClientEndpointId::Local, target,
+        }] if target.is_some() == workspace)
+        );
+    }
+}
+
+#[test]
+fn reconnecting_local_selection_still_reaches_the_runtime() {
+    let (mut state, _) = state_with_remote();
+    state.mark_endpoint_disconnected(&ClientEndpointId::Local);
+    let mut outcome = ClientShellInput::default();
+    state.focus_or_activate(
+        ClientEndpointId::Local,
+        ClientEndpointFocusTarget::Workspace("local-workspace".into()),
+        &mut outcome,
+    );
+    assert!(
+        matches!(outcome.actions.as_slice(), [ClientShellAction::ActivateEndpoint {
+        endpoint_id: ClientEndpointId::Local,
+        target: Some(ClientEndpointFocusTarget::Workspace(id)),
+    }] if id == "local-workspace")
+    );
+}
+
+#[test]
 fn machine_arrow_toggles_inactive_machine_without_switching() {
     for sidebar_collapsed in [false, true] {
         for status in [
diff --git a/src/client/shell/tests/mobile.rs b/src/client/shell/tests/mobile.rs
index 0c6f1908..4dac5490 100644
--- a/src/client/shell/tests/mobile.rs
+++ b/src/client/shell/tests/mobile.rs
@@ -119,6 +119,31 @@ fn mobile_switcher_can_activate_an_online_saved_machine() {
         }] if activated == &endpoint_id
     ));
 
+    // Local is still projected until the remote handoff completes.
+    state.mode = ClientShellMode::Navigate;
+    state.compose(44, 30).unwrap();
+    let local = state
+        .hits
+        .mobile_targets
+        .iter()
+        .find_map(|(rect, target)| {
+            matches!(target, ClientMobileTarget::Machine(ClientEndpointId::Local)).then_some(*rect)
+        })
+        .unwrap();
+    let outcome = state.handle_raw_events(vec![RawInputEvent::Mouse(MouseEvent {
+        kind: MouseEventKind::Down(MouseButton::Left),
+        column: local.x,
+        row: local.y,
+        modifiers: KeyModifiers::empty(),
+    })]);
+    assert!(matches!(
+        outcome.actions.as_slice(),
+        [ClientShellAction::ActivateEndpoint {
+            endpoint_id: ClientEndpointId::Local,
+            target: None,
+        }]
+    ));
+
     state.set_endpoint_status(&endpoint_id, ClientEndpointStatus::Online);
     assert!(state.activate_endpoint_projection(&endpoint_id));
     let mut remote_surface = surface();
diff --git a/src/client/shell_runtime.rs b/src/client/shell_runtime.rs
index 00bb1406..2e8d1c72 100644
--- a/src/client/shell_runtime.rs
+++ b/src/client/shell_runtime.rs
@@ -6,7 +6,7 @@ pub(super) fn dispatch_client_shell_actions(
     endpoints: &mut endpoint::EndpointRegistry,
     mut shell: Option<&mut shell::ClientShellState>,
     detached_process_children: &mut Vec<std::process::Child>,
-    event_tx: &tokio::sync::mpsc::Sender<ClientLoopEvent>,
+    scheduled_activation: &mut Option<ClientLoopEvent>,
 ) -> Result<(Vec<crossterm::event::MouseEvent>, bool), ClientError> {
     let mut replay_mouse = Vec::new();
     let mut repaint = false;
@@ -32,7 +32,7 @@ pub(super) fn dispatch_client_shell_actions(
                 endpoint_id,
                 target,
             } => {
-                let _ = event_tx.try_send(ClientLoopEvent::ActivateEndpoint {
+                *scheduled_activation = Some(ClientLoopEvent::ActivateEndpoint {
                     endpoint_id,
                     target,
                     force: false,
@@ -169,6 +169,41 @@ fn install_pending_activation(
     *pending = Some(activation);
 }
 
+fn local_activation_metadata_ready(
+    state: &ClientState,
+    endpoints: &endpoint::EndpointRegistry,
+) -> bool {
+    endpoints
+        .connection(&endpoint::ClientEndpointId::Local)
+        .is_some_and(|connection| {
+            state.shell.as_ref().is_some_and(|shell| {
+                shell
+                    .endpoint_snapshot_identity(
+                        &endpoint::ClientEndpointId::Local,
+                        connection.generation,
+                    )
+                    .is_some()
+            })
+        })
+}
+
+pub(super) fn take_ready_local_activation(
+    state: &mut ClientState,
+    endpoints: &endpoint::EndpointRegistry,
+) -> Option<ClientLoopEvent> {
+    if !local_activation_metadata_ready(state, endpoints) {
+        return None;
+    }
+    state
+        .deferred_local_activation
+        .take()
+        .map(|intent| ClientLoopEvent::ActivateEndpoint {
+            endpoint_id: intent.endpoint_id,
+            target: intent.target,
+            force: false,
+        })
+}
+
 pub(super) fn begin_endpoint_activation(
     state: &mut ClientState,
     endpoints: &mut endpoint::EndpointRegistry,
@@ -179,26 +214,46 @@ pub(super) fn begin_endpoint_activation(
     target: Option<shell::ClientEndpointFocusTarget>,
     force: bool,
     now: std::time::Instant,
-    event_tx: &tokio::sync::mpsc::Sender<ClientLoopEvent>,
+    scheduled_activation: &mut Option<ClientLoopEvent>,
 ) -> Result<(), ClientError> {
-    if let Some(activation) = pending.as_mut() {
-        if activation.can_retarget(&endpoint_id) {
-            let retarget_error = activation.retarget(target, endpoints).err();
-            if let Some(error) = retarget_error {
-                rollback_endpoint_activation(state, endpoints, pending, error, false);
-            }
-        } else {
-            // Once rollback starts, even a request for the original target is a new intent. It
-            // replaces the retained successor instead of mutating the transaction being retired.
-            let outcome = activation.supersede(endpoint_id, target, endpoints);
-            if let endpoint::ActivationRollback::Unavailable(message) = outcome {
-                *pending = None;
-                present_handoff_unavailable(state, message);
-            }
+    state.deferred_local_activation = None;
+    if endpoint_id.is_local() && !local_activation_metadata_ready(state, endpoints) {
+        state.deferred_local_activation = Some(endpoint::EndpointActivationIntent {
+            endpoint_id,
+            target,
+        });
+        if let Some(shell) = state.shell.as_mut() {
+            shell.receive_endpoint_unavailable(
+                "Local is reconnecting; selection will resume when it is ready".into(),
+            );
         }
         return Ok(());
     }
-    let already_active = !force
+    let replace_pending = endpoint_id.is_local()
+        && pending
+            .as_ref()
+            .is_some_and(|activation| !activation.can_retarget(&endpoint_id));
+    if !replace_pending {
+        if let Some(activation) = pending.as_mut() {
+            if activation.can_retarget(&endpoint_id) {
+                let retarget_error = activation.retarget(target, endpoints).err();
+                if let Some(error) = retarget_error {
+                    rollback_endpoint_activation(state, endpoints, pending, error, false);
+                }
+            } else {
+                // Once rollback starts, even a request for the original target is a new intent.
+                // Retain it until restoration finishes; Local can instead abandon this handoff.
+                let outcome = activation.supersede(endpoint_id, target, endpoints);
+                if let endpoint::ActivationRollback::Unavailable(message) = outcome {
+                    *pending = None;
+                    present_handoff_unavailable(state, message);
+                }
+            }
+            return Ok(());
+        }
+    }
+    let already_active = !replace_pending
+        && !force
         && endpoints.active_id() == &endpoint_id
         && endpoints
             .connection(&endpoint_id)
@@ -212,7 +267,7 @@ pub(super) fn begin_endpoint_activation(
                 endpoints,
                 Some(shell),
                 &mut state.detached_process_children,
-                event_tx,
+                scheduled_activation,
             )?;
             if repaint {
                 if let Some(frame) = shell.compose(state.reported_size.0, state.reported_size.1) {
@@ -233,7 +288,7 @@ pub(super) fn begin_endpoint_activation(
         state.reported_cell_size.1,
         state.pixel_geometry_exact,
     );
-    match endpoint::PendingEndpointActivation::begin(
+    match endpoint::PendingEndpointActivation::prepare(
         shell,
         endpoints,
         endpoint_id.clone(),
@@ -241,7 +296,17 @@ pub(super) fn begin_endpoint_activation(
         resize,
         *next_surface_serial,
         now,
-    ) {
+    )
+    .and_then(|activation| {
+        // Preserve the old transaction if Local fails preflight. After retiring it,
+        // all send failures belong to the prepared replacement's rollback path.
+        if replace_pending {
+            if let Some(previous) = pending.take() {
+                previous.abandon(endpoints);
+            }
+        }
+        activation.start(endpoints)
+    }) {
         Ok(activation) => install_pending_activation(
             state,
             endpoint_commands,
@@ -388,7 +453,8 @@ pub(super) fn complete_endpoint_activation(
     if let Some(frame) = frame {
         state.present_frame(frame);
     }
-    if let Some(intent) = successor {
+    // A Local selection made while reconnecting is newer than this transaction's successor.
+    if let Some(intent) = successor.filter(|_| state.deferred_local_activation.is_none()) {
         return Ok(Some(ClientLoopEvent::ActivateEndpoint {
             endpoint_id: intent.endpoint_id,
             target: intent.target,
@@ -619,7 +685,7 @@ pub(super) fn finish_client_shell_input(
     pending_activation: &mut Option<endpoint::PendingEndpointActivation>,
     endpoint_commands: &mut endpoint_commands::EndpointCommands,
     prefix_input_source: &mut impl crate::platform::PrefixInputSource,
-    event_tx: &tokio::sync::mpsc::Sender<ClientLoopEvent>,
+    scheduled_activation: &mut Option<ClientLoopEvent>,
 ) -> Result<bool, ClientError> {
     apply_client_shell_input_source_changes(state, prefix_input_source);
     if outcome.detach {
@@ -658,7 +724,7 @@ pub(super) fn finish_client_shell_input(
         endpoints,
         state.shell.as_mut(),
         &mut state.detached_process_children,
-        event_tx,
+        scheduled_activation,
     )?;
     let frame = if dispatch_repaint {
         state
diff --git a/src/client/state.rs b/src/client/state.rs
index c26ca165..83b07429 100644
--- a/src/client/state.rs
+++ b/src/client/state.rs
@@ -38,6 +38,8 @@ pub(super) struct ClientState {
     /// During a source-off-first handoff the currently blitted frame remains authoritative until
     /// an acknowledged target snapshot/surface pair commits.
     pub(super) presentation_frozen: bool,
+    /// Latest explicit Local selection awaiting this client's replacement Local connection.
+    pub(super) deferred_local_activation: Option<endpoint::EndpointActivationIntent>,
     pub(super) draw_host_cursor: bool,
     pub(super) detached_process_children: Vec<std::process::Child>,
     pub(super) shell: Option<shell::ClientShellState>,
@@ -57,6 +59,47 @@ impl Drop for ClientState {
 }
 
 impl ClientState {
+    #[cfg(test)]
+    pub(super) fn test_new() -> Self {
+        Self {
+            blit_encoder: render_ansi::BlitEncoder::new(),
+            mouse_capture_active: false,
+            endpoint_mouse_capture_requested: false,
+            endpoint_sgr_pixels_requested: false,
+            host_theme_updates: Vec::new(),
+            direct_mouse_capture_preference: false,
+            shell_mouse_capture_preference: false,
+            direct_keyboard_protocol: Default::default(),
+            pane_keyboard_report_all: false,
+            keyboard_report_all_active: false,
+            reported_size: (100, 30),
+            reported_cell_size: (0, 0),
+            sound_config: Default::default(),
+            kitty_graphics_enabled: false,
+            pixel_geometry_enabled: false,
+            pixel_geometry_exact: false,
+            #[cfg(unix)]
+            direct_graphics_response: Default::default(),
+            #[cfg(unix)]
+            retired_direct_graphics: None,
+            #[cfg(unix)]
+            pending_surface_graphics: HashMap::new(),
+            attach_escape: None,
+            #[cfg(unix)]
+            mouse_scroll_lines: 3,
+            remote_image_paste_key: None,
+            redraw_on_focus_gained: false,
+            repaint_pending: false,
+            presentation_frozen: false,
+            deferred_local_activation: None,
+            draw_host_cursor: false,
+            detached_process_children: Vec::new(),
+            shell: Some(shell::ClientShellState::new(
+                shell::ClientShellConfig::from_config(&crate::config::Config::default()),
+            )),
+        }
+    }
+
     pub(super) fn request_repaint(&mut self) {
         self.repaint_pending = true;
     }