diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2485410e83..096743a24c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -240,7 +240,7 @@ jobs: -ExePath $exe ` -Session "ci-conpty-invalid-windows-2022-$env:GITHUB_RUN_ID-$env:GITHUB_RUN_ATTEMPT" } catch { - if ($_.Exception.Message -notlike "workspace create failed with exit code*") { + if ($_.Exception.Message -notlike "*Herdr's app-local ConPTY bundle is invalid:*") { throw } $rejected = $true diff --git a/Cargo.toml b/Cargo.toml index 64a665e858..4138a87624 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -74,6 +74,7 @@ windows-sys = { version = "0.61.2", features = [ "Win32_System_Memory", "Win32_System_Ole", "Win32_System_Pipes", + "Win32_System_RemoteDesktop", "Win32_System_Threading", "Win32_UI_Input_Ime", "Win32_UI_Input_KeyboardAndMouse", diff --git a/docs/next/website/src/content/docs/session-state.mdx b/docs/next/website/src/content/docs/session-state.mdx index 3871268904..e6ea32d43c 100644 --- a/docs/next/website/src/content/docs/session-state.mdx +++ b/docs/next/website/src/content/docs/session-state.mdx @@ -101,6 +101,8 @@ Unlike snapshot restore, pane history replay, and native agent session restore, A successful handoff preserves long-lived server-owned session state: pane PTYs and processes, agent identity and durable metadata, and plugin/session state needed by the replacement server. It does not preserve transient coordination across the replacement boundary. In-flight CLI or API requests, waits, subscription streams, client sockets, and pane-to-pane messages may be interrupted; clients should reconnect and retry them. +On Windows, live handoff requires the bundled ConPTY runtime. It is unavailable when Herdr uses the system runtime, including servers started with `HERDR_WINDOWS_CONPTY=system`. + Live handoff is experimental and opt-in: ```bash diff --git a/docs/next/website/src/content/docs/windows-beta.mdx b/docs/next/website/src/content/docs/windows-beta.mdx index 888080c810..fcbd694897 100644 --- a/docs/next/website/src/content/docs/windows-beta.mdx +++ b/docs/next/website/src/content/docs/windows-beta.mdx @@ -30,6 +30,7 @@ For internal testing, `HERDR_MANIFEST_URL` can point the installer at a custom m | Capability | Status | | --- | --- | | Local persistent sessions | supported | +| Live server handoff | experimental; requires the bundled ConPTY runtime | | Native panes through ConPTY | supported | | Windows Terminal / PowerShell app attach | supported | | `herdr --remote` and saved SSH machines to Linux/macOS/Windows hosts | supported; interactive attach and saved-machine setup can install or update Windows packages after confirmation; background reconnect only discovers installed packages | @@ -118,7 +119,6 @@ For text paste, use `ctrl+shift+v` in Windows Terminal. Multiline text paste is | Capability | Status | | --- | --- | | Direct terminal attach (`herdr terminal attach`) | unsupported | -| Live server handoff | unsupported | | Unix file-descriptor handoff | unsupported | | Unix foreground process groups | unsupported | | Herdr clipboard image bridge in local native panes | unsupported | @@ -132,7 +132,7 @@ herdr --remote workbox The target host can run Linux, macOS, or Windows. On Windows hosts, remote attach reuses a compatible package from `PATH` or the active managed release. Interactive direct attach and saved-machine setup prompt before installing or updating the complete package when needed. Background saved reconnect only discovers installed packages and cannot prompt for installation or updates. Herdr uses the installed Windows OpenSSH client and your SSH configuration. Windows OpenSSH does not use Herdr's Unix control-socket reuse, so key authentication through Windows `ssh-agent` is recommended to avoid repeated prompts during remote setup. -Windows updates run through the Windows installer and update the active versioned release path. New terminals and reconnected SSH sessions receive that path; start Herdr there to use the updated client. Compatible running servers keep their panes alive. Restart a server later only when you need server-side changes from the release. Live handoff is Unix-only. +Windows updates run through the Windows installer and update the active versioned release path. New terminals and reconnected SSH sessions receive that path; start Herdr there to use the updated client. Compatible running servers keep their panes alive. Restart a server later only when you need server-side changes from the release, or use experimental live handoff to replace it while preserving pane processes. Live handoff requires the bundled ConPTY runtime; servers started with `HERDR_WINDOWS_CONPTY=system` cannot use it. ## Reporting Windows issues diff --git a/scripts/windows_smoke_conpty_path.ps1 b/scripts/windows_smoke_conpty_path.ps1 index 484f2a1f41..15cb54e0fb 100644 --- a/scripts/windows_smoke_conpty_path.ps1 +++ b/scripts/windows_smoke_conpty_path.ps1 @@ -73,10 +73,15 @@ try { throw "command failed with exit code $LASTEXITCODE`: $exe --default-config" } - $server = Start-Process -FilePath $exe -ArgumentList "server" -PassThru -WindowStyle Hidden + $serverErrorPath = Join-Path $fakeDir "server-stderr.log" + $server = Start-Process -FilePath $exe -ArgumentList "server" -PassThru -WindowStyle Hidden -RedirectStandardError $serverErrorPath $deadline = (Get-Date).AddSeconds(10) do { Start-Sleep -Milliseconds 250 + if ($server.HasExited -and $server.ExitCode -ne 0) { + $serverError = Get-Content -LiteralPath $serverErrorPath -Raw + throw "server exited with exit code $($server.ExitCode) before becoming ready: $serverError" + } $status = & $exe status server 2>&1 if ($LASTEXITCODE -eq 0 -and (($status -join "`n") -match "status: running")) { break @@ -96,7 +101,8 @@ try { $ErrorActionPreference = $savedErrorActionPreference } if ($createdExitCode -ne 0) { - throw "workspace create failed with exit code $createdExitCode`: $($created -join "`n")" + $serverError = Get-Content -LiteralPath $serverErrorPath -Raw + throw "workspace create failed with exit code $createdExitCode`: $($created -join "`n")`n$serverError" } $paneId = (($created -join "`n") | ConvertFrom-Json).result.root_pane.pane_id if ([string]::IsNullOrWhiteSpace($paneId)) { diff --git a/src/api/mod.rs b/src/api/mod.rs index d28a82aef7..24c67f2f70 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -7,6 +7,8 @@ mod subscriptions; mod wait; pub use event_hub::EventHub; +#[cfg(windows)] +pub(crate) use server::start_server_from_handoff; pub use server::ServerHandle; pub(crate) use server::{api_method_name, start_server_with_stop_control}; pub use status::{read_runtime_status_at, RuntimeStatus}; diff --git a/src/api/server.rs b/src/api/server.rs index ea7597e7cb..88b7203645 100644 --- a/src/api/server.rs +++ b/src/api/server.rs @@ -4,8 +4,12 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; -use interprocess::local_socket::traits::{ListenerExt as _, Stream as _}; -use tracing::{debug, error, info, warn}; +#[cfg(unix)] +use interprocess::local_socket::traits::ListenerExt as _; +use interprocess::local_socket::traits::Stream as _; +#[cfg(unix)] +use tracing::error; +use tracing::{debug, info, warn}; #[cfg(all(test, unix))] use std::fs; @@ -34,8 +38,10 @@ const MAX_INITIAL_REQUEST_BYTES: usize = 1024 * 1024; pub struct ServerHandle { _thread: std::thread::JoinHandle<()>, path: PathBuf, - identity: SocketFileIdentity, + identity: Option, running: Arc, + #[cfg(windows)] + listener_control: crate::platform::WindowsListenerControl, } impl Drop for ServerHandle { @@ -52,7 +58,33 @@ impl Drop for ServerHandle { impl ServerHandle { pub(crate) fn remove_socket_file_if_owned(&self) -> std::io::Result<()> { - remove_socket_file_if_owned(&self.path, &self.identity) + self.identity.as_ref().map_or(Ok(()), |identity| { + remove_socket_file_if_owned(&self.path, identity) + }) + } + + #[cfg(windows)] + pub(crate) fn pause_listener_for_handoff(&self) -> std::io::Result<()> { + self.listener_control.pause() + } + + #[cfg(windows)] + pub(crate) fn duplicate_listener_for_handoff( + &self, + target: &std::process::Child, + ) -> std::io::Result { + self.listener_control.duplicate_for_handoff(target) + } + + #[cfg(windows)] + pub(crate) fn resume_listener_after_handoff(&self) -> std::io::Result<()> { + self.listener_control.resume() + } + + #[cfg(windows)] + pub(crate) fn release_listener_after_handoff(&mut self) -> std::io::Result<()> { + self.identity = None; + self.listener_control.release_after_commit() } } @@ -65,8 +97,11 @@ pub(crate) fn start_server_with_stop_control( } fn default_capabilities() -> Option { + let live_handoff = crate::platform::capabilities().live_handoff; + #[cfg(windows)] + let live_handoff = live_handoff && crate::pty::backend::windows_handoff_available(); Some(ServerCapabilities { - live_handoff: crate::platform::capabilities().live_handoff, + live_handoff, detached_server_daemon: crate::platform::current_process_is_detached_server_daemon(), endpoint_protocol_generation: Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION), surface_interest: true, @@ -88,45 +123,129 @@ fn start_server_inner( let identity = socket_file_identity(&path)?; info!(path = %path.display(), "api server listening"); - let running = Arc::new(AtomicBool::new(true)); - let listener_running = Arc::clone(&running); - let thread = std::thread::spawn(move || { - for stream in listener.incoming() { - match stream { - Ok(stream) => { - let api_tx = api_tx.clone(); - let event_hub = event_hub.clone(); - let capabilities = capabilities.clone(); - let server_stop = server_stop.clone(); - let connection_running = Arc::clone(&listener_running); - std::thread::spawn(move || { - if let Err(err) = handle_connection_with_stop( - stream, - &api_tx, - &event_hub, - &connection_running, - capabilities, - server_stop.as_ref(), - ) { - warn!(err = %err, "api connection failed"); - } - }); - } - Err(err) => { - error!(err = %err, "api listener accept failed"); - break; + #[cfg(windows)] + { + let listener = crate::platform::TransferableLocalListener::bound(listener)?; + Ok(start_windows_server_on_listener( + listener, + path, + identity, + api_tx, + event_hub, + capabilities, + server_stop, + )) + } + + #[cfg(unix)] + { + let running = Arc::new(AtomicBool::new(true)); + let listener_running = Arc::clone(&running); + let thread = std::thread::spawn(move || { + for stream in listener.incoming() { + match stream { + Ok(stream) => { + let api_tx = api_tx.clone(); + let event_hub = event_hub.clone(); + let capabilities = capabilities.clone(); + let server_stop = server_stop.clone(); + let connection_running = Arc::clone(&listener_running); + std::thread::spawn(move || { + if let Err(err) = handle_connection_with_stop( + stream, + &api_tx, + &event_hub, + &connection_running, + capabilities, + server_stop.as_ref(), + ) { + warn!(err = %err, "api connection failed"); + } + }); + } + Err(err) => { + error!(err = %err, "api listener accept failed"); + break; + } } } - } - debug!("api server thread exiting"); - }); + debug!("api server thread exiting"); + }); - Ok(ServerHandle { - _thread: thread, + Ok(ServerHandle { + _thread: thread, + path, + identity: Some(identity), + running, + }) + } +} + +#[cfg(windows)] +pub(crate) fn start_server_from_handoff( + listener: crate::platform::TransferableLocalListener, + identity: SocketFileIdentity, + api_tx: ApiRequestSender, + event_hub: EventHub, + server_stop: Arc, +) -> ServerHandle { + let path = socket_path(); + info!(path = %path.display(), "api server listener adopted after handoff"); + start_windows_server_on_listener( + listener, path, identity, + api_tx, + event_hub, + default_capabilities(), + Some(server_stop), + ) +} + +#[cfg(windows)] +fn start_windows_server_on_listener( + listener: crate::platform::TransferableLocalListener, + path: PathBuf, + identity: SocketFileIdentity, + api_tx: ApiRequestSender, + event_hub: EventHub, + capabilities: Option, + server_stop: Option>, +) -> ServerHandle { + let running = Arc::new(AtomicBool::new(true)); + let listener_running = Arc::clone(&running); + let should_stop = Arc::clone(&running); + let (thread, listener_control) = crate::platform::spawn_transferable_listener( + listener, + "api listener", + move || !should_stop.load(Ordering::Acquire), + move |stream| { + let api_tx = api_tx.clone(); + let event_hub = event_hub.clone(); + let capabilities = capabilities.clone(); + let server_stop = server_stop.clone(); + let connection_running = Arc::clone(&listener_running); + std::thread::spawn(move || { + if let Err(err) = handle_connection_with_stop( + stream, + &api_tx, + &event_hub, + &connection_running, + capabilities, + server_stop.as_ref(), + ) { + warn!(err = %err, "api connection failed"); + } + }); + }, + ); + ServerHandle { + _thread: thread, + path, + identity: Some(identity), running, - }) + listener_control, + } } fn prepare_socket_path(path: &Path) -> std::io::Result<()> { diff --git a/src/app/mod.rs b/src/app/mod.rs index 757a7f6636..40413a3bb1 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -30,7 +30,7 @@ mod window_title; mod worktrees; use std::collections::HashMap; -#[cfg(unix)] +#[cfg(any(unix, windows))] use std::io; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -92,7 +92,7 @@ impl AppPolicy { background_updates: false, }; - #[cfg(unix)] + #[cfg(any(unix, windows))] pub(crate) const HANDOFF_REPLACEMENT: Self = Self { restore_session: false, persist_session: true, @@ -630,7 +630,7 @@ impl App { app } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn new_from_handoff( config: &Config, config_diagnostic: Option, @@ -690,11 +690,16 @@ impl App { self.terminal_runtimes.set_handoff_readers_paused(false); } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn assume_handoff_ownership(&mut self) { self.terminal_runtimes.assume_handoff_ownership(); } + #[cfg(windows)] + pub fn activate_handoff_runtimes(&self) { + self.terminal_runtimes.activate_after_handoff() + } + pub(crate) fn ensure_default_workspace(&mut self) -> bool { if !self.state.workspaces.is_empty() { return false; diff --git a/src/handoff_runtime.rs b/src/handoff_runtime.rs index 29ba85f4e1..8b68e0b335 100644 --- a/src/handoff_runtime.rs +++ b/src/handoff_runtime.rs @@ -1,4 +1,4 @@ -#[cfg(unix)] +#[cfg(any(unix, windows))] use serde::{Deserialize, Serialize}; /// Long-lived pane runtime transferred during server replacement. @@ -8,7 +8,7 @@ use serde::{Deserialize, Serialize}; /// preserve transient coordination such as in-flight requests, waits, /// subscriptions, client sockets, or pane-to-pane messages; clients reconnect /// and retry those operations after replacement. -#[cfg(unix)] +#[cfg(any(unix, windows))] #[derive(Debug, Clone, Serialize, Deserialize)] pub(crate) struct HandoffRuntimeState { pub pane_id: u32, @@ -29,7 +29,7 @@ pub(crate) struct HandoffRuntimeState { pub initial_history_ansi: Option, } -#[cfg(unix)] +#[cfg(any(unix, windows))] impl HandoffRuntimeState { pub fn with_pane_id(mut self, pane_id: crate::layout::PaneId) -> Self { self.pane_id = pane_id.raw(); @@ -41,6 +41,8 @@ impl HandoffRuntimeState { pub(crate) struct ImportedHandoffRuntime { #[cfg(unix)] pub master_fd: std::os::fd::RawFd, - #[cfg(unix)] + #[cfg(windows)] + pub windows_pty: crate::pty::backend::WindowsPtyHandoff, + #[cfg(any(unix, windows))] pub state: HandoffRuntimeState, } diff --git a/src/input/mod.rs b/src/input/mod.rs index 13b8ffa3f1..f1529a8f3a 100644 --- a/src/input/mod.rs +++ b/src/input/mod.rs @@ -21,7 +21,7 @@ pub(crate) use keybindings::{ pub(crate) use lease::{InputLeaseKey, InputLeaseTable, RepeatPlan}; #[cfg(not(windows))] pub use model::ime_compatible_keyboard_enhancement_flags; -#[cfg(any(unix, test))] +#[cfg(any(unix, windows, test))] pub use model::MouseProtocolMode; pub use model::WindowsKeyRecord; pub use model::{ diff --git a/src/input/model.rs b/src/input/model.rs index 3828174ffc..fa9ded1dff 100644 --- a/src/input/model.rs +++ b/src/input/model.rs @@ -339,7 +339,7 @@ impl KeyboardProtocol { } } -#[cfg(any(unix, test))] +#[cfg(any(unix, windows, test))] #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum MouseProtocolMode { @@ -350,7 +350,7 @@ pub enum MouseProtocolMode { AnyMotion, } -#[cfg(any(unix, test))] +#[cfg(any(unix, windows, test))] impl MouseProtocolMode { #[cfg(test)] pub fn reporting_enabled(self) -> bool { diff --git a/src/ipc.rs b/src/ipc.rs index 291b6ca24c..0402fe7e0d 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -29,7 +29,7 @@ pub(crate) struct SocketFileIdentity { #[cfg(unix)] ino: u64, #[cfg(windows)] - marker: Vec, + pub(crate) marker: Vec, } pub(crate) fn connect_local_stream(path: &Path) -> io::Result { @@ -73,7 +73,7 @@ pub(crate) fn bind_local_listener(path: &Path) -> io::Result { .name(name) .reclaim_name(false) .create_sync()?; - fs::write(path, windows_socket_marker())?; + fs::write(path, crate::platform::windows_socket_marker())?; Ok(listener) } } @@ -156,7 +156,7 @@ pub(crate) fn bind_private_local_listener(path: &Path) -> io::Result String { - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|duration| duration.as_nanos()) - .unwrap_or(0); - format!("{}:{now}", std::process::id()) -} - #[cfg(unix)] pub(crate) fn restrict_socket_permissions(path: &Path, mode: u32) -> io::Result<()> { let mut permissions = fs::metadata(path)?.permissions(); diff --git a/src/pane.rs b/src/pane.rs index 2178605134..d15cecc9d4 100644 --- a/src/pane.rs +++ b/src/pane.rs @@ -7,9 +7,11 @@ use std::sync::{ }; use bytes::Bytes; -use portable_pty::CommandBuilder; #[cfg(all(test, unix))] -use portable_pty::{native_pty_system, PtySize}; +use portable_pty::native_pty_system; +use portable_pty::CommandBuilder; +#[cfg(any(windows, all(test, unix)))] +use portable_pty::PtySize; use ratatui::{layout::Rect, Frame}; #[cfg(test)] use tokio::sync::watch; @@ -40,7 +42,7 @@ use self::agent_detection::{ DetectionScreenReadInput, PendingIdleConfirmation, ScreenDetectionPublishInput, AGENT_PENDING_IDLE_RECHECK, AGENT_STARTUP_GRACE_WINDOW, }; -#[cfg(unix)] +#[cfg(any(unix, windows))] pub use self::terminal::InputState; use self::terminal::{GhosttyPaneTerminal, PaneTerminal}; pub(crate) use self::terminal::{ @@ -706,7 +708,7 @@ fn probe_foreground_process(pid: u32, foreground_pgid: Option) -> ProcessPr ) } -#[cfg(unix)] +#[cfg(any(unix, windows))] fn spawn_basic_detection_task( pane_id: PaneId, child_pid: Arc, @@ -1256,6 +1258,8 @@ pub struct PaneRuntime { child_pid: Arc, reported_cwd: Arc>>, child_wait_completed: Option>, + #[cfg(windows)] + handoff_wait_start: Option>, kitty_keyboard_flags: Arc, content_seq: Arc, content_write_lock: Arc>, @@ -1299,10 +1303,6 @@ impl PaneRuntimeIo { } #[cfg(windows)] - #[cfg_attr( - windows, - allow(dead_code, reason = "used by the stacked Windows handoff integration") - )] fn windows_handoff_supported(&self) -> bool { match self { PaneRuntimeIo::Actor(actor) => actor.supports_handoff(), @@ -1312,10 +1312,6 @@ impl PaneRuntimeIo { } #[cfg(windows)] - #[cfg_attr( - windows, - allow(dead_code, reason = "used by the stacked Windows handoff integration") - )] fn duplicate_windows_handoff( &self, target: &std::process::Child, @@ -1339,10 +1335,6 @@ impl PaneRuntimeIo { } #[cfg(any(unix, windows))] - #[cfg_attr( - windows, - allow(dead_code, reason = "used by the stacked Windows handoff integration") - )] fn begin_handoff(&self, timeout: std::time::Duration) -> std::io::Result<()> { match self { PaneRuntimeIo::Actor(actor) => actor.begin_handoff(timeout), @@ -1352,10 +1344,6 @@ impl PaneRuntimeIo { } #[cfg(any(unix, windows))] - #[cfg_attr( - windows, - allow(dead_code, reason = "used by the stacked Windows handoff integration") - )] fn set_handoff_paused(&self, paused: bool) -> std::io::Result<()> { match self { PaneRuntimeIo::Actor(actor) => { @@ -1371,10 +1359,6 @@ impl PaneRuntimeIo { } #[cfg(any(unix, windows))] - #[cfg_attr( - windows, - allow(dead_code, reason = "used by the stacked Windows handoff integration") - )] fn release_after_commit(&self) -> std::io::Result<()> { match self { PaneRuntimeIo::Actor(actor) => actor.release_after_commit(), @@ -1383,6 +1367,15 @@ impl PaneRuntimeIo { } } + #[cfg(windows)] + fn activate_after_handoff(&self) -> std::io::Result<()> { + match self { + PaneRuntimeIo::Actor(actor) => actor.activate_after_handoff(), + #[cfg(test)] + PaneRuntimeIo::TestChannel { .. } => Ok(()), + } + } + fn resize( &self, rows: u16, @@ -1598,7 +1591,7 @@ fn shutdown_pane_processes( ); } -#[cfg(unix)] +#[cfg(any(unix, windows))] fn truncate_handoff_history(history: String, max_bytes: usize) -> String { if history.len() <= max_bytes { return history; @@ -1915,19 +1908,11 @@ impl PaneRuntime { } #[cfg(windows)] - #[cfg_attr( - windows, - allow(dead_code, reason = "used by the stacked Windows handoff integration") - )] pub(crate) fn windows_handoff_supported(&self) -> bool { self.io.windows_handoff_supported() } #[cfg(windows)] - #[cfg_attr( - windows, - allow(dead_code, reason = "used by the stacked Windows handoff integration") - )] pub(crate) fn duplicate_windows_handoff( &self, target: &std::process::Child, @@ -1936,10 +1921,6 @@ impl PaneRuntime { } #[cfg(any(unix, windows))] - #[cfg_attr( - windows, - allow(dead_code, reason = "used by the stacked Windows handoff integration") - )] pub fn preserve_for_handoff(mut self) { if let Err(err) = self.io.release_after_commit() { warn!( @@ -1955,16 +1936,28 @@ impl PaneRuntime { self.preserve_processes_on_drop = true; } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn assume_handoff_ownership(&mut self) { self.preserve_processes_on_drop = false; + #[cfg(windows)] + if let Some(start) = self.handoff_wait_start.take() { + let _ = start.send(()); + } + } + + #[cfg(windows)] + pub(crate) fn activate_after_handoff(&self) -> std::io::Result<()> { + self.io.activate_after_handoff()?; + if std::env::var("HERDR_TEST_HANDOFF_IMPORT_FAIL").as_deref() == Ok("activation_ack") { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "test handoff activation acknowledgement timeout", + )); + } + Ok(()) } #[cfg(any(unix, windows))] - #[cfg_attr( - windows, - allow(dead_code, reason = "used by the stacked Windows handoff integration") - )] pub fn set_handoff_reader_paused(&self, paused: bool) { if let Err(err) = self.io.set_handoff_paused(paused) { warn!( @@ -1977,15 +1970,11 @@ impl PaneRuntime { } #[cfg(any(unix, windows))] - #[cfg_attr( - windows, - allow(dead_code, reason = "used by the stacked Windows handoff integration") - )] pub fn pause_handoff_reader(&self, timeout: std::time::Duration) -> std::io::Result<()> { self.io.begin_handoff(timeout) } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn handoff_runtime_state( &self, pane_id: u32, @@ -2010,7 +1999,7 @@ impl PaneRuntime { } } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn handoff_history_ansi(&self) -> Option { if self.terminal.alternate_screen_active() { return None; @@ -2194,8 +2183,8 @@ impl PaneRuntime { ) } - #[cfg(unix)] - pub fn from_handoff_fd( + #[cfg(any(unix, windows))] + pub fn from_handoff( import: crate::handoff_runtime::ImportedHandoffRuntime, scrollback_limit_bytes: usize, host_terminal_theme: crate::terminal_theme::TerminalTheme, @@ -2204,7 +2193,11 @@ impl PaneRuntime { render_notify: Arc, render_dirty: Arc, ) -> std::io::Result { - let crate::handoff_runtime::ImportedHandoffRuntime { master_fd, state } = import; + #[cfg(unix)] + let master_fd = import.master_fd; + #[cfg(windows)] + let windows_pty = import.windows_pty; + let state = import.state; let crate::handoff_runtime::HandoffRuntimeState { pane_id, child_pid, @@ -2219,10 +2212,32 @@ impl PaneRuntime { initial_history_ansi, } = state; let pane_id = PaneId::from_raw(pane_id); + #[cfg(unix)] use std::os::fd::FromRawFd; + #[cfg(unix)] let master_fd = unsafe { std::os::fd::OwnedFd::from_raw_fd(master_fd) }; + #[cfg(windows)] + let spawned = unsafe { + crate::pty::backend::adopt_windows_handoff( + windows_pty, + PtySize { + rows, + cols, + pixel_width: cell_width_px.min(u16::MAX as u32) as u16, + pixel_height: cell_height_px.min(u16::MAX as u32) as u16, + }, + ) + }?; + #[cfg(windows)] + if spawned.child.process_id() != Some(child_pid) { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "transferred pane process identity does not match its manifest", + )); + } + let (response_tx, _response_rx) = mpsc::channel::(1); let mut terminal = crate::ghostty::Terminal::new(cols, rows, scrollback_limit_bytes) .map_err(|e| std::io::Error::other(e.to_string()))?; @@ -2258,6 +2273,46 @@ impl PaneRuntime { let content_write_lock = Arc::new(Mutex::new(())); let detection_content_seq = Arc::new(AtomicU64::new(0)); + #[cfg(unix)] + let child_wait_completed = None; + #[cfg(windows)] + let (handoff_wait_start, wait_start) = tokio::sync::oneshot::channel(); + #[cfg(windows)] + let child_wait_completed = { + let completed = Arc::new(AtomicBool::new(false)); + let completed_for_wait = Arc::clone(&completed); + let wait_events = events.clone(); + let rt = tokio::runtime::Handle::current(); + let mut child = spawned.child; + // Before COMMIT, dropping the runtime cancels this task and closes + // only the duplicate child handle; no exit event is published. + tokio::spawn(async move { + if wait_start.await.is_err() { + return; + } + tokio::task::spawn_blocking(move || { + let exit_reason = match child.wait() { + Ok(status) => { + let exit_reason = crate::platform::classify_child_exit(&status); + let status_text = format!("{status:?}"); + crate::logging::pane_exited(pane_id.raw(), &status_text); + exit_reason + } + Err(err) => { + crate::logging::pane_exit_failed(pane_id.raw(), &err.to_string()); + crate::platform::ChildExitReason::WaitFailed + } + }; + completed_for_wait.store(true, Ordering::Release); + let _ = rt.block_on(wait_events.send(AppEvent::PaneDied { + pane_id, + exit_reason, + })); + }); + }); + Some(completed) + }; + let io = { let terminal = terminal.clone(); let response_writer = response_tx.clone(); @@ -2318,22 +2373,31 @@ impl PaneRuntime { terminal_responses: result.terminal_responses, } }); - let exit_events = events.clone(); - let on_reader_exit = Box::new(move || { - // Imported handoff panes have no child wait handle, so their exit cause is - // unknowable. Checkpoint conservatively; normal autosave settles clean exits. - let _ = rt.block_on(exit_events.send(AppEvent::PaneDied { - pane_id, - exit_reason: crate::platform::ChildExitReason::Handoff, - })); - debug!(pane = pane_id.raw(), "handoff PTY actor exiting"); - }); + #[cfg(unix)] + let on_reader_exit = { + let exit_events = events.clone(); + Some(Box::new(move || { + // Unix imports have no child wait handle, so their exit cause is unknown. + let _ = rt.block_on(exit_events.send(AppEvent::PaneDied { + pane_id, + exit_reason: crate::platform::ChildExitReason::Handoff, + })); + debug!(pane = pane_id.raw(), "handoff PTY actor exiting"); + }) as Box) + }; + #[cfg(windows)] + let on_reader_exit = None; PaneRuntimeIo::Actor(PtyIoActor::spawn(PtyIoActorConfig { pane_id: pane_id.raw(), + #[cfg(unix)] master_fd, + #[cfg(windows)] + master: spawned.master, + #[cfg(windows)] + handoff_child: spawned.handoff_child, initially_quiesced: true, on_read, - on_reader_exit: Some(on_reader_exit), + on_reader_exit, })?) }; @@ -2354,7 +2418,9 @@ impl PaneRuntime { current_size: Cell::new((rows, cols, cell_width_px, cell_height_px)), child_pid, reported_cwd, - child_wait_completed: None, + child_wait_completed, + #[cfg(windows)] + handoff_wait_start: Some(handoff_wait_start), kitty_keyboard_flags, content_seq, content_write_lock, @@ -2933,6 +2999,8 @@ impl PaneRuntime { child_pid, reported_cwd, child_wait_completed: Some(child_wait_completed), + #[cfg(windows)] + handoff_wait_start: None, kitty_keyboard_flags, content_seq, content_write_lock, @@ -3097,7 +3165,7 @@ impl PaneRuntime { result } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn input_state(&self) -> Option { self.terminal.input_state() } @@ -3599,6 +3667,8 @@ impl PaneRuntime { child_pid: Arc::new(AtomicU32::new(0)), reported_cwd: Arc::new(Mutex::new(None)), child_wait_completed: None, + #[cfg(windows)] + handoff_wait_start: None, kitty_keyboard_flags: Arc::new(AtomicU16::new(0)), content_seq: Arc::new(AtomicU64::new(0)), content_write_lock: Arc::new(Mutex::new(())), @@ -4451,6 +4521,8 @@ mod tests { child_pid: Arc::new(AtomicU32::new(0)), reported_cwd: Arc::new(Mutex::new(None)), child_wait_completed: None, + #[cfg(windows)] + handoff_wait_start: None, kitty_keyboard_flags: Arc::new(AtomicU16::new(0)), content_seq: Arc::new(AtomicU64::new(0)), content_write_lock: Arc::new(Mutex::new(())), @@ -4488,6 +4560,8 @@ mod tests { child_pid: Arc::new(AtomicU32::new(0)), reported_cwd: Arc::new(Mutex::new(None)), child_wait_completed: None, + #[cfg(windows)] + handoff_wait_start: None, kitty_keyboard_flags: Arc::new(AtomicU16::new(0)), content_seq: Arc::new(AtomicU64::new(0)), content_write_lock: Arc::new(Mutex::new(())), diff --git a/src/pane/kitty_keyboard.rs b/src/pane/kitty_keyboard.rs index bc8876e6e5..2c7da81e88 100644 --- a/src/pane/kitty_keyboard.rs +++ b/src/pane/kitty_keyboard.rs @@ -127,7 +127,7 @@ impl KittyKeyboardTracker { } } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub(crate) fn replay_ansi(&self) -> Option { let mut ansi = String::new(); if self.stack.is_empty() { @@ -191,7 +191,7 @@ mod tests { assert_eq!(tracker.flags, 0); assert!(tracker.stack.is_empty()); assert_eq!(tracker.modify_other_keys_level(), 0); - #[cfg(unix)] + #[cfg(any(unix, windows))] assert_eq!(tracker.replay_ansi(), None); } @@ -201,7 +201,7 @@ mod tests { tracker.observe(b"\x1b[>4;1m"); assert_eq!(tracker.modify_other_keys_level(), 1); - #[cfg(unix)] + #[cfg(any(unix, windows))] assert_eq!(tracker.replay_ansi().as_deref(), Some("\x1b[>4;1m")); tracker.observe(b"\x1b[>4;2m"); diff --git a/src/pane/osc.rs b/src/pane/osc.rs index a281af77ae..723cb14b01 100644 --- a/src/pane/osc.rs +++ b/src/pane/osc.rs @@ -498,7 +498,7 @@ impl AgentOscStateTracker { self.terminal_title.as_deref() } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub(super) fn seed_terminal_title(&mut self, title: Option) { self.terminal_title = title; } diff --git a/src/pane/terminal.rs b/src/pane/terminal.rs index 4a6bc17321..909b157630 100644 --- a/src/pane/terminal.rs +++ b/src/pane/terminal.rs @@ -7,7 +7,7 @@ use std::time::{Duration, Instant}; use bytes::Bytes; use ratatui::style::{Color, Modifier, Style}; use ratatui::{layout::Rect, Frame}; -#[cfg(any(unix, test))] +#[cfg(any(unix, windows, test))] use serde::{Deserialize, Serialize}; use tokio::sync::mpsc; use tracing::{debug, error}; @@ -130,7 +130,7 @@ fn decscusr_cursor_shape(style: crate::ghostty::CursorVisualStyle, blinking: boo } } -#[cfg(any(unix, test))] +#[cfg(any(unix, windows, test))] #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub struct InputState { pub alternate_screen: bool, @@ -423,7 +423,7 @@ impl PaneTerminal { Some((RetainedTextBuffer::new_search(cols, rows, 0), active_screen)) } - #[cfg(any(unix, test))] + #[cfg(any(unix, windows, test))] pub fn input_state(&self) -> Option { self.ghostty.input_state() } @@ -614,7 +614,7 @@ impl PaneTerminal { self.ghostty.keyboard_protocol().unwrap_or(fallback) } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn kitty_keyboard_state_ansi(&self) -> Option { self.ghostty .kitty_keyboard_state_ansi() @@ -1294,7 +1294,7 @@ impl GhosttyPaneTerminal { .and_then(|core| core.agent_osc_state.terminal_title().map(str::to_string)) } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn seed_terminal_title(&self, title: Option) { if let Ok(mut core) = self.core.lock() { core.agent_osc_state.seed_terminal_title(title); @@ -1585,7 +1585,7 @@ impl GhosttyPaneTerminal { } } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn seed_handoff_input_state(&self, input_state: InputState) { let Ok(mut core) = self.core.lock() else { return; @@ -1649,7 +1649,7 @@ impl GhosttyPaneTerminal { } } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn seed_keyboard_protocol_flags(&self, flags: u16) { if flags == 0 { return; @@ -1657,7 +1657,7 @@ impl GhosttyPaneTerminal { self.seed_keyboard_protocol_ansi(&format!("\x1b[>{flags}u")); } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn seed_keyboard_protocol_ansi(&self, ansi: &str) { if ansi.is_empty() { return; @@ -1804,7 +1804,7 @@ impl GhosttyPaneTerminal { )) } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn kitty_keyboard_state_ansi(&self) -> Option { let core = self.core.lock().ok()?; core.kitty_keyboard.replay_ansi() @@ -1864,7 +1864,7 @@ impl GhosttyPaneTerminal { // This aggregate snapshot performs multiple terminal queries. Pane-scaled // callers should add a narrow accessor instead. - #[cfg(any(unix, test))] + #[cfg(any(unix, windows, test))] pub fn input_state(&self) -> Option { let Ok(core) = self.core.lock() else { return None; diff --git a/src/persist.rs b/src/persist.rs index c95355e60c..52e29943df 100644 --- a/src/persist.rs +++ b/src/persist.rs @@ -12,7 +12,7 @@ mod writer; pub use self::io::{clear_history, load, load_history}; pub use self::restore::restore; -#[cfg(unix)] +#[cfg(any(unix, windows))] pub use self::restore::{handoff_pane_aliases, restore_handoff}; pub use self::snapshot::{ capture, capture_history, DirectionSnapshot, LayoutSnapshot, SessionHistorySnapshot, diff --git a/src/persist/restore.rs b/src/persist/restore.rs index 3c5775c9e1..f6916e871c 100644 --- a/src/persist/restore.rs +++ b/src/persist/restore.rs @@ -91,7 +91,7 @@ pub fn restore( ) } -#[cfg(unix)] +#[cfg(any(unix, windows))] pub fn restore_handoff( snapshot: &SessionSnapshot, scrollback_limit_bytes: usize, @@ -117,7 +117,7 @@ pub fn restore_handoff( ) } -#[cfg(unix)] +#[cfg(any(unix, windows))] pub fn handoff_pane_aliases( snapshot: &SessionSnapshot, workspaces: &[Workspace], @@ -137,14 +137,14 @@ pub fn handoff_pane_aliases( aliases } -#[cfg(unix)] +#[cfg(any(unix, windows))] fn collect_snapshot_pane_ids(node: &LayoutSnapshot) -> Vec { let mut ids = Vec::new(); collect_snapshot_ids_inner(node, &mut ids); ids } -#[cfg(unix)] +#[cfg(any(unix, windows))] fn collect_snapshot_ids_inner(node: &LayoutSnapshot, ids: &mut Vec) { match node { LayoutSnapshot::Pane(id) => ids.push(*id), @@ -184,7 +184,7 @@ fn collect_layout_snapshot_pane_ids(node: &LayoutSnapshot, ids: &mut Vec) { } } -#[cfg(unix)] +#[cfg(any(unix, windows))] fn restore_with_imports_strict( snapshot: &SessionSnapshot, history: Option<&SessionHistorySnapshot>, @@ -566,20 +566,19 @@ fn restore_tab( continue; } - #[cfg(not(unix))] + #[cfg(not(any(unix, windows)))] if imported_runtime.is_some() { failed_imports += 1; continue; } let runtime_result = { - #[cfg(unix)] + #[cfg(any(unix, windows))] if let Some(imported) = imported_runtime { - TerminalRuntime::from_handoff_fd( - crate::handoff_runtime::ImportedHandoffRuntime { - master_fd: imported.master_fd, - state: imported.state.with_pane_id(*id), - }, + let mut imported = imported; + imported.state = imported.state.with_pane_id(*id); + TerminalRuntime::from_handoff( + imported, runtime_context.scrollback_limit_bytes, crate::terminal_theme::TerminalTheme::default(), None, @@ -605,7 +604,7 @@ fn restore_tab( ) } - #[cfg(not(unix))] + #[cfg(not(any(unix, windows)))] { TerminalRuntime::spawn_with_initial_history( *id, diff --git a/src/platform/mod.rs b/src/platform/mod.rs index eb84e0b1d7..05beb627b4 100644 --- a/src/platform/mod.rs +++ b/src/platform/mod.rs @@ -123,9 +123,9 @@ pub(crate) struct PlatformCapabilities { pub(crate) preserve_legacy_doubled_escape_input: bool, } -pub(crate) const fn capabilities() -> PlatformCapabilities { +pub(crate) fn capabilities() -> PlatformCapabilities { PlatformCapabilities { - live_handoff: cfg!(unix), + live_handoff: cfg!(any(unix, windows)), direct_terminal_attach: cfg!(unix), preserve_legacy_doubled_escape_input: cfg!(target_os = "macos"), } diff --git a/src/platform/windows.rs b/src/platform/windows.rs index 4a52f6e132..ac9221038e 100644 --- a/src/platform/windows.rs +++ b/src/platform/windows.rs @@ -414,6 +414,408 @@ impl SynchronousIoCancel { } } +const LISTENER_HANDOFF_TIMEOUT: Duration = Duration::from_secs(2); + +pub(crate) fn windows_socket_marker() -> String { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_nanos()) + .unwrap_or(0); + format!("{}:{now}", std::process::id()) +} + +pub(crate) struct WindowsHandoffMarker { + file: std::fs::File, + path: std::path::PathBuf, + identity: crate::ipc::SocketFileIdentity, +} + +impl WindowsHandoffMarker { + pub(crate) fn prepare(path: &std::path::Path) -> std::io::Result { + // Acquire write access before READY without changing the source marker. + let file = std::fs::OpenOptions::new().write(true).open(path)?; + Ok(Self { + file, + path: path.to_owned(), + identity: crate::ipc::SocketFileIdentity { + marker: windows_socket_marker().into_bytes(), + }, + }) + } + + pub(crate) fn publish(mut self) -> crate::ipc::SocketFileIdentity { + use std::io::Write as _; + let result = (|| { + if std::env::var("HERDR_TEST_HANDOFF_IMPORT_FAIL").as_deref() == Ok("marker_publish") { + return Err(std::io::Error::other( + "test handoff marker publication failure", + )); + } + self.file.write_all(&self.identity.marker)?; + self.file.set_len(self.identity.marker.len() as u64) + })(); + if let Err(err) = result { + // COMMIT has transferred ownership. Metadata failure cannot stop it. + tracing::warn!(path = %self.path.display(), %err, "failed to publish handoff socket marker; continuing as owner"); + } + self.identity + } +} + +/// Bounded IO only for the private replacement transaction. Public streams +/// remain blocking: PIPE_NOWAIT can return zero while their write buffer is full. +pub(crate) struct WindowsHandoffStream { + stream: crate::ipc::LocalStream, + deadline: std::cell::Cell, +} + +impl WindowsHandoffStream { + pub(crate) fn new(stream: crate::ipc::LocalStream, timeout: Duration) -> std::io::Result { + use interprocess::local_socket::traits::Stream as _; + stream.set_nonblocking(true)?; + Ok(Self { + stream, + deadline: std::cell::Cell::new(std::time::Instant::now() + timeout), + }) + } + + pub(crate) fn set_read_timeout(&self, timeout: Option) -> std::io::Result<()> { + self.deadline + .set(std::time::Instant::now() + timeout.unwrap_or(Duration::from_secs(30))); + Ok(()) + } + + fn check_deadline(&self) -> std::io::Result<()> { + if std::time::Instant::now() >= self.deadline.get() { + Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "handoff IO timed out", + )) + } else { + Ok(()) + } + } +} + +impl std::io::Read for WindowsHandoffStream { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + loop { + self.check_deadline()?; + match crate::ipc::poll_local_stream_read_count(&mut self.stream, buffer)? { + crate::ipc::LocalStreamReadCount::Data(count) => return Ok(count), + crate::ipc::LocalStreamReadCount::Closed => return Ok(0), + crate::ipc::LocalStreamReadCount::Pending => { + std::thread::sleep(Duration::from_millis(1)) + } + } + } + } +} + +impl std::io::Write for WindowsHandoffStream { + fn write(&mut self, buffer: &[u8]) -> std::io::Result { + if buffer.is_empty() { + return Ok(0); + } + loop { + self.check_deadline()?; + match std::io::Write::write(&mut self.stream, buffer) { + Ok(0) => {} + Err(err) + if matches!( + err.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::Interrupted + ) => {} + result => return result, + } + std::thread::sleep(Duration::from_millis(1)); + } + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +pub(crate) enum TransferableLocalListener { + Bound(crate::ipc::LocalListener), + Adopted( + interprocess::os::windows::named_pipe::PipeListener< + interprocess::os::windows::named_pipe::pipe_mode::Bytes, + interprocess::os::windows::named_pipe::pipe_mode::Bytes, + >, + ), +} + +impl TransferableLocalListener { + pub(crate) fn bound(listener: crate::ipc::LocalListener) -> std::io::Result { + use interprocess::local_socket::traits::Listener as _; + use interprocess::local_socket::ListenerNonblockingMode; + + listener.set_nonblocking(ListenerNonblockingMode::Accept)?; + Ok(Self::Bound(listener)) + } + + pub(crate) fn from_handoff_handle( + handle: OwnedHandle, + path: &std::path::Path, + ) -> std::io::Result { + use interprocess::os::windows::named_pipe::PipeListenerOptions; + + let path = format!(r"\\.\pipe\{}", path.to_string_lossy()); + let options = PipeListenerOptions::new() + .path(path) + .nonblocking(true) + .to_owned()?; + Ok(Self::Adopted( + interprocess::os::windows::named_pipe::PipeListener::from_handle_and_options( + handle, options, + ), + )) + } + + pub(crate) fn accept(&self) -> std::io::Result { + use interprocess::local_socket::traits::Listener as _; + + match self { + Self::Bound(listener) => listener.accept(), + Self::Adopted(listener) => { + let stream = listener.accept()?; + stream.set_nonblocking(false)?; + Ok(crate::ipc::LocalStream::NamedPipe(stream.into())) + } + } + } + + fn raw_handle(&self) -> usize { + use std::os::windows::io::AsRawHandle as _; + + match self { + Self::Bound(crate::ipc::LocalListener::NamedPipe(listener)) => { + listener.as_ref().as_raw_handle() as usize + } + Self::Adopted(listener) => listener.as_raw_handle() as usize, + } + } +} + +pub(crate) struct WindowsListenerHandoff { + handle: usize, + target_process: Option, +} + +impl WindowsListenerHandoff { + pub(crate) fn into_raw_handle(mut self) -> usize { + self.target_process.take(); + std::mem::take(&mut self.handle) + } +} + +impl Drop for WindowsListenerHandoff { + fn drop(&mut self) { + let Some(target_process) = self.target_process.as_ref() else { + return; + }; + if self.handle != 0 && self.handle != INVALID_HANDLE_VALUE as usize { + let _ = close_handle_in_process(self.handle, target_process.as_raw_handle() as usize); + } + } +} + +enum ListenerControlCommand { + Pause(std::sync::mpsc::Sender>), + Duplicate { + target_process: OwnedHandle, + reply: std::sync::mpsc::Sender>, + }, + Resume(std::sync::mpsc::Sender>), + Release(std::sync::mpsc::Sender>), +} + +#[derive(Clone)] +pub(crate) struct WindowsListenerControl { + tx: std::sync::mpsc::Sender, +} + +impl WindowsListenerControl { + pub(crate) fn pause(&self) -> std::io::Result<()> { + self.request(ListenerControlCommand::Pause) + } + + pub(crate) fn resume(&self) -> std::io::Result<()> { + self.request(ListenerControlCommand::Resume) + } + + pub(crate) fn release_after_commit(&self) -> std::io::Result<()> { + self.request(ListenerControlCommand::Release) + } + + pub(crate) fn duplicate_for_handoff( + &self, + target: &std::process::Child, + ) -> std::io::Result { + use std::os::windows::io::{AsRawHandle as _, BorrowedHandle}; + + let target_process = + unsafe { BorrowedHandle::borrow_raw(target.as_raw_handle()) }.try_clone_to_owned()?; + self.request(|reply| ListenerControlCommand::Duplicate { + target_process, + reply, + }) + } + + fn request( + &self, + command: impl FnOnce(std::sync::mpsc::Sender>) -> ListenerControlCommand, + ) -> std::io::Result { + let (reply_tx, reply_rx) = std::sync::mpsc::channel(); + let command = command(reply_tx); + let pause = matches!(command, ListenerControlCommand::Pause(_)); + self.tx + .send(command) + .map_err(|_| std::io::Error::new(std::io::ErrorKind::BrokenPipe, "listener stopped"))?; + match reply_rx.recv_timeout(LISTENER_HANDOFF_TIMEOUT) { + Ok(result) => result, + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + // A late pause is followed by this queued resume, so a timed-out + // request cannot leave the source listener frozen. + if pause { + let (resume_tx, _) = std::sync::mpsc::channel(); + let _ = self.tx.send(ListenerControlCommand::Resume(resume_tx)); + } + Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "listener handoff control timed out", + )) + } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "listener stopped during handoff control", + )), + } + } +} + +pub(crate) fn spawn_transferable_listener( + listener: TransferableLocalListener, + label: &'static str, + mut should_stop: S, + mut on_accept: F, +) -> (std::thread::JoinHandle<()>, WindowsListenerControl) +where + F: FnMut(crate::ipc::LocalStream) + Send + 'static, + S: FnMut() -> bool + Send + 'static, +{ + let (control_tx, control_rx) = std::sync::mpsc::channel(); + let thread = std::thread::spawn(move || { + let mut paused = false; + loop { + let command = if paused { + match control_rx.recv_timeout(Duration::from_millis(10)) { + Ok(command) => Some(command), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => None, + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break, + } + } else { + control_rx.try_recv().ok() + }; + if let Some(command) = command { + match command { + ListenerControlCommand::Pause(reply) => { + paused = true; + let _ = reply.send(Ok(())); + } + ListenerControlCommand::Duplicate { + target_process, + reply, + } => { + let result = if paused { + duplicate_handle_into_process( + listener.raw_handle(), + target_process.as_raw_handle() as usize, + ) + .map(|handle| WindowsListenerHandoff { + handle, + target_process: Some(target_process), + }) + } else { + Err(std::io::Error::other( + "listener must be paused before handoff duplication", + )) + }; + let _ = reply.send(result); + } + ListenerControlCommand::Resume(reply) => { + paused = false; + let _ = reply.send(Ok(())); + } + ListenerControlCommand::Release(reply) => { + let _ = reply.send(Ok(())); + break; + } + } + continue; + } + if paused { + continue; + } + if should_stop() { + break; + } + match listener.accept() { + Ok(stream) => on_accept(stream), + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => { + std::thread::sleep(Duration::from_millis(10)); + } + Err(err) if err.kind() == std::io::ErrorKind::Interrupted => {} + Err(err) => { + tracing::error!(err = %err, "{label} accept failed"); + std::thread::sleep(Duration::from_millis(50)); + } + } + } + tracing::debug!("{label} accept thread exiting"); + }); + (thread, WindowsListenerControl { tx: control_tx }) +} + +pub(crate) fn named_pipe_peer_pid(stream: &crate::ipc::LocalStream) -> std::io::Result { + use std::os::windows::io::{AsHandle as _, AsRawHandle as _}; + + let crate::ipc::LocalStream::NamedPipe(pipe) = stream; + let mut pid = 0; + let ok = unsafe { + windows_sys::Win32::System::Pipes::GetNamedPipeClientProcessId( + pipe.as_handle().as_raw_handle(), + &mut pid, + ) + }; + if ok == 0 { + Err(std::io::Error::last_os_error()) + } else { + Ok(pid) + } +} + +pub(crate) fn ensure_same_process_session(child_pid: u32) -> std::io::Result<()> { + use windows_sys::Win32::System::RemoteDesktop::ProcessIdToSessionId; + + let mut source_session = 0; + let mut child_session = 0; + if unsafe { ProcessIdToSessionId(std::process::id(), &mut source_session) } == 0 + || unsafe { ProcessIdToSessionId(child_pid, &mut child_session) } == 0 + { + return Err(std::io::Error::last_os_error()); + } + if source_session != child_session { + return Err(std::io::Error::new( + std::io::ErrorKind::PermissionDenied, + "handoff replacement started in a different Windows session", + )); + } + Ok(()) +} + pub(crate) fn duplicate_handle_into_process( source: usize, target_process: usize, @@ -2996,6 +3398,72 @@ mod tests { AllocConsole, FreeConsole, GetConsoleProcessList, GetConsoleWindow, }; + #[test] + fn windows_private_handoff_io_preserves_full_writes_and_bounds_stalled_peers() { + use std::io::{Read, Write}; + fn pair(label: &str) -> (crate::ipc::LocalStream, crate::ipc::LocalStream) { + let path = std::env::temp_dir() + .join(format!("herdr-handoff-io-{}-{label}", std::process::id())); + let listener = super::TransferableLocalListener::bound( + crate::ipc::bind_local_listener(&path).unwrap(), + ) + .unwrap(); + let connect_path = path.clone(); + let client = + thread::spawn(move || crate::ipc::connect_local_stream(&connect_path).unwrap()); + let deadline = Instant::now() + Duration::from_secs(3); + let server = loop { + match listener.accept() { + Ok(stream) => break stream, + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => { + assert!(Instant::now() < deadline); + thread::sleep(Duration::from_millis(1)); + } + Err(err) => panic!("{err}"), + } + }; + fs::remove_file(path).unwrap(); + (server, client.join().unwrap()) + } + let (server, mut client) = pair("full"); + let receiver = thread::spawn(move || { + thread::sleep(Duration::from_millis(100)); + let mut payload = vec![0; 1024 * 1024]; + client.read_exact(&mut payload).unwrap(); + assert!(payload + .iter() + .enumerate() + .all(|(index, byte)| *byte == (index % 251) as u8)); + let mut commit = [0; 10]; + client.read_exact(&mut commit).unwrap(); + assert_eq!(&commit, b"committed\n"); + }); + let mut stream = super::WindowsHandoffStream::new(server, Duration::from_secs(5)).unwrap(); + let payload: Vec<_> = (0..1024 * 1024).map(|index| (index % 251) as u8).collect(); + stream.write_all(&payload).unwrap(); + stream.write_all(b"committed\n").unwrap(); + receiver.join().unwrap(); + assert!(crate::ipc::is_connection_closed_error( + &stream.write_all(b"closed").unwrap_err() + )); + let (server, _stalled_client) = pair("stall"); + let start = Instant::now(); + let mut stream = + super::WindowsHandoffStream::new(server, Duration::from_millis(100)).unwrap(); + assert_eq!( + stream.write_all(&payload).unwrap_err().kind(), + std::io::ErrorKind::TimedOut + ); + assert!(start.elapsed() < Duration::from_secs(1)); + stream + .set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); + assert_eq!( + stream.read_exact(&mut [0]).unwrap_err().kind(), + std::io::ErrorKind::TimedOut + ); + } + #[test] fn windows_standard_plugin_runtime_paths_drop_only_disk_and_unc_verbatim_prefixes() { assert_eq!( diff --git a/src/server/handoff.rs b/src/server/handoff.rs index aaa87f382b..ec13369cca 100644 --- a/src/server/handoff.rs +++ b/src/server/handoff.rs @@ -1,38 +1,42 @@ -#[cfg(unix)] +#[cfg(any(unix, windows))] use std::io::{self, Read, Write}; #[cfg(unix)] use std::os::fd::{AsRawFd, RawFd}; #[cfg(unix)] use std::os::unix::net::{UnixListener, UnixStream}; #[cfg(unix)] +type HandoffStream = UnixStream; +#[cfg(windows)] +type HandoffStream = crate::platform::WindowsHandoffStream; +#[cfg(any(unix, windows))] use std::path::{Path, PathBuf}; -#[cfg(unix)] +#[cfg(any(unix, windows))] use std::process::{Child, Command}; -#[cfg(unix)] +#[cfg(any(unix, windows))] use std::time::Duration; -#[cfg(unix)] +#[cfg(any(unix, windows))] use serde::{Deserialize, Serialize}; -#[cfg(unix)] +#[cfg(any(unix, windows))] use tracing::{info, warn}; -#[cfg(unix)] +#[cfg(any(unix, windows))] const HANDOFF_VERSION: u32 = 1; -#[cfg(unix)] +#[cfg(any(unix, windows))] const READY_TIMEOUT: Duration = Duration::from_secs(30); -#[cfg(unix)] +#[cfg(any(unix, windows))] const OWNED_ACK_TIMEOUT: Duration = Duration::from_millis(500); // Descriptors are transferred in batches of this size. A single SCM_RIGHTS // control message caps out at 253 descriptors on Linux and 254 on macOS, so the // batch stays well below both limits and the number of panes stays unbounded. #[cfg(unix)] const FDS_PER_MESSAGE: usize = 64; -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) const MAX_REPLAY_BYTES_PER_PANE: usize = 8 * 1024; -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) const COMMIT_TIMEOUT: Duration = READY_TIMEOUT; -#[cfg(unix)] +#[cfg(any(unix, windows))] #[derive(Serialize, Deserialize)] pub(crate) struct HandoffManifest { pub version: u32, @@ -49,19 +53,24 @@ pub(crate) struct HandoffManifest { pub api_window_title: Option, } -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) struct ReceivedHandoff { pub manifest: HandoffManifest, + #[cfg(unix)] pub fds: Vec, - pub stream: UnixStream, + #[cfg(windows)] + pub ptys: Vec, + #[cfg(windows)] + pub listeners: [crate::platform::TransferableLocalListener; 2], + pub stream: HandoffStream, } -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) fn handoff_socket_path() -> PathBuf { crate::session::data_dir().join(format!("herdr-handoff-{}.sock", std::process::id())) } -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) fn spawn_handoff_import( import_exe: Option<&Path>, socket_path: &Path, @@ -107,33 +116,6 @@ pub(crate) fn spawn_handoff_import( }) } -#[cfg(unix)] -pub(crate) fn cleanup_failed_import_child(child: &mut Child) { - let pid = child.id(); - match child.try_wait() { - Ok(Some(status)) => { - info!(pid, status = %status, "handoff import server exited during rollback"); - return; - } - Ok(None) => {} - Err(err) => { - warn!(pid, err = %err, "failed to inspect handoff import server before rollback"); - } - } - - if let Err(err) = child.kill() { - warn!(pid, err = %err, "failed to kill handoff import server during rollback"); - } - match child.wait() { - Ok(status) => { - info!(pid, status = %status, "handoff import server reaped during rollback"); - } - Err(err) => { - warn!(pid, err = %err, "failed to reap handoff import server during rollback"); - } - } -} - #[cfg(unix)] pub(crate) fn bind_listener(socket_path: &Path) -> io::Result { let _ = std::fs::remove_file(socket_path); @@ -189,8 +171,8 @@ pub(crate) fn send_fds_and_wait_restored(stream: &mut UnixStream, fds: &[RawFd]) Ok(()) } -#[cfg(unix)] -pub(crate) fn wait_ready(stream: &mut UnixStream) -> io::Result<()> { +#[cfg(any(unix, windows))] +pub(crate) fn wait_ready(stream: &mut HandoffStream) -> io::Result<()> { stream.set_read_timeout(Some(READY_TIMEOUT))?; let ready = read_line_unbuffered(&mut *stream)?; if ready.trim_end() != "ready" { @@ -199,14 +181,14 @@ pub(crate) fn wait_ready(stream: &mut UnixStream) -> io::Result<()> { Ok(()) } -#[cfg(unix)] -pub(crate) fn report_committed(stream: &mut UnixStream) -> io::Result<()> { - stream.write_all(b"committed\n")?; - stream.flush() +#[cfg(any(unix, windows))] +pub(crate) fn report_committed(stream: &mut HandoffStream) -> io::Result<()> { + // Completing this write is irreversible. Do not add a fallible flush. + stream.write_all(b"committed\n") } -#[cfg(unix)] -pub(crate) fn wait_owned_ack(stream: &mut UnixStream) { +#[cfg(any(unix, windows))] +pub(crate) fn wait_owned_ack(stream: &mut HandoffStream) { if let Err(err) = stream.set_read_timeout(Some(OWNED_ACK_TIMEOUT)) { warn!(err = %err, "failed to set handoff ownership ack timeout"); return; @@ -235,6 +217,19 @@ pub(crate) fn receive(socket_path: &Path, token: &str) -> io::Result io::Result<()> { if manifest.version != HANDOFF_VERSION { return Err(io::Error::other(format!( "unsupported handoff version {}", @@ -262,31 +257,24 @@ pub(crate) fn receive(socket_path: &Path, token: &str) -> io::Result io::Result<()> { +pub(crate) fn report_restored(stream: &mut HandoffStream) -> io::Result<()> { stream.write_all(b"restored\n")?; stream.flush() } -#[cfg(unix)] -pub(crate) fn report_ready(stream: &mut UnixStream) -> io::Result<()> { +#[cfg(any(unix, windows))] +pub(crate) fn report_ready(stream: &mut HandoffStream) -> io::Result<()> { stream.write_all(b"ready\n")?; stream.flush() } -#[cfg(unix)] -pub(crate) fn wait_committed(stream: &mut UnixStream) -> io::Result<()> { - stream.set_read_timeout(Some(READY_TIMEOUT))?; +#[cfg(any(unix, windows))] +pub(crate) fn wait_committed(stream: &mut HandoffStream) -> io::Result<()> { + stream.set_read_timeout(Some(COMMIT_TIMEOUT))?; let committed = read_line_unbuffered(&mut *stream)?; if committed.trim_end() != "committed" { return Err(io::Error::other("handoff source did not commit")); @@ -294,13 +282,13 @@ pub(crate) fn wait_committed(stream: &mut UnixStream) -> io::Result<()> { Ok(()) } -#[cfg(unix)] -pub(crate) fn report_owned(stream: &mut UnixStream) -> io::Result<()> { +#[cfg(any(unix, windows))] +pub(crate) fn report_owned(stream: &mut HandoffStream) -> io::Result<()> { stream.write_all(b"owned\n")?; stream.flush() } -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) fn manifest_for( snapshot: crate::persist::SessionSnapshot, panes: Vec, @@ -320,6 +308,144 @@ pub(crate) fn manifest_for( } } +#[cfg(windows)] +#[derive(Serialize, Deserialize)] +struct WindowsResources { + panes: Vec<[usize; 6]>, + listeners: [usize; 2], +} + +#[cfg(windows)] +pub(crate) fn bind_listener(path: &Path) -> io::Result { + crate::platform::TransferableLocalListener::bound(crate::ipc::bind_private_local_listener( + path, + )?) +} + +#[cfg(windows)] +pub(crate) fn accept_windows_handoff( + listener: crate::platform::TransferableLocalListener, + child: &Child, + token: &str, + manifest: &HandoffManifest, + panes: Vec, + listeners: [crate::platform::WindowsListenerHandoff; 2], +) -> io::Result { + let deadline = std::time::Instant::now() + READY_TIMEOUT; + let stream = loop { + match listener.accept() { + Ok(stream) => break stream, + Err(err) if err.kind() == io::ErrorKind::WouldBlock => { + if std::time::Instant::now() >= deadline { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "handoff accept timed out", + )); + } + std::thread::sleep(Duration::from_millis(5)); + } + Err(err) => return Err(err), + } + }; + if crate::platform::named_pipe_peer_pid(&stream)? != child.id() { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "handoff peer is not the spawned replacement", + )); + } + let mut stream = HandoffStream::new(stream, READY_TIMEOUT)?; + if read_line_unbuffered(&mut stream)?.trim_end() != token { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "handoff import token mismatch", + )); + } + serde_json::to_writer(&mut stream, manifest).map_err(io::Error::other)?; + stream.write_all(b"\n")?; + if read_line_unbuffered(&mut stream)?.trim_end() != "validated" { + return Err(io::Error::other("handoff import did not validate manifest")); + } + // From this point, only the target owns these duplicates. On failure the + // caller kills and reaps the exact child before resuming the source. + let resources = WindowsResources { + panes: panes + .into_iter() + .map(|pty| pty.into_raw_handles()) + .collect(), + listeners: listeners.map(|listener| listener.into_raw_handle()), + }; + serde_json::to_writer(&mut stream, &resources).map_err(io::Error::other)?; + stream.write_all(b"\n")?; + wait_ready(&mut stream)?; + Ok(stream) +} + +#[cfg(windows)] +pub(crate) fn receive(path: &Path, token: &str) -> io::Result { + use std::os::windows::io::{FromRawHandle, OwnedHandle}; + let mut stream = HandoffStream::new(crate::ipc::connect_local_stream(path)?, READY_TIMEOUT)?; + writeln!(stream, "{token}")?; + let manifest: HandoffManifest = + serde_json::from_str(&read_line_unbuffered(&mut stream)?).map_err(io::Error::other)?; + validate_manifest(&manifest)?; + stream.write_all(b"validated\n")?; + let resources: WindowsResources = + serde_json::from_str(&read_line_unbuffered(&mut stream)?).map_err(io::Error::other)?; + let mut handles = std::collections::HashSet::new(); + if resources.panes.len() != manifest.panes.len() + || !resources + .panes + .iter() + .flatten() + .chain(resources.listeners.iter()) + .all(|handle| *handle != 0 && *handle != usize::MAX && handles.insert(*handle)) + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "handoff resources do not match manifest", + )); + } + // The private source transfers target-local handles; validate cardinality + // and uniqueness before establishing exactly one RAII owner for each. + let ptys = resources + .panes + .into_iter() + .map(|handles| unsafe { crate::pty::backend::WindowsPtyHandoff::from_raw_handles(handles) }) + .collect(); + let [api, client] = resources + .listeners + .map(|handle| unsafe { OwnedHandle::from_raw_handle(handle as _) }); + let listeners = [ + crate::platform::TransferableLocalListener::from_handoff_handle( + api, + &crate::api::socket_path(), + )?, + crate::platform::TransferableLocalListener::from_handoff_handle( + client, + &crate::server::socket_paths::client_socket_path(), + )?, + ]; + Ok(ReceivedHandoff { + manifest, + ptys, + listeners, + stream, + }) +} + +#[cfg(any(unix, windows))] +pub(crate) fn cleanup_failed_import_child(child: &mut Child) -> io::Result<()> { + if child.try_wait()?.is_none() { + if let Err(error) = child.kill() { + if child.try_wait()?.is_none() { + return Err(error); + } + } + child.wait()?; + } + Ok(()) +} + #[cfg(unix)] fn restrict_socket_permissions(path: &Path) -> io::Result<()> { use std::os::unix::fs::PermissionsExt; @@ -351,8 +477,8 @@ fn accept_with_timeout( } } -#[cfg(unix)] -fn read_line_unbuffered(stream: &mut UnixStream) -> io::Result { +#[cfg(any(unix, windows))] +fn read_line_unbuffered(stream: &mut HandoffStream) -> io::Result { let mut bytes = Vec::new(); let mut byte = [0u8; 1]; loop { @@ -520,7 +646,7 @@ fn recv_fd_batch(stream: &UnixStream, wanted: usize) -> io::Result> { Ok(out) } -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) fn log_import_result(panes: usize) { info!(panes, "handoff import ready"); } diff --git a/src/server/headless.rs b/src/server/headless.rs index 01d91bbd3e..602feff64e 100644 --- a/src/server/headless.rs +++ b/src/server/headless.rs @@ -21,15 +21,12 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +#[cfg(unix)] use interprocess::local_socket::traits::Listener as _; -#[cfg(windows)] -use interprocess::local_socket::traits::Stream as _; #[cfg(unix)] use interprocess::local_socket::ListenerNonblockingMode; use ratatui::layout::Rect; use tokio::sync::mpsc; -#[cfg(windows)] -use tracing::error; use tracing::{debug, info, warn}; use base64::Engine; @@ -39,9 +36,10 @@ use crate::api; use crate::app; use crate::config; use crate::events::AppEvent; +#[cfg(unix)] +use crate::ipc::LocalListener; use crate::ipc::{ - bind_local_listener, remove_socket_file_if_owned, socket_file_identity, LocalListener, - SocketFileIdentity, + bind_local_listener, remove_socket_file_if_owned, socket_file_identity, SocketFileIdentity, }; use crate::protocol::{ self, AttachScrollDirection, AttachScrollSource, FrameData, ServerMessage, MAX_FRAME_SIZE, @@ -197,12 +195,13 @@ pub struct HeadlessServer { #[cfg(unix)] api_tx: Option, // Kept on every platform so dropping HeadlessServer owns API server shutdown. - #[cfg_attr(windows, allow(dead_code))] api_server: Option, #[cfg(unix)] client_listener: LocalListener, + #[cfg(windows)] + client_listener_control: crate::platform::WindowsListenerControl, client_socket_path: PathBuf, - client_socket_identity: SocketFileIdentity, + client_socket_identity: Option, clients: HashMap, #[cfg(unix)] next_client_id: u64, @@ -258,33 +257,23 @@ pub struct HeadlessServer { #[cfg(windows)] fn spawn_windows_client_accept_thread( - listener: LocalListener, + listener: crate::platform::TransferableLocalListener, should_quit: Arc, server_event_tx: mpsc::Sender, -) { - std::thread::spawn(move || { - let mut next_client_id = 1_u64; - while !should_quit.load(Ordering::Acquire) { - let stream = match listener.accept() { - Ok(stream) => stream, - Err(err) => { - if should_quit.load(Ordering::Acquire) { - break; - } - error!(err = %err, "client listener accept failed"); - std::thread::sleep(Duration::from_millis(50)); - continue; - } - }; - +) -> crate::platform::WindowsListenerControl { + let stop = should_quit.clone(); + let mut next_client_id = 1_u64; + let (_, control) = crate::platform::spawn_transferable_listener( + listener, + "client listener", + move || stop.load(Ordering::Acquire), + move |mut stream| { let client_id = next_client_id; next_client_id = next_client_id.saturating_add(1); - - if let Err(err) = stream.set_nonblocking(true) { - warn!(err = %err, "failed to set client stream nonblocking"); - continue; + if let Err(err) = crate::ipc::set_local_stream_polling(&mut stream, true) { + warn!(err = %err, "failed to configure client stream polling"); + return; } - let should_quit = should_quit.clone(); let server_event_tx = server_event_tx.clone(); std::thread::spawn(move || { @@ -297,8 +286,9 @@ fn spawn_windows_client_accept_thread( debug!(client_id, err = %err, "client handshake failed"); } }); - } - }); + }, + ); + control } impl HeadlessServer { @@ -323,14 +313,41 @@ impl HeadlessServer { let client_socket_identity = socket_file_identity(&client_path)?; info!(path = %client_path.display(), "client protocol socket listening"); - // Set non-blocking on Unix so we can poll it from the event loop. + #[cfg(windows)] + let listener = crate::platform::TransferableLocalListener::bound(listener)?; #[cfg(unix)] listener.set_nonblocking(ListenerNonblockingMode::Accept)?; + Ok(Self::new_on_listener( + app, + config_diagnostics, + api_tx, + api_server, + should_quit, + listener, + client_socket_identity, + )) + } + + fn new_on_listener( + app: app::App, + config_diagnostics: &[String], + api_tx: Option, + api_server: Option, + should_quit: Arc, + #[cfg(unix)] listener: LocalListener, + #[cfg(windows)] listener: crate::platform::TransferableLocalListener, + client_socket_identity: SocketFileIdentity, + ) -> Self { + let client_path = client_socket_path(); // Channel for server events from client threads. let (server_event_tx, server_event_rx) = mpsc::channel(64); #[cfg(windows)] - spawn_windows_client_accept_thread(listener, should_quit.clone(), server_event_tx.clone()); + let client_listener_control = spawn_windows_client_accept_thread( + listener, + should_quit.clone(), + server_event_tx.clone(), + ); let server_keybindings = app_keybindings(&app); let headless_size = app.state.headless_size; @@ -338,15 +355,17 @@ impl HeadlessServer { server_config_diagnostic_summaries(config_diagnostics); #[cfg(not(unix))] let _ = api_tx; - Ok(Self { + Self { app, #[cfg(unix)] api_tx, api_server, #[cfg(unix)] client_listener: listener, + #[cfg(windows)] + client_listener_control, client_socket_path: client_path, - client_socket_identity, + client_socket_identity: Some(client_socket_identity), clients: HashMap::new(), #[cfg(unix)] next_client_id: 1, @@ -379,7 +398,7 @@ impl HeadlessServer { should_quit, server_event_rx, server_event_tx, - }) + } } /// Runs the headless server event loop until shutdown. @@ -734,7 +753,7 @@ impl HeadlessServer { stamp } - #[cfg(unix)] + #[cfg(any(unix, windows))] fn resize_shared_runtime_to_effective_size(&mut self) { self.resize_shared_runtime_to_effective_size_with_pending_agent_resumes(true); } @@ -1104,8 +1123,7 @@ impl HeadlessServer { ) } - /// Windows named-pipe clients can block in connect unless the server has a - /// pending blocking accept. The dedicated accept thread handles that path. + /// The dedicated Windows accept thread polls the pending pipe instance. #[cfg(windows)] fn accept_client_connections(&mut self) -> io::Result<()> { Ok(()) @@ -1773,7 +1791,7 @@ impl HeadlessServer { } } - #[cfg(unix)] + #[cfg(any(unix, windows))] fn disconnect_all_clients_for_handoff(&mut self) { let client_ids = self.clients.keys().copied().collect::>(); for client_id in client_ids { diff --git a/src/server/headless/bootstrap.rs b/src/server/headless/bootstrap.rs index e5ff400dcc..63ea6f1e6a 100644 --- a/src/server/headless/bootstrap.rs +++ b/src/server/headless/bootstrap.rs @@ -125,7 +125,7 @@ fn take_startup_cwd() -> Option { (!cwd.is_empty()).then(|| PathBuf::from(cwd)) } -#[cfg(unix)] +#[cfg(any(unix, windows))] fn run_handoff_import_server(socket_path: &Path, token: &str) -> io::Result<()> { let loaded_config = config::Config::load(); let mut received = crate::server::handoff::receive(socket_path, token)?; @@ -136,6 +136,7 @@ fn run_handoff_import_server(socket_path: &Path, token: &str) -> io::Result<()> let should_quit = Arc::new(AtomicBool::new(false)); let mut imports = HashMap::new(); + #[cfg(unix)] for (pane, fd) in received.manifest.panes.into_iter().zip(received.fds) { let pane_id = pane.pane_id; imports.insert( @@ -147,6 +148,19 @@ fn run_handoff_import_server(socket_path: &Path, token: &str) -> io::Result<()> ); } + #[cfg(windows)] + for (pane, windows_pty) in received.manifest.panes.into_iter().zip(received.ptys) { + imports.insert( + pane.pane_id, + crate::handoff_runtime::ImportedHandoffRuntime { + windows_pty, + state: pane, + }, + ); + } + #[cfg(windows)] + let [api_listener, client_listener] = received.listeners; + let rt = tokio::runtime::Builder::new_multi_thread() .enable_all() .build() @@ -161,37 +175,79 @@ fn run_handoff_import_server(socket_path: &Path, token: &str) -> io::Result<()> &received.manifest.snapshot, &mut imports, )?; - crate::server::handoff::report_restored(&mut received.stream)?; - if std::env::var("HERDR_TEST_HANDOFF_IMPORT_FAIL").as_deref() == Ok("after_restored") { - return Err(io::Error::other( - "test handoff import failure after restored", - )); - } - wait_for_old_public_sockets_to_close(Duration::from_secs(5))?; + #[cfg(unix)] + let mut server = { + crate::server::handoff::report_restored(&mut received.stream)?; + if std::env::var("HERDR_TEST_HANDOFF_IMPORT_FAIL").as_deref() == Ok("after_restored") { + return Err(io::Error::other( + "test handoff import failure after restored", + )); + } + wait_for_old_public_sockets_to_close(Duration::from_secs(5))?; - let api_server = api::start_server_with_stop_control( - api_tx.clone(), - event_hub.clone(), - should_quit.clone(), - )?; - let mut server = HeadlessServer::new( - app, - &loaded_config.diagnostics, - Some(api_tx.clone()), - Some(api_server), - should_quit, - )?; - // Carried across before any client attaches, so the first title sent is - // the override rather than the configured one it replaced. + let api_server = api::start_server_with_stop_control( + api_tx.clone(), + event_hub.clone(), + should_quit.clone(), + )?; + let server = HeadlessServer::new( + app, + &loaded_config.diagnostics, + Some(api_tx.clone()), + Some(api_server), + should_quit, + )?; + crate::server::handoff::report_ready(&mut received.stream)?; + crate::server::handoff::wait_committed(&mut received.stream)?; + server + }; + #[cfg(windows)] + let mut server = { + let api_marker = crate::platform::WindowsHandoffMarker::prepare(&api::socket_path())?; + let client_marker = + crate::platform::WindowsHandoffMarker::prepare(&client_socket_path())?; + if std::env::var("HERDR_TEST_HANDOFF_IMPORT_FAIL").as_deref() == Ok("after_restored") { + return Err(io::Error::other( + "test handoff import failure after restored", + )); + } + crate::server::handoff::report_ready(&mut received.stream)?; + crate::server::handoff::wait_committed(&mut received.stream)?; + if std::env::var("HERDR_TEST_HANDOFF_IMPORT_FAIL").as_deref() == Ok("marker_delayed") { + std::thread::sleep(Duration::from_secs(1)); + } + let api_server = api::start_server_from_handoff( + api_listener, + api_marker.publish(), + api_tx.clone(), + event_hub.clone(), + should_quit.clone(), + ); + HeadlessServer::new_on_listener( + app, + &loaded_config.diagnostics, + Some(api_tx.clone()), + Some(api_server), + should_quit, + client_listener, + client_marker.publish(), + ) + }; server.api_window_title = received.manifest.api_window_title.take(); - crate::server::handoff::report_ready(&mut received.stream)?; - crate::server::handoff::wait_committed(&mut received.stream)?; server.app.assume_handoff_ownership(); - server.app.unpause_handoff_readers(); - server.pending_handoff_repaint_nudge = true; - if let Err(err) = crate::server::handoff::report_owned(&mut received.stream) { - warn!(err = %err, "failed to report handoff ownership; continuing as owner"); + #[cfg(unix)] + { + server.app.unpause_handoff_readers(); + server.pending_handoff_repaint_nudge = true; + } + #[cfg(windows)] + server.app.activate_handoff_runtimes(); + if std::env::var("HERDR_TEST_HANDOFF_IMPORT_FAIL").as_deref() != Ok("lost_owned") { + if let Err(err) = crate::server::handoff::report_owned(&mut received.stream) { + warn!(err = %err, "failed to report handoff ownership; continuing as owner"); + } } + drop(received.stream); info!("handoff import server started"); print_ready_message(&api::socket_path(), &client_socket_path()); server.app.run_plugin_startup_hooks(); @@ -203,7 +259,7 @@ fn run_handoff_import_server(socket_path: &Path, token: &str) -> io::Result<()> result } -#[cfg(not(unix))] +#[cfg(not(any(unix, windows)))] fn run_handoff_import_server(_socket_path: &Path, _token: &str) -> io::Result<()> { Err(io::Error::other("live handoff is only supported on Unix")) } diff --git a/src/server/headless/lifecycle.rs b/src/server/headless/lifecycle.rs index f7ad87dd97..930004e588 100644 --- a/src/server/headless/lifecycle.rs +++ b/src/server/headless/lifecycle.rs @@ -21,30 +21,35 @@ pub(super) fn wait_for_live_handoff_response_write( } impl HeadlessServer { - #[cfg(unix)] + #[cfg(any(unix, windows))] pub(super) fn perform_live_handoff( &mut self, params: crate::api::schema::ServerLiveHandoffParams, ) -> io::Result<()> { - info!("starting live handoff"); - let import_exe = params.import_exe.as_deref().map(std::path::PathBuf::from); - let socket_path = crate::server::handoff::handoff_socket_path(); + use crate::server::handoff; + #[cfg(windows)] + if !crate::pty::backend::windows_handoff_available() + || self + .app + .terminal_runtimes + .values() + .any(|runtime| !runtime.windows_handoff_supported()) + { + return Err(io::Error::new( + io::ErrorKind::Unsupported, + "live handoff requires the bundled ConPTY runtime", + )); + } + let socket_path = handoff::handoff_socket_path(); let token = format!( "{}-{}", std::process::id(), - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) + SystemTime::now() + .duration_since(UNIX_EPOCH) .unwrap_or_default() .as_nanos() ); - let listener = match crate::server::handoff::bind_listener(&socket_path) { - Ok(listener) => listener, - Err(err) => { - self.handoff_in_progress = false; - return Err(err); - } - }; - + let listener = handoff::bind_listener(&socket_path)?; let mut pane_by_terminal = HashMap::new(); for ws in &self.app.state.workspaces { for tab in &ws.tabs { @@ -55,165 +60,178 @@ impl HeadlessServer { } self.handoff_in_progress = true; - self.disconnect_all_clients_for_handoff(); - let _ = reject_pending_client_connections(&self.client_listener); - let mut paused_terminal_ids = Vec::new(); - for terminal_id in pane_by_terminal.keys() { - if let Some(runtime) = self.app.terminal_runtimes.get(terminal_id) { - if let Err(err) = runtime.pause_handoff_reader(Duration::from_secs(2)) { - self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); - return Err(err); - } - paused_terminal_ids.push(terminal_id.clone()); - } - } - - let snapshot = crate::persist::capture( - &self.app.state.workspaces, - &self.app.state.terminals, - &self.app.terminal_runtimes, - self.app.state.active, - self.app.state.selected, - ); - - let mut handoff_entries = Vec::new(); - for (terminal_id, runtime) in self.app.terminal_runtimes.iter() { - let Some(pane_id) = pane_by_terminal.get(terminal_id).copied() else { - continue; - }; - let mut handoff_runtime = runtime.handoff_runtime_state(pane_id); - let has_agent_session = self - .app - .state - .terminals - .get(terminal_id) - .is_some_and(|terminal| terminal.persisted_agent_session.is_some()); - if !has_agent_session { - handoff_runtime.initial_history_ansi = runtime.handoff_history_ansi(); + let mut import_child = None; + #[cfg(unix)] + let mut public_sockets_released = false; + let transaction = (|| { + #[cfg(windows)] + { + self.api_server + .as_ref() + .ok_or_else(|| io::Error::other("API listener unavailable"))? + .pause_listener_for_handoff()?; + self.client_listener_control.pause()?; } - handoff_entries.push((terminal_id.clone(), handoff_runtime)); - } - - let panes = handoff_entries - .iter() - .map(|(_, runtime)| runtime.clone()) - .collect(); - let manifest = crate::server::handoff::manifest_for( - snapshot, - panes, - params.expected_protocol, - params.expected_version, - self.api_window_title.clone(), - ); - let mut import_child = match crate::server::handoff::spawn_handoff_import( - import_exe.as_deref(), - &socket_path, - &token, - ) { - Ok(child) => child, - Err(err) => { - self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); - return Err(err); + self.disconnect_all_clients_for_handoff(); + #[cfg(unix)] + let _ = reject_pending_client_connections(&self.client_listener); + for terminal_id in pane_by_terminal.keys() { + if let Some(runtime) = self.app.terminal_runtimes.get(terminal_id) { + runtime.pause_handoff_reader(Duration::from_secs(2))?; + paused_terminal_ids.push(terminal_id.clone()); + } } - }; - let child_pid = import_child.id(); - info!(pid = child_pid, socket = %socket_path.display(), "spawned handoff import server"); - - let mut fds = Vec::new(); - let duplicate_result = (|| { - for (terminal_id, _) in &handoff_entries { - let Some(runtime) = self.app.terminal_runtimes.get(terminal_id) else { + let snapshot = crate::persist::capture( + &self.app.state.workspaces, + &self.app.state.terminals, + &self.app.terminal_runtimes, + self.app.state.active, + self.app.state.selected, + ); + let mut entries = Vec::new(); + for (terminal_id, runtime) in self.app.terminal_runtimes.iter() { + let Some(pane_id) = pane_by_terminal.get(terminal_id).copied() else { continue; }; - fds.push(runtime.duplicate_handoff_fd()?); + let mut state = runtime.handoff_runtime_state(pane_id); + if self + .app + .state + .terminals + .get(terminal_id) + .is_none_or(|terminal| terminal.persisted_agent_session.is_none()) + { + state.initial_history_ansi = runtime.handoff_history_ansi(); + } + entries.push((runtime, state)); } - Ok::<(), io::Error>(()) + let manifest = handoff::manifest_for( + snapshot, + entries.iter().map(|(_, state)| state.clone()).collect(), + params.expected_protocol, + params.expected_version, + self.api_window_title.clone(), + ); + let _child = import_child.insert(handoff::spawn_handoff_import( + params.import_exe.as_deref().map(Path::new), + &socket_path, + &token, + )?); + #[cfg(windows)] + let mut stream = { + let child = _child; + crate::platform::ensure_same_process_session(child.id())?; + let panes = entries + .iter() + .map(|(runtime, _)| runtime.duplicate_windows_handoff(child)) + .collect::>>()?; + let api = self + .api_server + .as_ref() + .ok_or_else(|| io::Error::other("API listener unavailable"))? + .duplicate_listener_for_handoff(child)?; + let client = self.client_listener_control.duplicate_for_handoff(child)?; + handoff::accept_windows_handoff( + listener, + child, + &token, + &manifest, + panes, + [api, client], + )? + }; + #[cfg(unix)] + let mut stream = { + use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; + let fds = entries + .iter() + .map(|(runtime, _)| { + runtime + .duplicate_handoff_fd() + .map(|fd| unsafe { OwnedFd::from_raw_fd(fd) }) + }) + .collect::>>()?; + let mut stream = + handoff::accept_and_validate_on(listener, &socket_path, &token, &manifest)?; + handoff::send_fds_and_wait_restored( + &mut stream, + &fds.iter().map(AsRawFd::as_raw_fd).collect::>(), + )?; + if let Some(api) = &self.api_server { + let _ = api.remove_socket_file_if_owned(); + } else { + let _ = std::fs::remove_file(crate::api::socket_path()); + } + if let Some(identity) = &self.client_socket_identity { + let _ = remove_socket_file_if_owned(&self.client_socket_path, identity); + } + public_sockets_released = true; + handoff::wait_ready(&mut stream)?; + stream + }; + handoff::report_committed(&mut stream)?; + Ok::<_, io::Error>(stream) })(); - if let Err(err) = duplicate_result { - for fd in fds { - let _ = unsafe { libc::close(fd) }; - } - crate::server::handoff::cleanup_failed_import_child(&mut import_child); - self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); - return Err(err); - } - - let mut stream = match crate::server::handoff::accept_and_validate_on( - listener, - &socket_path, - &token, - &manifest, - ) { + let _ = std::fs::remove_file(&socket_path); + let mut stream = match transaction { Ok(stream) => stream, - Err(err) => { - for fd in fds { - let _ = unsafe { libc::close(fd) }; + Err(error) => { + // No source authority resumes while the target can still run. + if let Some(child) = import_child.as_mut() { + handoff::cleanup_failed_import_child(child).map_err(|cleanup| { + io::Error::other(format!("{error}; replacement cleanup failed: {cleanup}")) + })?; } - crate::server::handoff::cleanup_failed_import_child(&mut import_child); - self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); - return Err(err); + #[cfg(unix)] + let restore = if public_sockets_released { + self.wait_then_restore_public_sockets_after_failed_handoff() + } else { + Ok(()) + }; + #[cfg(windows)] + let restore = { + let api = self + .api_server + .as_ref() + .map(|api| api.resume_listener_after_handoff()) + .transpose(); + let client = self.client_listener_control.resume(); + api.and(client) + }; + for terminal_id in &paused_terminal_ids { + if let Some(runtime) = self.app.terminal_runtimes.get(terminal_id) { + runtime.set_handoff_reader_paused(false); + } + } + self.handoff_in_progress = false; + return Err(match restore { + Ok(()) => error, + Err(restore) => io::Error::other(format!( + "{error}; source listener restore failed: {restore}" + )), + }); } }; - - let send_result = crate::server::handoff::send_fds_and_wait_restored(&mut stream, &fds); - for fd in fds { - let _ = unsafe { libc::close(fd) }; - } - if let Err(err) = send_result { - crate::server::handoff::cleanup_failed_import_child(&mut import_child); - self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); - return Err(err); - } - - if let Some(api_server) = &self.api_server { - let _ = api_server.remove_socket_file_if_owned(); - } else { - let _ = std::fs::remove_file(crate::api::socket_path()); - } - let _ = remove_socket_file_if_owned(&self.client_socket_path, &self.client_socket_identity); - if let Err(err) = crate::server::handoff::wait_ready(&mut stream) { - crate::server::handoff::cleanup_failed_import_child(&mut import_child); - match self.wait_then_restore_public_sockets_after_failed_handoff() { - Ok(()) => { - self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); - } - Err(restore_err) => { - self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); - return Err(io::Error::other(format!( - "handoff replacement server did not become ready: {err}; old server could not restore public sockets: {restore_err}" - ))); + // COMMIT was fully delivered. OWNED and source release errors cannot roll back. + #[cfg(windows)] + { + if let Some(api) = &mut self.api_server { + if let Err(err) = api.release_listener_after_handoff() { + warn!(%err, "failed to release old API acceptance"); } } - return Err(io::Error::other(format!( - "handoff replacement server did not become ready: {err}" - ))); - } - if let Err(err) = crate::server::handoff::report_committed(&mut stream) { - crate::server::handoff::cleanup_failed_import_child(&mut import_child); - match self.wait_then_restore_public_sockets_after_failed_handoff() { - Ok(()) => { - self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); - } - Err(restore_err) => { - self.rollback_handoff_before_commit(&socket_path, &paused_terminal_ids); - return Err(io::Error::other(format!( - "handoff replacement server was ready, but commit failed: {err}; old server could not restore public sockets: {restore_err}" - ))); - } + self.client_socket_identity = None; + if let Err(err) = self.client_listener_control.release_after_commit() { + warn!(%err, "failed to release old client acceptance"); } - return Err(err); } - for (terminal_id, runtime) in self.app.terminal_runtimes.drain_for_handoff() { - if !pane_by_terminal.contains_key(&terminal_id) { - continue; + if pane_by_terminal.contains_key(&terminal_id) { + runtime.preserve_for_handoff(); } - debug!(terminal = %terminal_id, "preserving pane runtime for handoff"); - runtime.preserve_for_handoff(); } - crate::server::handoff::wait_owned_ack(&mut stream); - + handoff::wait_owned_ack(&mut stream); Ok(()) } @@ -224,7 +242,7 @@ impl HeadlessServer { info!("live handoff completed; old server exiting"); } - #[cfg(not(unix))] + #[cfg(not(any(unix, windows)))] pub(super) fn perform_live_handoff( &mut self, _params: crate::api::schema::ServerLiveHandoffParams, @@ -254,7 +272,7 @@ impl HeadlessServer { self.api_server = Some(api_server); self.client_listener = listener; self.client_socket_path = client_path; - self.client_socket_identity = client_socket_identity; + self.client_socket_identity = Some(client_socket_identity); Ok(()) } @@ -265,21 +283,6 @@ impl HeadlessServer { self.restore_public_sockets_after_failed_handoff() } - #[cfg(unix)] - fn rollback_handoff_before_commit( - &mut self, - socket_path: &Path, - paused_terminal_ids: &[crate::terminal::TerminalId], - ) { - for terminal_id in paused_terminal_ids { - if let Some(runtime) = self.app.terminal_runtimes.get(terminal_id) { - runtime.set_handoff_reader_paused(false); - } - } - self.handoff_in_progress = false; - let _ = std::fs::remove_file(socket_path); - } - #[cfg(unix)] pub(super) fn nudge_handoff_panes_on_first_client_attach(&mut self) { if !self.pending_handoff_repaint_nudge { @@ -353,15 +356,15 @@ impl HeadlessServer { /// Removes socket files created by the server. pub(super) fn cleanup_sockets(&self) -> io::Result<()> { - if let Err(err) = - remove_socket_file_if_owned(&self.client_socket_path, &self.client_socket_identity) - { - if err.kind() != io::ErrorKind::NotFound { - warn!( - path = %self.client_socket_path.display(), - err = %err, - "failed to remove client socket on shutdown" - ); + if let Some(identity) = &self.client_socket_identity { + if let Err(err) = remove_socket_file_if_owned(&self.client_socket_path, identity) { + if err.kind() != io::ErrorKind::NotFound { + warn!( + path = %self.client_socket_path.display(), + err = %err, + "failed to remove client socket on shutdown" + ); + } } } Ok(()) diff --git a/src/server/headless/tests/mod.rs b/src/server/headless/tests/mod.rs index 093b5625ff..337605da2f 100644 --- a/src/server/headless/tests/mod.rs +++ b/src/server/headless/tests/mod.rs @@ -1,4 +1,6 @@ use super::*; +#[cfg(windows)] +mod windows_handoff; #[path = "pane_graphics.rs"] mod pane_graphics_tests; @@ -62,7 +64,11 @@ fn test_headless_server_with_event_hub(event_hub: api::EventHub) -> HeadlessServ let (server_event_tx, server_event_rx) = mpsc::channel(64); let should_quit = Arc::new(AtomicBool::new(false)); #[cfg(windows)] - spawn_windows_client_accept_thread(listener, should_quit.clone(), server_event_tx.clone()); + let client_listener_control = spawn_windows_client_accept_thread( + crate::platform::TransferableLocalListener::bound(listener).unwrap(), + should_quit.clone(), + server_event_tx.clone(), + ); let server_keybindings = app_keybindings(&app); let headless_size = app.state.headless_size; @@ -73,8 +79,10 @@ fn test_headless_server_with_event_hub(event_hub: api::EventHub) -> HeadlessServ api_server: None, #[cfg(unix)] client_listener: listener, + #[cfg(windows)] + client_listener_control, client_socket_path: socket_path, - client_socket_identity, + client_socket_identity: Some(client_socket_identity), clients: HashMap::new(), #[cfg(unix)] next_client_id: 1, diff --git a/src/server/headless/tests/windows_handoff.rs b/src/server/headless/tests/windows_handoff.rs new file mode 100644 index 0000000000..bc1b6416be --- /dev/null +++ b/src/server/headless/tests/windows_handoff.rs @@ -0,0 +1,576 @@ +//! Native ownership tests for the Windows headless handoff boundary. +use crate::{ + ipc, + protocol::{self, ClientMessage, ServerMessage}, +}; +use serde_json::{json, Value}; +use std::{ + fs, + io::{BufRead, BufReader, Write}, + path::PathBuf, + process::{Child, Command, Stdio}, + thread, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, +}; + +struct Session { + root: PathBuf, + name: String, + binary: PathBuf, + child: Option, +} + +impl Session { + fn new(mode: &str) -> Self { + let root = std::env::temp_dir().join(format!( + "herdr-native-handoff-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + let name = "handoff-owned".to_owned(); + fs::create_dir_all(root.join("herdr-dev")).unwrap(); + fs::write( + root.join("herdr-dev/config.toml"), + "onboarding = false\n[terminal]\ndefault_shell = 'pwsh.exe -NoLogo -NoProfile'\n", + ) + .unwrap(); + let binary = + PathBuf::from(std::env::var_os("HERDR_HANDOFF_TEST_BIN").expect( + "set HERDR_HANDOFF_TEST_BIN to the staged debug herdr.exe with conpty bundle", + )); + let mut session = Self { + root, + name, + binary, + child: None, + }; + let mut command = session.command(); + command.arg("server"); + if mode == "system" { + command.env("HERDR_WINDOWS_CONPTY", "system"); + } else if !mode.is_empty() { + command.env("HERDR_TEST_HANDOFF_IMPORT_FAIL", mode); + } + session.child = Some(command.spawn().unwrap()); + wait(|| session.api_path().exists()); + assert_eq!(session.request("ping", json!({}))["result"]["type"], "pong"); + session + } + + fn command(&self) -> Command { + let mut command = Command::new(&self.binary); + command + .env("XDG_CONFIG_HOME", &self.root) + .env("XDG_STATE_HOME", &self.root) + .env("HERDR_SESSION", &self.name) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + for key in [ + "HERDR_SOCKET_PATH", + "HERDR_CLIENT_SOCKET_PATH", + "HERDR_CONFIG_PATH", + "HERDR_STARTUP_CWD", + "HERDR_WORKSPACE_ID", + "HERDR_TAB_ID", + "HERDR_PANE_ID", + "HERDR_WINDOWS_CONPTY", + "HERDR_TEST_HANDOFF_IMPORT_FAIL", + ] { + command.env_remove(key); + } + crate::platform::detach_server_daemon_command(&mut command); + command + } + + fn directory(&self) -> PathBuf { + self.root.join("herdr-dev/sessions").join(&self.name) + } + fn api_path(&self) -> PathBuf { + self.directory().join("herdr.sock") + } + fn client_path(&self) -> PathBuf { + self.directory().join("herdr-client.sock") + } + fn pid(&self) -> u32 { + fs::read_to_string(self.api_path()) + .unwrap() + .split(':') + .next() + .unwrap() + .parse() + .unwrap() + } + + fn request(&self, method: &str, params: Value) -> Value { + let stream = ipc::connect_local_stream(&self.api_path()).unwrap(); + let mut stream = + crate::platform::WindowsHandoffStream::new(stream, Duration::from_secs(15)).unwrap(); + writeln!( + stream, + "{}", + json!({"id":"native", "method":method, "params":params}) + ) + .unwrap(); + let mut line = String::new(); + BufReader::new(stream).read_line(&mut line).unwrap(); + serde_json::from_str(&line).unwrap_or_else(|error| panic!("{method}: {error}: {line}")) + } + + fn ok(&self, method: &str, params: Value) -> Value { + let value = self.request(method, params); + assert!(value.get("error").is_none(), "{method}: {value}"); + value["result"].clone() + } + + fn panes(&self) -> [String; 2] { + let first = self.ok("workspace.create", json!({"cwd":self.root,"focus":true}))["root_pane"] + ["pane_id"] + .as_str() + .unwrap() + .to_owned(); + let second = self.ok( + "pane.split", + json!({"target_pane_id":first,"direction":"right","focus":false}), + )["pane"]["pane_id"] + .as_str() + .unwrap() + .to_owned(); + [first, second] + } + + fn input(&self, pane: &str, text: &str) { + self.ok( + "pane.send_input", + json!({"pane_id":pane,"text":text,"keys":["Enter"]}), + ); + } + + fn observe_child(&self, pane: &str, label: &str) -> String { + let path = self.root.join(label); + self.input(pane, &format!("[IO.File]::WriteAllText('{}', \"$PID|$([Console]::WindowWidth)|$([Console]::WindowHeight)|ü🦀\")", path.to_string_lossy().replace('\'', "''"))); + let mut value = String::new(); + wait(|| { + value = fs::read_to_string(&path).unwrap_or_default(); + value.ends_with("ü🦀") + }); + value + } + + fn tui(&self, cols: u16, rows: u16) -> crate::platform::WindowsHandoffStream { + let stream = ipc::connect_local_stream(&self.client_path()).unwrap(); + let mut stream = + crate::platform::WindowsHandoffStream::new(stream, Duration::from_secs(10)).unwrap(); + protocol::write_message(&mut stream, &hello(cols, rows)).unwrap(); + let welcome: ServerMessage = + protocol::read_message(&mut stream, protocol::MAX_FRAME_SIZE).unwrap(); + assert!( + matches!(welcome, ServerMessage::Welcome { error: None, .. }), + "{welcome:?}" + ); + stream + } +} + +impl Drop for Session { + fn drop(&mut self) { + use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle}; + use windows_sys::Win32::System::Threading::{ + OpenProcess, TerminateProcess, WaitForSingleObject, PROCESS_SYNCHRONIZE, + PROCESS_TERMINATE, + }; + // The marker belongs to the unique session launched by this test. + let owner = fs::read_to_string(self.api_path()) + .ok() + .and_then(|marker| marker.split(':').next()?.parse::().ok()) + .filter(|pid| *pid != std::process::id()) + .and_then(|pid| { + let raw = unsafe { OpenProcess(PROCESS_SYNCHRONIZE | PROCESS_TERMINATE, 0, pid) }; + (!raw.is_null()).then(|| unsafe { OwnedHandle::from_raw_handle(raw) }) + }); + // Only this test's named session is addressed, including after replacement. + if owner.is_some() { + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + self.request("server.stop", json!({})) + })); + } + if let Some(owner) = owner { + if unsafe { WaitForSingleObject(owner.as_raw_handle(), 5000) } == 258 { + unsafe { + TerminateProcess(owner.as_raw_handle(), 1); + WaitForSingleObject(owner.as_raw_handle(), 5000); + } + } + } + if let Some(child) = self.child.as_mut() { + let deadline = Instant::now() + Duration::from_secs(5); + while child.try_wait().ok().flatten().is_none() && Instant::now() < deadline { + thread::sleep(Duration::from_millis(20)); + } + if child.try_wait().ok().flatten().is_none() { + let _ = child.kill(); + } + let _ = child.wait(); + } + if std::thread::panicking() { + eprintln!("native handoff evidence: {}", self.root.display()); + } else { + let _ = fs::remove_dir_all(&self.root); + } + } +} + +fn wait(mut ready: impl FnMut() -> bool) { + let deadline = Instant::now() + Duration::from_secs(10); + while !ready() { + assert!( + Instant::now() < deadline, + "native handoff condition timed out" + ); + thread::sleep(Duration::from_millis(25)); + } +} + +fn hello(cols: u16, rows: u16) -> ClientMessage { + ClientMessage::TerminalHello { + version: protocol::PROTOCOL_VERSION, + cols, + rows, + cell_width_px: 8, + cell_height_px: 16, + pixel_mouse: false, + } +} + +#[test] +#[ignore = "requires HERDR_HANDOFF_TEST_BIN staged with the verified ConPTY bundle"] +fn windows_handoff_preserves_two_children_io_resize_and_exit() { + for mode in ["", "lost_owned", "activation_ack", "marker_delayed"] { + let session = Session::new(mode); + let panes = session.panes(); + let old_tui = session.tui(120, 40); + let before = [ + session.observe_child(&panes[0], "before0"), + session.observe_child(&panes[1], "before1"), + ]; + let source = session.pid(); + session.ok("server.live_handoff", json!({})); // This is the original accepted API connection's full reply. + if mode == "marker_delayed" { + // The source exits before publication. Both prepared files must stay linked. + wait(|| session.pid() != source); + assert!(session.client_path().exists()); + } + assert_ne!(source, session.pid()); + if mode == "lost_owned" { + assert!( + fs::read_to_string(session.directory().join("herdr-server.log")) + .unwrap() + .contains("ownership ack was not received") + ); + } + if mode == "activation_ack" { + assert!( + fs::read_to_string(session.directory().join("herdr-server.log")) + .unwrap() + .contains("failed to acknowledge handoff activation") + ); + } + drop(old_tui); + let mut tui = session.tui(160, 50); + protocol::write_message( + &mut tui, + &ClientMessage::Resize { + cols: 160, + rows: 50, + cell_width_px: 8, + cell_height_px: 16, + pixel_mouse: false, + }, + ) + .unwrap(); + for (index, pane) in panes.iter().enumerate() { + let after = session.observe_child(pane, &format!("after{index}")); + assert_eq!( + before[index].split('|').next(), + after.split('|').next(), + "shell process changed" + ); + assert_ne!(before[index], after, "ConPTY size did not change"); + let pid = after.split('|').next().unwrap().parse::().unwrap(); + assert_eq!( + session.ok("pane.process_info", json!({"pane_id":pane}))["process_info"] + ["shell_pid"], + pid + ); + } + tui.set_read_timeout(Some(Duration::from_secs(10))).unwrap(); + protocol::write_message(&mut tui, &ClientMessage::Detach).unwrap(); + drop(tui); + let reconnected = session.tui(140, 45); + session.input(&panes[1], "exit"); + wait(|| { + !session.ok("pane.list", json!({}))["panes"] + .as_array() + .unwrap() + .iter() + .any(|pane| pane["pane_id"] == panes[1]) + }); + drop(reconnected); + eprintln!("{mode:?}: source {source} -> target {}, two child PIDs and Unicode preserved; resize/reconnect/exit passed", session.pid()); + } +} + +#[test] +#[ignore = "requires HERDR_HANDOFF_TEST_BIN staged with the verified ConPTY bundle"] +fn windows_handoff_precommit_failure_resumes_original_children_and_listeners() { + use std::os::windows::fs::OpenOptionsExt as _; + for mode in ["after_restored", "marker_locked"] { + let session = Session::new(mode); + let panes = session.panes(); + let source = session.pid(); + let before = session.observe_child(&panes[0], "before"); + let marker = fs::read(session.api_path()).unwrap(); + let marker_lock = (mode == "marker_locked").then(|| { + fs::OpenOptions::new() + .read(true) + .share_mode(windows_sys::Win32::Storage::FileSystem::FILE_SHARE_READ) + .open(session.api_path()) + .unwrap() + }); + let response = session.request("server.live_handoff", json!({})); + assert!(response.get("error").is_some(), "{response}"); + assert_eq!(source, session.pid()); + assert_eq!(marker, fs::read(session.api_path()).unwrap()); + drop(marker_lock); + let tui = session.tui(120, 40); + let after = session.observe_child(&panes[0], "after"); + assert_eq!(before.split('|').next(), after.split('|').next()); + assert_eq!( + session.ok("pane.list", json!({}))["panes"] + .as_array() + .unwrap() + .len(), + 2 + ); + drop(tui); + } +} + +#[test] +#[ignore = "requires HERDR_HANDOFF_TEST_BIN staged with the verified ConPTY bundle"] +fn windows_handoff_empty_session_and_system_backend_refusal() { + let empty = Session::new(""); + let source = empty.pid(); + empty.ok("server.live_handoff", json!({})); + assert_ne!(empty.pid(), source); + assert!(empty.ok("pane.list", json!({}))["panes"] + .as_array() + .unwrap() + .is_empty()); + drop(empty.tui(120, 40)); + let system = Session::new("system"); + let source = system.pid(); + assert_eq!( + system.ok("ping", json!({}))["capabilities"]["live_handoff"], + false + ); + assert!(system + .request("server.live_handoff", json!({})) + .get("error") + .is_some()); + assert_eq!(system.pid(), source); + drop(system.tui(120, 40)); +} + +#[test] +#[ignore = "requires HERDR_HANDOFF_TEST_BIN staged with the verified ConPTY bundle"] +fn windows_handoff_both_pending_listeners_are_inert_until_commit_and_resume_on_rollback() { + use crate::platform::{spawn_transferable_listener, TransferableLocalListener}; + use crate::server::handoff; + for mode in ["rollback", "commit", "marker_publish"] { + let commit = mode != "rollback"; + let mut session = Session::new(""); + session.ok("server.stop", json!({})); + session.child.take().unwrap().wait().unwrap(); + let paths = [session.api_path(), session.client_path()]; + let (accepted_tx, accepted_rx) = std::sync::mpsc::channel(); + let listeners = paths.each_ref().map(|path| { + TransferableLocalListener::bound(ipc::bind_local_listener(path).unwrap()).unwrap() + }); + let original_markers = paths.each_ref().map(|path| fs::read(path).unwrap()); + let mut threads = Vec::new(); + let controls = listeners + .into_iter() + .enumerate() + .map(|(index, listener)| { + let tx = accepted_tx.clone(); + let (thread, control) = spawn_transferable_listener( + listener, + "handoff test source", + || false, + move |stream| { + tx.send((index, stream)).unwrap(); + }, + ); + threads.push(thread); + control + }) + .collect::>(); + let old_client = ipc::connect_local_stream(&paths[0]).unwrap(); + let (_, mut old_response) = accepted_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + for control in &controls { + control.pause().unwrap(); + } + let private_path = session.directory().join("test-handoff.sock"); + let private_listener = handoff::bind_listener(&private_path).unwrap(); + let target = session + .command() + .args(["server", "--handoff-import"]) + .arg(&private_path) + .arg("owned-test") + .env("HERDR_TEST_HANDOFF_IMPORT_FAIL", mode) + .spawn() + .unwrap(); + session.child = Some(target); + let target = session.child.as_ref().unwrap(); + let resources = [ + controls[0].duplicate_for_handoff(target).unwrap(), + controls[1].duplicate_for_handoff(target).unwrap(), + ]; + let snapshot = crate::persist::capture( + &[], + &std::collections::HashMap::new(), + &crate::terminal::TerminalRuntimeRegistry::new(), + None, + 0, + ); + let manifest = handoff::manifest_for(snapshot, Vec::new(), None, None, None); + let mut transaction = handoff::accept_windows_handoff( + private_listener, + target, + "owned-test", + &manifest, + Vec::new(), + resources, + ) + .unwrap(); + assert_eq!( + original_markers, + paths.each_ref().map(|path| fs::read(path).unwrap()) + ); + let mut api = ipc::connect_local_stream(&paths[0]).unwrap(); + let mut tui = ipc::connect_local_stream(&paths[1]).unwrap(); + writeln!( + api, + "{}", + json!({"id":"pending","method":"ping","params":{}}) + ) + .unwrap(); + protocol::write_message(&mut tui, &hello(120, 40)).unwrap(); + for _ in 0..10 { + for stream in [&mut api, &mut tui] { + assert!( + matches!( + ipc::poll_local_stream_read_count(stream, &mut [0; 1]).unwrap(), + ipc::LocalStreamReadCount::Pending + ), + "target answered before COMMIT" + ); + } + assert!( + accepted_rx.try_recv().is_err(), + "source accepted after idle ACK" + ); + thread::sleep(Duration::from_millis(10)); + } + if commit { + handoff::report_committed(&mut transaction).unwrap(); + for control in &controls { + control.release_after_commit().unwrap(); + } + handoff::wait_owned_ack(&mut transaction); + if mode == "marker_publish" { + assert!( + fs::read_to_string(session.directory().join("herdr-server.log")) + .unwrap() + .contains("failed to publish handoff socket marker") + ); + assert_eq!( + original_markers, + paths.each_ref().map(|path| fs::read(path).unwrap()) + ); + assert!(session + .child + .as_mut() + .unwrap() + .try_wait() + .unwrap() + .is_none()); + } + let api = + crate::platform::WindowsHandoffStream::new(api, Duration::from_secs(5)).unwrap(); + let mut line = String::new(); + BufReader::new(api).read_line(&mut line).unwrap(); + assert_eq!( + serde_json::from_str::(&line).unwrap()["result"]["type"], + "pong" + ); + let mut tui = + crate::platform::WindowsHandoffStream::new(tui, Duration::from_secs(5)).unwrap(); + let welcome: ServerMessage = + protocol::read_message(&mut tui, protocol::MAX_FRAME_SIZE).unwrap(); + assert!(matches!( + welcome, + ServerMessage::Welcome { error: None, .. } + )); + } else { + handoff::cleanup_failed_import_child(session.child.as_mut().unwrap()).unwrap(); + for control in &controls { + control.resume().unwrap(); + } + let mut replies = Vec::new(); + for _ in 0..2 { + let (index, mut stream) = accepted_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + if index == 0 { + let mut request = String::new(); + BufReader::new(&mut stream).read_line(&mut request).unwrap(); + } else { + let _: ClientMessage = + protocol::read_message(&mut stream, protocol::MAX_FRAME_SIZE).unwrap(); + } + stream.write_all(b"resumed\n").unwrap(); + replies.push(stream); + } + for stream in [api, tui] { + let stream = + crate::platform::WindowsHandoffStream::new(stream, Duration::from_secs(5)) + .unwrap(); + let mut line = String::new(); + BufReader::new(stream).read_line(&mut line).unwrap(); + assert_eq!(line, "resumed\n"); + } + for control in &controls { + control.release_after_commit().unwrap(); + } + for path in &paths { + fs::remove_file(path).unwrap(); + } + } + // An accepted source connection remains writable after either outcome. + old_response.write_all(b"original-response\n").unwrap(); + let old_client = + crate::platform::WindowsHandoffStream::new(old_client, Duration::from_secs(5)).unwrap(); + let mut line = String::new(); + BufReader::new(old_client).read_line(&mut line).unwrap(); + assert_eq!(line, "original-response\n"); + if mode == "marker_publish" { + session.ok("server.stop", json!({})); + } + for thread in threads { + thread.join().unwrap(); + } + } +} diff --git a/src/server/mod.rs b/src/server/mod.rs index 5d249587f3..7e02dc43e3 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -9,7 +9,7 @@ pub(crate) mod client_shell_graphics; pub(crate) mod client_transport; pub(crate) mod clients; pub(crate) mod clipboard_image; -#[cfg(unix)] +#[cfg(any(unix, windows))] pub(crate) mod handoff; pub mod headless; pub(crate) mod keybindings; diff --git a/src/terminal/runtime.rs b/src/terminal/runtime.rs index 4679579189..edf4b20fab 100644 --- a/src/terminal/runtime.rs +++ b/src/terminal/runtime.rs @@ -26,27 +26,27 @@ impl TerminalRuntime { self.0.duplicate_handoff_fd() } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn preserve_for_handoff(self) { self.0.preserve_for_handoff() } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn assume_handoff_ownership(&mut self) { self.0.assume_handoff_ownership(); } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn set_handoff_reader_paused(&self, paused: bool) { self.0.set_handoff_reader_paused(paused); } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn pause_handoff_reader(&self, timeout: std::time::Duration) -> std::io::Result<()> { self.0.pause_handoff_reader(timeout) } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn handoff_runtime_state( &self, pane_id: u32, @@ -54,13 +54,13 @@ impl TerminalRuntime { self.0.handoff_runtime_state(pane_id) } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub fn handoff_history_ansi(&self) -> Option { self.0.handoff_history_ansi() } - #[cfg(unix)] - pub fn from_handoff_fd( + #[cfg(any(unix, windows))] + pub fn from_handoff( import: crate::handoff_runtime::ImportedHandoffRuntime, scrollback_limit_bytes: usize, host_terminal_theme: crate::terminal_theme::TerminalTheme, @@ -69,7 +69,7 @@ impl TerminalRuntime { render_notify: Arc, render_dirty: Arc, ) -> std::io::Result { - crate::pane::PaneRuntime::from_handoff_fd( + crate::pane::PaneRuntime::from_handoff( import, scrollback_limit_bytes, host_terminal_theme, @@ -81,6 +81,24 @@ impl TerminalRuntime { .map(Self) } + #[cfg(windows)] + pub(crate) fn windows_handoff_supported(&self) -> bool { + self.0.windows_handoff_supported() + } + + #[cfg(windows)] + pub(crate) fn duplicate_windows_handoff( + &self, + target: &std::process::Child, + ) -> std::io::Result { + self.0.duplicate_windows_handoff(target) + } + + #[cfg(windows)] + pub(crate) fn activate_after_handoff(&self) -> std::io::Result<()> { + self.0.activate_after_handoff() + } + // Wrapper mirrors pane runtime construction arguments. #[allow(clippy::too_many_arguments)] pub fn spawn( diff --git a/src/terminal/runtime_registry.rs b/src/terminal/runtime_registry.rs index a909985e89..70ee45d919 100644 --- a/src/terminal/runtime_registry.rs +++ b/src/terminal/runtime_registry.rs @@ -37,7 +37,7 @@ impl TerminalRuntimeRegistry { self.runtimes.values() } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub(crate) fn iter(&self) -> impl Iterator { self.runtimes.iter() } @@ -49,7 +49,7 @@ impl TerminalRuntimeRegistry { } } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub(crate) fn assume_handoff_ownership(&mut self) { for runtime in self.runtimes.values_mut() { runtime.assume_handoff_ownership(); @@ -67,13 +67,24 @@ impl TerminalRuntimeRegistry { } } - #[cfg(unix)] + #[cfg(any(unix, windows))] pub(crate) fn drain_for_handoff( &mut self, ) -> impl Iterator + '_ { self.runtimes.drain() } + #[cfg(windows)] + pub(crate) fn activate_after_handoff(&self) { + for (terminal_id, runtime) in &self.runtimes { + if let Err(err) = runtime.activate_after_handoff() { + // A timeout leaves activation queued. Continue with every pane; + // after COMMIT an ACK error cannot unwind the owning server. + tracing::warn!(terminal = %terminal_id, %err, "failed to acknowledge handoff activation; continuing as owner"); + } + } + } + #[cfg(test)] pub(crate) fn drain(&mut self) -> impl Iterator + '_ { self.runtimes.drain()