From e42e5d224b42e1d6fbcfc638563961cbc1e2bba4 Mon Sep 17 00:00:00 2001 From: Dom Giandinoto Date: Wed, 7 Oct 2026 16:56:49 -0700 Subject: [PATCH 1/5] Piped stdin cancels/closes on command completion. Pipe classification changed from just relying on interactive trait of stdout --- .../isolation_session/common/manager.rs | 141 ++++++++++++------ .../isolation_session/common/pipe_relay.rs | 53 +++++-- 2 files changed, 140 insertions(+), 54 deletions(-) diff --git a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs index c17df26810..b98602a981 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs @@ -19,9 +19,11 @@ use crate::isolation_session_bindings::bindings::{ IsoSessionUserResult, }; use windows::Win32::Foundation::{HANDLE, WAIT_OBJECT_0}; +use windows::Win32::Storage::FileSystem::{GetFileType, FILE_TYPE_CHAR}; use windows::Win32::System::Com::{CoDecrementMTAUsage, CoIncrementMTAUsage, CO_MTA_USAGE_COOKIE}; use windows::Win32::System::Console::{ - GetStdHandle, STD_ERROR_HANDLE, STD_INPUT_HANDLE, STD_OUTPUT_HANDLE, + GetConsoleMode, GetStdHandle, CONSOLE_MODE, STD_ERROR_HANDLE, STD_INPUT_HANDLE, + STD_OUTPUT_HANDLE, }; use windows::Win32::System::Threading::{CreateEventW, SetEvent, WaitForSingleObject}; use windows::UI::WindowId; @@ -44,6 +46,23 @@ use super::process_options::{ // The OS account name is lifecycle addressing data, not diagnostic identity. const AGENT_IDENTITY_MARKER: &str = "isolation-session-agent"; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum StdinHandleKind { + Console, + PipeOrFile, +} + +fn classify_stdin_handle(handle: HANDLE) -> StdinHandleKind { + let file_type = unsafe { GetFileType(handle) }; + if file_type == FILE_TYPE_CHAR { + let mut mode = CONSOLE_MODE(0); + if unsafe { GetConsoleMode(handle, &mut mode) }.is_ok() { + return StdinHandleKind::Console; + } + } + StdinHandleKind::PipeOrFile +} + /// Keeps the process's MTA alive for as long as the lifecycle's WinRT objects, /// which outlive the call that created them. /// `CoIncrementMTAUsage` holds a reference releasable from any thread; @@ -426,10 +445,7 @@ impl IsolationSessionManager { let wxc_stdin = unsafe { GetStdHandle(STD_INPUT_HANDLE) } .map_err(|e| lifecycle_err(format!("GetStdHandle(stdin) failed: {}", e)))?; - // Manual-reset stop event for the stdin relay. Effective for a waitable - // `h_read` (console = TTY mode); for pipe handles it has no effect on a - // blocked `ReadFile`, so that relay misses its join and ends only when - // the calling process exits. Only `wxc-exec` reaches the non-TTY case. + // Manual-reset stop event for console-backed stdin relays. let stdin_stop_event = unsafe { CreateEventW(None, true, false, PCWSTR::null()) .map_err(|e| lifecycle_err(format!("CreateEventW(stdin stop): {}", e)))? @@ -525,13 +541,15 @@ impl IsolationSessionManager { } else { None }; - // Stdin: in interactive mode use the console-aware relay so - // `WINDOW_BUFFER_SIZE_EVENT` records propagate as - // `ResizeConsole(cols, rows)` calls on the agent's inner ConPTY. - // In non-interactive mode the agent's stdin is plain byte-oriented - // and the simpler stop-aware pipe relay is appropriate. + // Stdin relay selection is based on the actual local stdin handle, + // not on `options.interactive` (which is derived from stdout). A + // pipe/file read must be cancelled with `PipeReadCanceller`; a + // console handle can be interrupted by the stop event and, when the + // agent side is interactive, needs the console-aware relay for + // resize records. enum StdinRelayKind { None, + CancellablePipe, Pipe(PipeRelayWithStopParams), Console(ConsoleRelayParams), } @@ -539,37 +557,47 @@ impl IsolationSessionManager { let stdin_relay_state = if stdin_handle_val == 0 { StdinRelayKind::None } else { - // Owned, like the output relays': `process.Close()` releases the - // original while this relay may still be writing. - let stdin_h_write = - duplicate_handle(HANDLE(stdin_handle_val as *mut core::ffi::c_void)) - .map_err(|e| lifecycle_err(format!("duplicate stdin handle: {}", e)))?; - // Shares the event object, so `signal_stop` still reaches the relay. - let stdin_h_stop = duplicate_handle(stdin_stop_owned.get()) - .map_err(|e| lifecycle_err(format!("duplicate stdin stop event: {}", e)))?; - if options.interactive { - // Clone the WinRT process handle so the relay thread holds - // its own ref-counted reference (WinRT clone = AddRef), - // released when the thread frees the params it owns. - let process_for_resize = process.clone(); - let impersonation = self.impersonation.clone(); - StdinRelayKind::Console(ConsoleRelayParams { - h_read: wxc_stdin, - h_write: stdin_h_write, - h_stop_event: stdin_h_stop, - resize_callback: Box::new(move |cols, rows| { - let _ = owned_thread::call(&impersonation, || { - let _ = process_for_resize.ResizeConsole(cols, rows); - Ok(()) - }); - }), - }) - } else { - StdinRelayKind::Pipe(PipeRelayWithStopParams { - h_read: wxc_stdin, - h_write: stdin_h_write, - h_stop_event: stdin_h_stop, - }) + match classify_stdin_handle(wxc_stdin) { + StdinHandleKind::PipeOrFile => StdinRelayKind::CancellablePipe, + StdinHandleKind::Console => { + // Owned, like the output relays': `process.Close()` + // releases the original while this relay may still be + // writing. + let stdin_h_write = + duplicate_handle(HANDLE(stdin_handle_val as *mut core::ffi::c_void)) + .map_err(|e| { + lifecycle_err(format!("duplicate stdin handle: {}", e)) + })?; + // Shares the event object, so `signal_stop` still reaches the relay. + let stdin_h_stop = + duplicate_handle(stdin_stop_owned.get()).map_err(|e| { + lifecycle_err(format!("duplicate stdin stop event: {}", e)) + })?; + if options.interactive { + // Clone the WinRT process handle so the relay thread holds + // its own ref-counted reference (WinRT clone = AddRef), + // released when the thread frees the params it owns. + let process_for_resize = process.clone(); + let impersonation = self.impersonation.clone(); + StdinRelayKind::Console(ConsoleRelayParams { + h_read: wxc_stdin, + h_write: stdin_h_write, + h_stop_event: stdin_h_stop, + resize_callback: Box::new(move |cols, rows| { + let _ = owned_thread::call(&impersonation, || { + let _ = process_for_resize.ResizeConsole(cols, rows); + Ok(()) + }); + }), + }) + } else { + StdinRelayKind::Pipe(PipeRelayWithStopParams { + h_read: wxc_stdin, + h_write: stdin_h_write, + h_stop_event: stdin_h_stop, + }) + } + } } }; @@ -577,6 +605,17 @@ impl IsolationSessionManager { // them on the heap under the relay thread's ownership. scope.stdin = match stdin_relay_state { StdinRelayKind::None => None, + StdinRelayKind::CancellablePipe => { + let (thread, canceller) = unsafe { + create_relay_thread( + wxc_stdin, + HANDLE(stdin_handle_val as *mut core::ffi::c_void), + ) + } + .map_err(|e| lifecycle_err(format!("create stdin relay: {}", e)))?; + scope.stdin_canceller = Some(canceller); + Some(thread) + } StdinRelayKind::Pipe(params) => Some( unsafe { create_relay_thread_with_stop(params) } .map_err(|e| lifecycle_err(format!("create stdin relay: {}", e)))?, @@ -1041,6 +1080,10 @@ struct RelayScope<'a> { /// ends keeps them open past the workload's exit, so EOF alone is not a /// bound. output_cancellers: Vec, + /// Ends a pipe/file-backed stdin relay. Console stdin relays use the stop + /// event instead because `CancelSynchronousIo` has console-handle edge + /// cases. + stdin_canceller: Option, finished: bool, } @@ -1057,6 +1100,7 @@ impl<'a> RelayScope<'a> { stderr: None, stdin: None, output_cancellers: Vec::new(), + stdin_canceller: None, finished: false, } } @@ -1074,6 +1118,12 @@ impl<'a> RelayScope<'a> { } } + fn cancel_stdin_read(&self) { + if let Some(canceller) = &self.stdin_canceller { + canceller.close(); + } + } + /// Whether the relay exited. An absent relay was never spawned. fn join(relay: &Option, timeout_ms: u32) -> bool { match relay { @@ -1091,12 +1141,14 @@ impl<'a> RelayScope<'a> { fn stop_stdin(&self) { self.signal_stop(); + self.cancel_stdin_read(); Self::join(&self.stdin, Self::JOIN_MS); } /// Teardown for a workload that has already exited. fn finish(&mut self) { self.signal_stop(); + self.cancel_stdin_read(); let stdout_drained = Self::join(&self.stdout, Self::DRAIN_MS); let stderr_drained = Self::join(&self.stderr, Self::DRAIN_MS); if !(stdout_drained && stderr_drained) { @@ -1116,6 +1168,7 @@ impl Drop for RelayScope<'_> { return; } self.signal_stop(); + self.cancel_stdin_read(); self.cancel_output_reads(); let _ = self.process.Terminate(); Self::join(&self.stdout, Self::JOIN_MS); @@ -1302,9 +1355,9 @@ fn plan_wait( } /// Three-tier graceful shutdown for an `IsoSessionProcess` that's still -/// running after `WaitForExit(timeout_ms)` returns. Tier 1: close stdin — -/// many REPLs exit on EOF alone. Tier 2: `SendCtrlClose` — ConPTY-only; -/// `E_NOTIMPL` outside ConPTY, benign. Tier 3: force-terminate, wait +/// running after `WaitForExit(timeout_ms)` returns. Tier 1 asks the platform +/// to close stdin, tier 2 tries `SendCtrlClose` (ConPTY-only; `E_NOTIMPL` +/// outside ConPTY is benign), and tier 3 force-terminates, then waits /// infinitely (`WaitForExit(0)` = INFINITE) for the kill to land. /// /// The first `ExitCode()` query is `?`-propagated: a failure there means diff --git a/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs b/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs index e179f9b543..0fe72ea74f 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs @@ -6,11 +6,12 @@ //! desktop-session boundary. //! //! Two relay variants: -//! - [`create_relay_thread`] — cancellable; used for stdout / stderr. Ends at -//! EOF or when its [`PipeReadCanceller`] fires. -//! - [`create_relay_thread_with_stop`] — stop-event-aware; used for stdin -//! in TTY (ConPTY) mode where the agent can exit while the local stdin -//! handle remains open. +//! - [`create_relay_thread`] — cancellable; used for stdout / stderr and for +//! pipe/file-backed stdin. Ends at EOF or when its [`PipeReadCanceller`] +//! fires. +//! - [`create_relay_thread_with_stop`] — stop-event-aware; used for +//! console-backed stdin when the agent can exit while the local stdin handle +//! remains open. use std::io::Read; @@ -132,10 +133,9 @@ pub(super) unsafe fn create_relay_thread( // ── Stop-event-aware pipe relay ──────────────────────────────────────────── // -// Used for the IsolationSession stdin relay in TTY (ConPTY) mode, where the -// agent process can exit naturally while wxc-exec's stdin remains open (held -// by the parent: node-pty, shell, etc.). Without an external signal the relay -// sits in `ReadFile` forever. +// Used for the IsolationSession stdin relay when wxc-exec's stdin is a console +// handle and the agent process can exit naturally while that handle remains +// open. Without an external signal the relay sits in `ReadFile` forever. // // `CancelSynchronousIo` is the obvious alternative but has documented edge // cases on console handles, and wxc-exec's stdin is a console handle in the @@ -149,7 +149,7 @@ pub(super) unsafe fn create_relay_thread( // signalled when open", so the wait returns immediately, the relay enters // `ReadFile`, and from there the stop event cannot interrupt it. For // pipe-backed stdin (the non-TTY case), use the simpler EOF-driven -// `create_relay_thread` and rely on natural EOF or process exit for cleanup. +// `create_relay_thread` and rely on natural EOF or its canceller for cleanup. /// Parameters for a stop-event-aware relay thread. The thread loops /// `WaitForMultipleObjects({h_stop_event, h_read})`; copies a chunk when @@ -407,6 +407,39 @@ mod tests { ); } + /// Pipe-backed stdin uses `create_relay_thread` so shutdown can cancel the + /// blocked source read. Once cancelled, the relay must also release its + /// duplicate of the agent stdin write handle so the reader observes EOF. + #[test] + fn a_cancelled_stdin_pipe_relay_releases_its_destination_handle() { + let (source_read, _source_write) = create_test_pipes(); + let (mut dest_read, dest_write) = create_test_pipes(); + + let (relay_thread, canceller) = + unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; + + // Mirror `process.Close()`: the caller drops its original agent-stdin + // write handle while the relay's duplicate is still live. + drop(dest_write); + + assert_eq!( + unsafe { WaitForSingleObject(relay_thread.get(), 250) }, + WAIT_TIMEOUT, + "the relay should still be blocked before the cancel" + ); + + canceller.close(); + + assert_eq!( + unsafe { WaitForSingleObject(relay_thread.get(), 5000) }, + WAIT_OBJECT_0, + "the cancelled stdin relay should exit promptly" + ); + + let dest_send = SendOwnedHandle::take(&mut dest_read); + assert_eq!(read_from_pipe(dest_send.get()), ""); + } + /// Closing the source handle the relay was built from does not disturb it: /// it reads a duplicate. This is what lets `IsoSessionProcess::Close` run /// unconditionally. From d57a447d381d7ee9d615f060bc9ae3f5aba8adfa Mon Sep 17 00:00:00 2001 From: Dom Giandinoto Date: Tue, 13 Oct 2026 17:52:41 -0700 Subject: [PATCH 2/5] ensuring agent's stdin closes on EOF for piped wxc stdin, clarifying StdinRelayKind values --- .../isolation_session/common/manager.rs | 34 +++-- .../isolation_session/common/pipe_relay.rs | 118 +++++++++++++++++- .../src/core/mxc_common/process_util.rs | 7 ++ 3 files changed, 144 insertions(+), 15 deletions(-) diff --git a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs index b98602a981..002b347aa3 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs @@ -37,7 +37,8 @@ use super::error::{ }; use super::owned_thread::{self, Impersonation}; use super::pipe_relay::{ - create_relay_thread, create_relay_thread_with_stop, duplicate_handle, PipeRelayWithStopParams, + create_relay_thread, create_relay_thread_with_eof, create_relay_thread_with_stop, + duplicate_handle, PipeRelayWithStopParams, }; use super::process_options::{ build_iso_process_options, with_service_timeout_grace, ProcessOptions, @@ -549,16 +550,16 @@ impl IsolationSessionManager { // resize records. enum StdinRelayKind { None, - CancellablePipe, - Pipe(PipeRelayWithStopParams), - Console(ConsoleRelayParams), + PipeOrFile, + ConsoleByteStream(PipeRelayWithStopParams), + ConsoleInteractive(ConsoleRelayParams), } let stdin_relay_state = if stdin_handle_val == 0 { StdinRelayKind::None } else { match classify_stdin_handle(wxc_stdin) { - StdinHandleKind::PipeOrFile => StdinRelayKind::CancellablePipe, + StdinHandleKind::PipeOrFile => StdinRelayKind::PipeOrFile, StdinHandleKind::Console => { // Owned, like the output relays': `process.Close()` // releases the original while this relay may still be @@ -579,7 +580,7 @@ impl IsolationSessionManager { // released when the thread frees the params it owns. let process_for_resize = process.clone(); let impersonation = self.impersonation.clone(); - StdinRelayKind::Console(ConsoleRelayParams { + StdinRelayKind::ConsoleInteractive(ConsoleRelayParams { h_read: wxc_stdin, h_write: stdin_h_write, h_stop_event: stdin_h_stop, @@ -591,7 +592,7 @@ impl IsolationSessionManager { }), }) } else { - StdinRelayKind::Pipe(PipeRelayWithStopParams { + StdinRelayKind::ConsoleByteStream(PipeRelayWithStopParams { h_read: wxc_stdin, h_write: stdin_h_write, h_stop_event: stdin_h_stop, @@ -605,22 +606,33 @@ impl IsolationSessionManager { // them on the heap under the relay thread's ownership. scope.stdin = match stdin_relay_state { StdinRelayKind::None => None, - StdinRelayKind::CancellablePipe => { + StdinRelayKind::PipeOrFile => { + // Ensure the agent's stdin will close on EOF for pipe/file-backed + // stdin. + let process_for_eof = process.clone(); + let impersonation = self.impersonation.clone(); + let on_eof: Box = Box::new(move || { + let _ = owned_thread::call(&impersonation, || { + let _ = process_for_eof.CloseStandardInput(); + Ok(()) + }); + }); let (thread, canceller) = unsafe { - create_relay_thread( + create_relay_thread_with_eof( wxc_stdin, HANDLE(stdin_handle_val as *mut core::ffi::c_void), + Some(on_eof), ) } .map_err(|e| lifecycle_err(format!("create stdin relay: {}", e)))?; scope.stdin_canceller = Some(canceller); Some(thread) } - StdinRelayKind::Pipe(params) => Some( + StdinRelayKind::ConsoleByteStream(params) => Some( unsafe { create_relay_thread_with_stop(params) } .map_err(|e| lifecycle_err(format!("create stdin relay: {}", e)))?, ), - StdinRelayKind::Console(params) => { + StdinRelayKind::ConsoleInteractive(params) => { Some(unsafe { create_console_relay_thread(params) }.map_err(|e| { lifecycle_err(format!("create console-aware stdin relay: {}", e)) })?) diff --git a/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs b/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs index 0fe72ea74f..c4c8d7d888 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs @@ -52,6 +52,7 @@ pub(super) fn duplicate_handle(handle: HANDLE) -> Result struct PipeRelayParams { reader: InterruptiblePipeReader, h_write: OwnedHandle, + on_eof: Option>, } /// Thread procedure for relaying data between two handles. @@ -66,9 +67,13 @@ unsafe extern "system" fn pipe_relay_thread_proc(param: *mut core::ffi::c_void) let mut params = unsafe { Box::from_raw(param as *mut PipeRelayParams) }; let mut buffer = [0u8; BUFFER_SIZE as usize]; - loop { + let reached_source_eof = loop { let bytes_read = match params.reader.read(&mut buffer) { - Ok(0) | Err(_) => break, + // `Ok(0)` is reported for a genuine source EOF *and* for a fired + // canceller, so disambiguate with `was_cancelled` to determine if + // this is a genuine EOF. + Ok(0) => break !params.reader.was_cancelled(), + Err(_) => break false, Ok(n) => n, }; @@ -82,10 +87,16 @@ unsafe extern "system" fn pipe_relay_thread_proc(param: *mut core::ffi::c_void) .is_err() || bytes_written as usize != bytes_read { - break; + break false; } let _ = FlushFileBuffers(params.h_write.get()); + }; + + if reached_source_eof { + if let Some(on_eof) = params.on_eof.take() { + on_eof(); + } } 0 @@ -101,12 +112,26 @@ unsafe extern "system" fn pipe_relay_thread_proc(param: *mut core::ffi::c_void) pub(super) unsafe fn create_relay_thread( h_read: HANDLE, h_write: HANDLE, +) -> Result<(OwnedHandle, PipeReadCanceller), WxcError> { + unsafe { create_relay_thread_with_eof(h_read, h_write, None) } +} + +/// As [`create_relay_thread`], but runs `on_eof` once if the relay ends because +/// the source reached a genuine, uncancelled EOF. Pipe/file-backed stdin uses +/// this to propagate end-of-input to the agent (`CloseStandardInput`); a fired +/// canceller or a write error does **not** run it, so the forced-teardown path +/// stays distinct from natural end-of-input. +pub(super) unsafe fn create_relay_thread_with_eof( + h_read: HANDLE, + h_write: HANDLE, + on_eof: Option>, ) -> Result<(OwnedHandle, PipeReadCanceller), WxcError> { let reader = InterruptiblePipeReader::new(duplicate_handle(h_read)?); let canceller = reader.canceller(); let raw = Box::into_raw(Box::new(PipeRelayParams { reader, h_write: duplicate_handle(h_write)?, + on_eof, })); match unsafe { CreateThread( @@ -440,7 +465,92 @@ mod tests { assert_eq!(read_from_pipe(dest_send.get()), ""); } - /// Closing the source handle the relay was built from does not disturb it: + /// A genuine, uncancelled source EOF must run the relay's `on_eof` hook — + /// this is what closes the agent's stdin so a read-to-EOF workload exits. + #[test] + fn a_pipe_relay_at_source_eof_runs_its_eof_callback() { + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + + let (source_read, source_write) = create_test_pipes(); + let (mut dest_read, dest_write) = create_test_pipes(); + + let fired = Arc::new(AtomicBool::new(false)); + let fired_cb = Arc::clone(&fired); + let on_eof: Box = + Box::new(move || fired_cb.store(true, Ordering::SeqCst)); + + let (relay_thread, _canceller) = unsafe { + create_relay_thread_with_eof(source_read.get(), dest_write.get(), Some(on_eof)).unwrap() + }; + + // Drain concurrently so the relay's post-write FlushFileBuffers returns. + let dest_send = SendOwnedHandle::take(&mut dest_read); + let reader = std::thread::spawn(move || read_from_pipe(dest_send.get())); + + let mut written = 0u32; + unsafe { + WriteFile( + source_write.get(), + Some(b"payload"), + Some(&mut written), + None, + ) + .unwrap(); + } + // Source EOF: the relay copies the payload, reads `Ok(0)`, runs on_eof. + drop(source_write); + + assert_eq!( + unsafe { WaitForSingleObject(relay_thread.get(), 5000) }, + WAIT_OBJECT_0, + "the relay should reach EOF and exit" + ); + // The hook runs inside the thread proc before it returns, so by the time + // the thread is signalled the flag is already set. + assert!( + fired.load(Ordering::SeqCst), + "a genuine source EOF must run the EOF callback" + ); + + drop(dest_write); + assert_eq!(reader.join().unwrap(), "payload"); + } + + /// A cancellation also surfaces as `Ok(0)`, but it is forced teardown, not + /// end-of-input — the `on_eof` hook must not run (otherwise shutdown would + /// spuriously close the agent's stdin). + #[test] + fn a_cancelled_pipe_relay_does_not_run_its_eof_callback() { + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + + let (source_read, _source_write) = create_test_pipes(); + let (_dest_read, dest_write) = create_test_pipes(); + + let fired = Arc::new(AtomicBool::new(false)); + let fired_cb = Arc::clone(&fired); + let on_eof: Box = + Box::new(move || fired_cb.store(true, Ordering::SeqCst)); + + let (relay_thread, canceller) = unsafe { + create_relay_thread_with_eof(source_read.get(), dest_write.get(), Some(on_eof)).unwrap() + }; + + // Parked in `ReadFile` (`_source_write` keeps it from reaching EOF). + canceller.close(); + + assert_eq!( + unsafe { WaitForSingleObject(relay_thread.get(), 5000) }, + WAIT_OBJECT_0, + "the cancelled relay should exit" + ); + assert!( + !fired.load(Ordering::SeqCst), + "cancellation is not EOF: the callback must not run" + ); + } + /// it reads a duplicate. This is what lets `IsoSessionProcess::Close` run /// unconditionally. #[test] diff --git a/src/mxc-sdk/src/core/mxc_common/process_util.rs b/src/mxc-sdk/src/core/mxc_common/process_util.rs index 2567d10719..37854065ed 100644 --- a/src/mxc-sdk/src/core/mxc_common/process_util.rs +++ b/src/mxc-sdk/src/core/mxc_common/process_util.rs @@ -163,6 +163,13 @@ impl InterruptiblePipeReader { PipeReadCanceller(Arc::downgrade(&self.0)) } + /// Whether a [`PipeReadCanceller`] has already fired for this reader. A + /// cancelled read reports `Ok(0)` just like a genuine EOF, so a relay that + /// must act only on real end-of-input checks this to tell them apart. + pub fn was_cancelled(&self) -> bool { + self.0.lock().cancelled + } + pub fn try_clone_owned_handle(&self) -> std::io::Result { self.0.handle.try_clone_owned_handle() } From 6bfed368be3ff3444eb9ae748dc3358e9d280bd6 Mon Sep 17 00:00:00 2001 From: Dom Giandinoto Date: Tue, 13 Oct 2026 18:02:07 -0700 Subject: [PATCH 3/5] fmt change --- src/mxc-sdk/src/backends/isolation_session/common/manager.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs index 002b347aa3..2ed1ca03eb 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs @@ -607,7 +607,7 @@ impl IsolationSessionManager { scope.stdin = match stdin_relay_state { StdinRelayKind::None => None, StdinRelayKind::PipeOrFile => { - // Ensure the agent's stdin will close on EOF for pipe/file-backed + // Ensure the agent's stdin will close on EOF for pipe/file-backed // stdin. let process_for_eof = process.clone(); let impersonation = self.impersonation.clone(); From 11e19ce85c0f4641f029e61882b588194c7434a2 Mon Sep 17 00:00:00 2001 From: Dom Giandinoto Date: Mon, 2 Nov 2026 11:05:40 -0800 Subject: [PATCH 4/5] pre-simplification --- .../backends/isolation-session/oneshot.md | 21 ++ .../isolation-session/state-aware-rust.md | 12 + .../isolation_session/common/manager.rs | 271 +++++++++++++++--- 3 files changed, 260 insertions(+), 44 deletions(-) diff --git a/docs/development/architecture/backends/isolation-session/oneshot.md b/docs/development/architecture/backends/isolation-session/oneshot.md index 12ef61bb3b..ee4f74e92e 100644 --- a/docs/development/architecture/backends/isolation-session/oneshot.md +++ b/docs/development/architecture/backends/isolation-session/oneshot.md @@ -163,6 +163,27 @@ exit-wait and exit code, console resize, and the graceful-shutdown ladder (close stdin → send-ctrl-close → terminate). It sets the interactive-console flag when `wxc-exec`'s stdout is a TTY. +For pipe/file-backed host stdin, source EOF calls `CloseStandardInput` only +when the workload uses ordinary pipes. The OS Preview client must also release +its duplicate input handle for the workload to see EOF. With ConPTY, MXC ends +the input relay without closing the console input, since that can terminate a +still-running workload. A console workload waiting for more input may therefore +remain running until its execution timeout. + +After a confirmed timeout with the workload still running, the relay-managed +path signals/cancels stdin and waits up to one second for its relay to exit +**before** starting the shutdown sequence with `CloseStandardInput`. The relay +must release its duplicate write handle for ordinary-pipe EOF to be possible. +This hook does not run before the normal execution wait or for an already-exited +workload. An exit during cooperative shutdown is still reported as a timeout. +SDK streaming waits do not own this relay and retain their existing behavior. + +If the stdin relay does not exit within the join allowance, or its thread wait +fails, execution returns a cleanup error and the existing error cleanup requests +termination. It does not claim that the input writer was released or proceed +through the cooperative shutdown sequence. Read cancellation cannot stop a +blocked destination write or flush; bounded write cancellation is not added here. + ## Bindings Workflow **Why a private WinMD.** The OS-side API ships its WinMD as part of an diff --git a/docs/development/architecture/backends/isolation-session/state-aware-rust.md b/docs/development/architecture/backends/isolation-session/state-aware-rust.md index 3a9faf9f40..cdc5a0bca8 100644 --- a/docs/development/architecture/backends/isolation-session/state-aware-rust.md +++ b/docs/development/architecture/backends/isolation-session/state-aware-rust.md @@ -415,6 +415,18 @@ waiter, so the caller decides when to block. Either way, two concurrent exec calls against the same `sandboxId` are not coordinated by MXC; the OS-side service serialises (or rejects, depending on session state) at its own layer. +`Relayed` execution uses the same [stdin EOF behavior](oneshot.md#os-api-dependency) +as create-and-run execution: pipe/file input ending closes workload stdin only +without ConPTY. With ConPTY, the input relay ends without disconnecting the +console. SDK `Piped` and explicit PTY operations are unchanged. + +On a confirmed timeout of a still-running workload, `Relayed` execution stops +and joins its stdin relay before the shutdown sequence calls +`CloseStandardInput`. This releases MXC's write duplicate before asking the OS +to close its writers. The join is limited to one second; failure to confirm the +relay exited is a cleanup error and takes the existing termination-on-error +path. A cooperative exit after the deadline still produces a timeout result. + ### Deprovision and concurrent sandboxes `deprovision` removes only its own agent user (`deprovision_agent_user`). diff --git a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs index 2ed1ca03eb..83f4f31924 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs @@ -18,7 +18,7 @@ use crate::isolation_session_bindings::bindings::{ IsoSessionFeature, IsoSessionOps, IsoSessionProcess, IsoSessionProcessResult, IsoSessionUserResult, }; -use windows::Win32::Foundation::{HANDLE, WAIT_OBJECT_0}; +use windows::Win32::Foundation::{HANDLE, WAIT_OBJECT_0, WAIT_TIMEOUT}; use windows::Win32::Storage::FileSystem::{GetFileType, FILE_TYPE_CHAR}; use windows::Win32::System::Com::{CoDecrementMTAUsage, CoIncrementMTAUsage, CO_MTA_USAGE_COOKIE}; use windows::Win32::System::Console::{ @@ -64,6 +64,19 @@ fn classify_stdin_handle(handle: HANDLE) -> StdinHandleKind { StdinHandleKind::PipeOrFile } +fn pipe_stdin_eof_callback( + options: &ProcessOptions, + close_stdin: impl FnOnce() + Send + 'static, +) -> Option> { + // Closing ConPTY input can terminate the workload. Only ordinary pipes + // should forward source EOF through CloseStandardInput. + if options.interactive { + None + } else { + Some(Box::new(close_stdin)) + } +} + /// Keeps the process's MTA alive for as long as the lifecycle's WinRT objects, /// which outlive the call that created them. /// `CoIncrementMTAUsage` holds a reference releasable from any thread; @@ -607,11 +620,9 @@ impl IsolationSessionManager { scope.stdin = match stdin_relay_state { StdinRelayKind::None => None, StdinRelayKind::PipeOrFile => { - // Ensure the agent's stdin will close on EOF for pipe/file-backed - // stdin. let process_for_eof = process.clone(); let impersonation = self.impersonation.clone(); - let on_eof: Box = Box::new(move || { + let on_eof = pipe_stdin_eof_callback(options, move || { let _ = owned_thread::call(&impersonation, || { let _ = process_for_eof.CloseStandardInput(); Ok(()) @@ -621,7 +632,7 @@ impl IsolationSessionManager { create_relay_thread_with_eof( wxc_stdin, HANDLE(stdin_handle_val as *mut core::ffi::c_void), - Some(on_eof), + on_eof, ) } .map_err(|e| lifecycle_err(format!("create stdin relay: {}", e)))?; @@ -639,7 +650,9 @@ impl IsolationSessionManager { } }; - let outcome = started.wait(options.timeout_ms)?; + let outcome = started.wait_with_before_shutdown(options.timeout_ms, || { + scope.stop_stdin_before_shutdown() + })?; scope.stop_stdin(); scope.finish(); @@ -943,6 +956,14 @@ impl StartedProcess { /// walks all three tiers and cannot then tell that exit from a live /// process. An exited process therefore returns its code directly. pub(super) fn wait(&self, timeout_ms: u32) -> Result { + self.wait_with_before_shutdown(timeout_ms, || Ok(())) + } + + fn wait_with_before_shutdown( + &self, + timeout_ms: u32, + before_shutdown: impl FnOnce() -> Result<(), IsolationSessionError>, + ) -> Result { // `WaitForExit` is a Win32 `WaitForSingleObject` on a kernel handle — // no COM round-trip. On timeout it returns -1; the ladder below then // decides what to do about a process that is still running. @@ -961,48 +982,34 @@ impl StartedProcess { .map_err(|e| transport_err(op::RUN_PROCESS, "get ExitCode failed", &e)) })?; - // An exited process never reaches the ladder: the ladder cannot tell a - // 259 exit from a live process, which is the ambiguity `plan_wait` - // exists to resolve while the evidence is still intact. match plan { - WaitPlan::Exited(exit_code) => { - self.record_exited(exit_code); - return Ok(ExecOutcome::Exited(exit_code)); - } - // The deadline was spent and the process is already gone, so there - // is nothing to kill and nothing to confirm — the sample that - // produced this plan is the confirmation. - WaitPlan::TimedOutAfterDeadline => { - self.record_timed_out(timeout_ms); - return Ok(ExecOutcome::TimedOut); + WaitPlan::Exited(exit_code) => self.record_exited(exit_code), + WaitPlan::TimedOutAfterDeadline | WaitPlan::KillThenReportTimeout => { + self.record_timed_out(timeout_ms) } - WaitPlan::KillThenReportTimeout => {} } - // Timed out: the process was still running, so it must be dead before - // this reports `TimedOut`, which promises exactly that. - self.record_timed_out(timeout_ms); - let (_, terminate_error) = wait_with_graceful_shutdown(&self.process)?; - if let Some(error_code) = terminate_error { - self.record_kill_failed(error_code); - } - let after = self - .process - .ExitCode() - .map_err(|e| transport_err(op::RUN_PROCESS, "get ExitCode failed", &e))?; - if after == STILL_ACTIVE { - // Unlike the general case, 259 is not ambiguous here: the sample - // above established the process was running, so this is the same - // process still running rather than a stale exit code. (The one - // exception is a workload that exits with 259 inside the ladder's - // own window, which yields a conservative "could not determine" - // rather than a false claim that it was killed.) - return Err(lifecycle_err( - "the sandboxed process was still running after close-stdin, Ctrl-Close and \ - terminate; it timed out but could not be confirmed killed", - )); - } - Ok(ExecOutcome::TimedOut) + complete_wait(plan, before_shutdown, || { + // Confirm termination before reporting a timeout, even if closing + // stdin lets the workload exit cooperatively. + let (_, terminate_error) = wait_with_graceful_shutdown(&self.process)?; + if let Some(error_code) = terminate_error { + self.record_kill_failed(error_code); + } + let after = self + .process + .ExitCode() + .map_err(|e| transport_err(op::RUN_PROCESS, "get ExitCode failed", &e))?; + if after == STILL_ACTIVE { + // A 259 exit during shutdown remains ambiguous: conservatively + // report failure rather than claiming the process was killed. + return Err(lifecycle_err( + "the sandboxed process was still running after close-stdin, Ctrl-Close and \ + terminate; it timed out but could not be confirmed killed", + )); + } + Ok(()) + }) } /// Kill the process now, reporting whether the kill request was *accepted*. @@ -1157,6 +1164,12 @@ impl<'a> RelayScope<'a> { Self::join(&self.stdin, Self::JOIN_MS); } + fn stop_stdin_before_shutdown(&self) -> Result<(), IsolationSessionError> { + self.signal_stop(); + self.cancel_stdin_read(); + wait_for_stdin_relay(self.stdin.as_ref(), Self::JOIN_MS) + } + /// Teardown for a workload that has already exited. fn finish(&mut self) { self.signal_stop(); @@ -1366,6 +1379,46 @@ fn plan_wait( } } +/// Run timeout cleanup only for a confirmed live process past its deadline. +/// The relay must release its input writer before the shutdown ladder closes +/// the platform's writers; otherwise the first tier cannot deliver pipe EOF. +fn complete_wait( + plan: WaitPlan, + before_shutdown: impl FnOnce() -> Result<(), IsolationSessionError>, + shutdown: impl FnOnce() -> Result<(), IsolationSessionError>, +) -> Result { + match plan { + WaitPlan::Exited(code) => Ok(ExecOutcome::Exited(code)), + WaitPlan::TimedOutAfterDeadline => Ok(ExecOutcome::TimedOut), + WaitPlan::KillThenReportTimeout => { + before_shutdown()?; + shutdown()?; + Ok(ExecOutcome::TimedOut) + } + } +} + +fn wait_for_stdin_relay( + relay: Option<&OwnedHandle>, + timeout_ms: u32, +) -> Result<(), IsolationSessionError> { + let Some(relay) = relay else { + return Ok(()); + }; + // SAFETY: the relay's owned thread handle remains valid throughout the wait. + match unsafe { WaitForSingleObject(relay.get(), timeout_ms) } { + WAIT_OBJECT_0 => Ok(()), + WAIT_TIMEOUT => Err(lifecycle_err( + "stdin relay did not exit before timeout shutdown; its input write handle may \ + still be open (a blocked write or flush cannot be cancelled by read cancellation)", + )), + _ => Err(lifecycle_err(format!( + "waiting for stdin relay before timeout shutdown failed: {}", + std::io::Error::last_os_error() + ))), + } +} + /// Three-tier graceful shutdown for an `IsoSessionProcess` that's still /// running after `WaitForExit(timeout_ms)` returns. Tier 1 asks the platform /// to close stdin, tier 2 tries `SendCtrlClose` (ConPTY-only; `E_NOTIMPL` @@ -1423,6 +1476,136 @@ fn wait_with_graceful_shutdown( mod tests { use super::*; + #[test] + fn stdin_shutdown_releases_relay_writer_before_closing_platform_input() { + use crate::mxc_common::process_util::create_local_pipe; + use windows::Win32::Foundation::ERROR_BROKEN_PIPE; + use windows::Win32::System::Pipes::PeekNamedPipe; + + let (source_read, _source_write) = create_local_pipe().unwrap(); + let (dest_read, dest_write) = create_local_pipe().unwrap(); + // SAFETY: the relay duplicates both live pipe handles. + let (relay, canceller) = + unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; + let plan = plan_wait(5000, WAIT_FOR_EXIT_TIMEOUT, probe(STILL_ACTIVE)).unwrap(); + assert_eq!(unsafe { WaitForSingleObject(relay.get(), 0) }, WAIT_TIMEOUT); + + let outcome = complete_wait( + plan, + || { + canceller.close(); + wait_for_stdin_relay(Some(&relay), 5000) + }, + || { + // Stand in for CloseStandardInput releasing the platform writer. + // The caller's source writer is still open throughout shutdown. + drop(dest_write); + let error = unsafe { PeekNamedPipe(dest_read.get(), None, 0, None, None, None) } + .expect_err("the relay must release its duplicate before shutdown"); + assert_eq!(error.code(), ERROR_BROKEN_PIPE.to_hresult()); + Ok(()) + }, + ) + .unwrap(); + assert_eq!(outcome, ExecOutcome::TimedOut); + } + + #[test] + fn stdin_shutdown_does_not_run_for_completed_waits() { + for (timeout_ms, waited, exit_code, expected) in [ + (5000, 7, 7, ExecOutcome::Exited(7)), + (5000, -1, -1, ExecOutcome::Exited(-1)), + ( + 5000, + STILL_ACTIVE, + STILL_ACTIVE, + ExecOutcome::Exited(STILL_ACTIVE), + ), + (0, -1, STILL_ACTIVE, ExecOutcome::Exited(-1)), + (5000, -1, 0, ExecOutcome::TimedOut), + ] { + let plan = plan_wait(timeout_ms, waited, probe(exit_code)).unwrap(); + let outcome = complete_wait( + plan, + || panic!("a completed wait must not prepare timeout shutdown"), + || panic!("a completed wait must not escalate shutdown"), + ) + .unwrap(); + assert_eq!(outcome, expected); + } + } + + #[test] + fn stdin_shutdown_reports_unreleased_relay_instead_of_escalating() { + use crate::mxc_common::process_util::create_local_pipe; + + let (source_read, _source_write) = create_local_pipe().unwrap(); + let (_dest_read, dest_write) = create_local_pipe().unwrap(); + // SAFETY: the relay duplicates both live pipe handles. + let (relay, canceller) = + unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; + let result = complete_wait( + WaitPlan::KillThenReportTimeout, + || wait_for_stdin_relay(Some(&relay), 0), + || panic!("shutdown must not proceed with an unconfirmed relay exit"), + ); + canceller.close(); + wait_for_stdin_relay(Some(&relay), 5000).unwrap(); + assert!(result + .unwrap_err() + .to_string() + .contains("stdin relay did not exit")); + } + + #[test] + fn stdin_shutdown_reports_wait_failure() { + let invalid = OwnedHandle::new(HANDLE::default()); + let error = wait_for_stdin_relay(Some(&invalid), 0).unwrap_err(); + assert!(error.to_string().contains("waiting for stdin relay")); + assert!(wait_for_stdin_relay(None, 0).is_ok()); + } + + #[test] + fn stdin_shutdown_propagates_shutdown_failure_after_preparation() { + let prepared = std::cell::Cell::new(false); + let result = complete_wait( + WaitPlan::KillThenReportTimeout, + || { + prepared.set(true); + Ok(()) + }, + || { + assert!(prepared.get()); + Err(lifecycle_err("shutdown failed")) + }, + ); + assert!(result.unwrap_err().to_string().contains("shutdown failed")); + } + + #[test] + fn pipe_stdin_eof_closes_only_non_conpty_input() { + use super::super::process_options::build_process_options; + use crate::mxc_common::models::ExecutionRequest; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + + for interactive in [false, true] { + let options = build_process_options(&ExecutionRequest::default(), interactive); + let closed = Arc::new(AtomicBool::new(false)); + let closed_by_callback = Arc::clone(&closed); + let on_eof = pipe_stdin_eof_callback(&options, move || { + closed_by_callback.store(true, Ordering::SeqCst); + }); + + assert_eq!(on_eof.is_some(), !interactive); + assert!(!closed.load(Ordering::SeqCst)); + if let Some(on_eof) = on_eof { + on_eof(); + } + assert_eq!(closed.load(Ordering::SeqCst), !interactive); + } + } + #[test] fn bounded_termination_requires_confirmation_after_a_timeout_sentinel() { assert!(termination_is_confirmed(0, None)); From edd89eebaa1ca5220ce5fb8f0c9c5cde057da277 Mon Sep 17 00:00:00 2001 From: Dom Giandinoto Date: Mon, 2 Nov 2026 11:24:48 -0800 Subject: [PATCH 5/5] post-simplification --- .../backends/isolation-session/oneshot.md | 27 +- .../isolation-session/state-aware-rust.md | 15 +- .../isolation_session/common/manager.rs | 301 +++++------------- .../isolation_session/common/pipe_relay.rs | 120 +------ .../src/core/mxc_common/process_util.rs | 7 - 5 files changed, 96 insertions(+), 374 deletions(-) diff --git a/docs/development/architecture/backends/isolation-session/oneshot.md b/docs/development/architecture/backends/isolation-session/oneshot.md index ee4f74e92e..37cb1ba15a 100644 --- a/docs/development/architecture/backends/isolation-session/oneshot.md +++ b/docs/development/architecture/backends/isolation-session/oneshot.md @@ -163,26 +163,13 @@ exit-wait and exit code, console resize, and the graceful-shutdown ladder (close stdin → send-ctrl-close → terminate). It sets the interactive-console flag when `wxc-exec`'s stdout is a TTY. -For pipe/file-backed host stdin, source EOF calls `CloseStandardInput` only -when the workload uses ordinary pipes. The OS Preview client must also release -its duplicate input handle for the workload to see EOF. With ConPTY, MXC ends -the input relay without closing the console input, since that can terminate a -still-running workload. A console workload waiting for more input may therefore -remain running until its execution timeout. - -After a confirmed timeout with the workload still running, the relay-managed -path signals/cancels stdin and waits up to one second for its relay to exit -**before** starting the shutdown sequence with `CloseStandardInput`. The relay -must release its duplicate write handle for ordinary-pipe EOF to be possible. -This hook does not run before the normal execution wait or for an already-exited -workload. An exit during cooperative shutdown is still reported as a timeout. -SDK streaming waits do not own this relay and retain their existing behavior. - -If the stdin relay does not exit within the join allowance, or its thread wait -fails, execution returns a cleanup error and the existing error cleanup requests -termination. It does not claim that the input writer was released or proceed -through the cooperative shutdown sequence. Read cancellation cannot stop a -blocked destination write or flush; bounded write cancellation is not added here. +On timeout, MXC stops its input-copying thread and waits up to one second for +it to release its pipe handle before calling `CloseStandardInput`. If the thread +is still running, MXC skips that step and continues with console-close and +termination, still reporting a timeout once the workload is confirmed stopped. + +A failed Windows thread wait is an error, unlike a wait whose time limit expires. +The OS client must also close its input handle for the workload to receive EOF. ## Bindings Workflow diff --git a/docs/development/architecture/backends/isolation-session/state-aware-rust.md b/docs/development/architecture/backends/isolation-session/state-aware-rust.md index cdc5a0bca8..2891860838 100644 --- a/docs/development/architecture/backends/isolation-session/state-aware-rust.md +++ b/docs/development/architecture/backends/isolation-session/state-aware-rust.md @@ -415,17 +415,10 @@ waiter, so the caller decides when to block. Either way, two concurrent exec calls against the same `sandboxId` are not coordinated by MXC; the OS-side service serialises (or rejects, depending on session state) at its own layer. -`Relayed` execution uses the same [stdin EOF behavior](oneshot.md#os-api-dependency) -as create-and-run execution: pipe/file input ending closes workload stdin only -without ConPTY. With ConPTY, the input relay ends without disconnecting the -console. SDK `Piped` and explicit PTY operations are unchanged. - -On a confirmed timeout of a still-running workload, `Relayed` execution stops -and joins its stdin relay before the shutdown sequence calls -`CloseStandardInput`. This releases MXC's write duplicate before asking the OS -to close its writers. The join is limited to one second; failure to confirm the -relay exited is a cleanup error and takes the existing termination-on-error -path. A cooperative exit after the deadline still produces a timeout result. +`Relayed` execution uses the same [timeout cleanup](oneshot.md#os-api-dependency) +as create-and-run execution: stop input copying before closing stdin, or skip +stdin-close if the thread has not exited within one second. SDK `Piped` and +explicit PTY operations keep their existing shutdown behavior. ### Deprovision and concurrent sandboxes diff --git a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs index 83f4f31924..1c8261f2dc 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/manager.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/manager.rs @@ -37,8 +37,7 @@ use super::error::{ }; use super::owned_thread::{self, Impersonation}; use super::pipe_relay::{ - create_relay_thread, create_relay_thread_with_eof, create_relay_thread_with_stop, - duplicate_handle, PipeRelayWithStopParams, + create_relay_thread, create_relay_thread_with_stop, duplicate_handle, PipeRelayWithStopParams, }; use super::process_options::{ build_iso_process_options, with_service_timeout_grace, ProcessOptions, @@ -64,19 +63,6 @@ fn classify_stdin_handle(handle: HANDLE) -> StdinHandleKind { StdinHandleKind::PipeOrFile } -fn pipe_stdin_eof_callback( - options: &ProcessOptions, - close_stdin: impl FnOnce() + Send + 'static, -) -> Option> { - // Closing ConPTY input can terminate the workload. Only ordinary pipes - // should forward source EOF through CloseStandardInput. - if options.interactive { - None - } else { - Some(Box::new(close_stdin)) - } -} - /// Keeps the process's MTA alive for as long as the lifecycle's WinRT objects, /// which outlive the call that created them. /// `CoIncrementMTAUsage` holds a reference releasable from any thread; @@ -620,19 +606,10 @@ impl IsolationSessionManager { scope.stdin = match stdin_relay_state { StdinRelayKind::None => None, StdinRelayKind::PipeOrFile => { - let process_for_eof = process.clone(); - let impersonation = self.impersonation.clone(); - let on_eof = pipe_stdin_eof_callback(options, move || { - let _ = owned_thread::call(&impersonation, || { - let _ = process_for_eof.CloseStandardInput(); - Ok(()) - }); - }); let (thread, canceller) = unsafe { - create_relay_thread_with_eof( + create_relay_thread( wxc_stdin, HANDLE(stdin_handle_val as *mut core::ffi::c_void), - on_eof, ) } .map_err(|e| lifecycle_err(format!("create stdin relay: {}", e)))?; @@ -650,10 +627,7 @@ impl IsolationSessionManager { } }; - let outcome = started.wait_with_before_shutdown(options.timeout_ms, || { - scope.stop_stdin_before_shutdown() - })?; - scope.stop_stdin(); + let outcome = started.wait(options.timeout_ms, || scope.stop_stdin())?; scope.finish(); Ok(outcome) @@ -715,7 +689,7 @@ impl IsolationSessionManager { // before the shutdown ladder destroys the evidence. waiter: Box::new(move || { owned_thread::call(&waiter_process.impersonation, || { - waiter_process.wait(timeout_ms) + waiter_process.wait(timeout_ms, || Ok(true)) }) .map_err(super::error::map_lifecycle_error) }), @@ -955,14 +929,13 @@ impl StartedProcess { /// first `ExitCode()` reads 259 for a workload that exited with 259, so it /// walks all three tiers and cannot then tell that exit from a live /// process. An exited process therefore returns its code directly. - pub(super) fn wait(&self, timeout_ms: u32) -> Result { - self.wait_with_before_shutdown(timeout_ms, || Ok(())) - } - - fn wait_with_before_shutdown( + /// + /// `before_shutdown` stops caller-owned input copying after a timeout. + /// Returning false skips the stdin-close step because a writer remains open. + pub(super) fn wait( &self, timeout_ms: u32, - before_shutdown: impl FnOnce() -> Result<(), IsolationSessionError>, + before_shutdown: impl FnOnce() -> Result, ) -> Result { // `WaitForExit` is a Win32 `WaitForSingleObject` on a kernel handle — // no COM round-trip. On timeout it returns -1; the ladder below then @@ -983,33 +956,40 @@ impl StartedProcess { })?; match plan { - WaitPlan::Exited(exit_code) => self.record_exited(exit_code), - WaitPlan::TimedOutAfterDeadline | WaitPlan::KillThenReportTimeout => { - self.record_timed_out(timeout_ms) + // An exited process must skip shutdown: the ladder cannot + // distinguish a real exit code of 259 from STILL_ACTIVE. + WaitPlan::Exited(exit_code) => { + self.record_exited(exit_code); + return Ok(ExecOutcome::Exited(exit_code)); } + // The deadline was spent, but the process is already gone. + // Nothing needs killing; retain the timeout result. + WaitPlan::TimedOutAfterDeadline => { + self.record_timed_out(timeout_ms); + return Ok(ExecOutcome::TimedOut); + } + WaitPlan::KillThenReportTimeout => {} } - complete_wait(plan, before_shutdown, || { - // Confirm termination before reporting a timeout, even if closing - // stdin lets the workload exit cooperatively. - let (_, terminate_error) = wait_with_graceful_shutdown(&self.process)?; - if let Some(error_code) = terminate_error { - self.record_kill_failed(error_code); - } - let after = self - .process - .ExitCode() - .map_err(|e| transport_err(op::RUN_PROCESS, "get ExitCode failed", &e))?; - if after == STILL_ACTIVE { - // A 259 exit during shutdown remains ambiguous: conservatively - // report failure rather than claiming the process was killed. - return Err(lifecycle_err( - "the sandboxed process was still running after close-stdin, Ctrl-Close and \ - terminate; it timed out but could not be confirmed killed", - )); - } - Ok(()) - }) + self.record_timed_out(timeout_ms); + let stdin_released = before_shutdown()?; + let (_, terminate_error) = wait_with_graceful_shutdown(&self.process, stdin_released)?; + if let Some(error_code) = terminate_error { + self.record_kill_failed(error_code); + } + let after = self + .process + .ExitCode() + .map_err(|e| transport_err(op::RUN_PROCESS, "get ExitCode failed", &e))?; + if after == STILL_ACTIVE { + // A 259 exit during shutdown remains ambiguous: conservatively + // report failure rather than claiming the process was killed. + return Err(lifecycle_err( + "the sandboxed process was still running after timeout shutdown; \ + it timed out but could not be confirmed killed", + )); + } + Ok(ExecOutcome::TimedOut) } /// Kill the process now, reporting whether the kill request was *accepted*. @@ -1145,29 +1125,23 @@ impl<'a> RelayScope<'a> { /// Whether the relay exited. An absent relay was never spawned. fn join(relay: &Option, timeout_ms: u32) -> bool { - match relay { - // SAFETY: an owned thread handle held by this scope. - Some(t) => (unsafe { WaitForSingleObject(t.get(), timeout_ms) }) == WAIT_OBJECT_0, - None => true, - } + wait_for_relay(relay.as_ref(), timeout_ms).unwrap_or(false) } - /// Each relay reads its own duplicate, so one still running does not keep - /// the pipe handles alive. + /// A still-running relay owns its duplicates until it exits. fn close_handles(&self) { let _ = self.process.Close(); } - fn stop_stdin(&self) { - self.signal_stop(); - self.cancel_stdin_read(); - Self::join(&self.stdin, Self::JOIN_MS); - } - - fn stop_stdin_before_shutdown(&self) -> Result<(), IsolationSessionError> { + fn stop_stdin(&mut self) -> Result { self.signal_stop(); self.cancel_stdin_read(); - wait_for_stdin_relay(self.stdin.as_ref(), Self::JOIN_MS) + let exited = wait_for_relay(self.stdin.as_ref(), Self::JOIN_MS)?; + if exited { + self.stdin = None; + self.stdin_canceller = None; + } + Ok(exited) } /// Teardown for a workload that has already exited. @@ -1260,7 +1234,7 @@ impl ClosingProcess { &self, timeout_ms: u32, ) -> Result { - owned_thread::call(&self.impersonation, || self.wait(timeout_ms)) + owned_thread::call(&self.impersonation, || self.wait(timeout_ms, || Ok(true))) } pub(super) fn terminate_process(&self) -> Result<(), IsolationSessionError> { @@ -1379,41 +1353,19 @@ fn plan_wait( } } -/// Run timeout cleanup only for a confirmed live process past its deadline. -/// The relay must release its input writer before the shutdown ladder closes -/// the platform's writers; otherwise the first tier cannot deliver pipe EOF. -fn complete_wait( - plan: WaitPlan, - before_shutdown: impl FnOnce() -> Result<(), IsolationSessionError>, - shutdown: impl FnOnce() -> Result<(), IsolationSessionError>, -) -> Result { - match plan { - WaitPlan::Exited(code) => Ok(ExecOutcome::Exited(code)), - WaitPlan::TimedOutAfterDeadline => Ok(ExecOutcome::TimedOut), - WaitPlan::KillThenReportTimeout => { - before_shutdown()?; - shutdown()?; - Ok(ExecOutcome::TimedOut) - } - } -} - -fn wait_for_stdin_relay( +fn wait_for_relay( relay: Option<&OwnedHandle>, timeout_ms: u32, -) -> Result<(), IsolationSessionError> { +) -> Result { let Some(relay) = relay else { - return Ok(()); + return Ok(true); }; // SAFETY: the relay's owned thread handle remains valid throughout the wait. match unsafe { WaitForSingleObject(relay.get(), timeout_ms) } { - WAIT_OBJECT_0 => Ok(()), - WAIT_TIMEOUT => Err(lifecycle_err( - "stdin relay did not exit before timeout shutdown; its input write handle may \ - still be open (a blocked write or flush cannot be cancelled by read cancellation)", - )), + WAIT_OBJECT_0 => Ok(true), + WAIT_TIMEOUT => Ok(false), _ => Err(lifecycle_err(format!( - "waiting for stdin relay before timeout shutdown failed: {}", + "waiting for relay failed: {}", std::io::Error::last_os_error() ))), } @@ -1424,6 +1376,7 @@ fn wait_for_stdin_relay( /// to close stdin, tier 2 tries `SendCtrlClose` (ConPTY-only; `E_NOTIMPL` /// outside ConPTY is benign), and tier 3 force-terminates, then waits /// infinitely (`WaitForExit(0)` = INFINITE) for the kill to land. +/// Tier 1 is skipped if the caller's stdin relay has not exited. /// /// The first `ExitCode()` query is `?`-propagated: a failure there means /// the kernel handle is broken, and the cleanup methods on the same @@ -1441,6 +1394,7 @@ fn wait_for_stdin_relay( /// independently determined the process was still running. fn wait_with_graceful_shutdown( process: &IsoSessionProcess, + stdin_released: bool, ) -> Result<(i32, Option), IsolationSessionError> { let mut exit_code = process .ExitCode() @@ -1449,11 +1403,15 @@ fn wait_with_graceful_shutdown( return Ok((exit_code, None)); } - let _ = process.CloseStandardInput(); - let _ = process.WaitForExit(5000); - exit_code = process.ExitCode().unwrap_or(STILL_ACTIVE); - if exit_code != STILL_ACTIVE { - return Ok((exit_code, None)); + // A live relay can still hold a writer, so closing platform stdin alone + // cannot deliver EOF. Skip that grace period if the relay did not stop. + if stdin_released { + let _ = process.CloseStandardInput(); + let _ = process.WaitForExit(5000); + exit_code = process.ExitCode().unwrap_or(STILL_ACTIVE); + if exit_code != STILL_ACTIVE { + return Ok((exit_code, None)); + } } let _ = process.SendCtrlClose(); @@ -1477,7 +1435,7 @@ mod tests { use super::*; #[test] - fn stdin_shutdown_releases_relay_writer_before_closing_platform_input() { + fn stdin_shutdown_join_distinguishes_running_and_released_relay() { use crate::mxc_common::process_util::create_local_pipe; use windows::Win32::Foundation::ERROR_BROKEN_PIPE; use windows::Win32::System::Pipes::PeekNamedPipe; @@ -1487,123 +1445,24 @@ mod tests { // SAFETY: the relay duplicates both live pipe handles. let (relay, canceller) = unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; - let plan = plan_wait(5000, WAIT_FOR_EXIT_TIMEOUT, probe(STILL_ACTIVE)).unwrap(); - assert_eq!(unsafe { WaitForSingleObject(relay.get(), 0) }, WAIT_TIMEOUT); - - let outcome = complete_wait( - plan, - || { - canceller.close(); - wait_for_stdin_relay(Some(&relay), 5000) - }, - || { - // Stand in for CloseStandardInput releasing the platform writer. - // The caller's source writer is still open throughout shutdown. - drop(dest_write); - let error = unsafe { PeekNamedPipe(dest_read.get(), None, 0, None, None, None) } - .expect_err("the relay must release its duplicate before shutdown"); - assert_eq!(error.code(), ERROR_BROKEN_PIPE.to_hresult()); - Ok(()) - }, - ) - .unwrap(); - assert_eq!(outcome, ExecOutcome::TimedOut); - } - - #[test] - fn stdin_shutdown_does_not_run_for_completed_waits() { - for (timeout_ms, waited, exit_code, expected) in [ - (5000, 7, 7, ExecOutcome::Exited(7)), - (5000, -1, -1, ExecOutcome::Exited(-1)), - ( - 5000, - STILL_ACTIVE, - STILL_ACTIVE, - ExecOutcome::Exited(STILL_ACTIVE), - ), - (0, -1, STILL_ACTIVE, ExecOutcome::Exited(-1)), - (5000, -1, 0, ExecOutcome::TimedOut), - ] { - let plan = plan_wait(timeout_ms, waited, probe(exit_code)).unwrap(); - let outcome = complete_wait( - plan, - || panic!("a completed wait must not prepare timeout shutdown"), - || panic!("a completed wait must not escalate shutdown"), - ) - .unwrap(); - assert_eq!(outcome, expected); - } - } - - #[test] - fn stdin_shutdown_reports_unreleased_relay_instead_of_escalating() { - use crate::mxc_common::process_util::create_local_pipe; - - let (source_read, _source_write) = create_local_pipe().unwrap(); - let (_dest_read, dest_write) = create_local_pipe().unwrap(); - // SAFETY: the relay duplicates both live pipe handles. - let (relay, canceller) = - unsafe { create_relay_thread(source_read.get(), dest_write.get()).unwrap() }; - let result = complete_wait( - WaitPlan::KillThenReportTimeout, - || wait_for_stdin_relay(Some(&relay), 0), - || panic!("shutdown must not proceed with an unconfirmed relay exit"), - ); + assert!(!wait_for_relay(Some(&relay), 0).unwrap()); canceller.close(); - wait_for_stdin_relay(Some(&relay), 5000).unwrap(); - assert!(result - .unwrap_err() - .to_string() - .contains("stdin relay did not exit")); + assert!(wait_for_relay(Some(&relay), 5000).unwrap()); + + // Stand in for CloseStandardInput releasing the platform writer. + // The caller's source writer is still open throughout shutdown. + drop(dest_write); + let error = unsafe { PeekNamedPipe(dest_read.get(), None, 0, None, None, None) } + .expect_err("the relay must release its duplicate before shutdown"); + assert_eq!(error.code(), ERROR_BROKEN_PIPE.to_hresult()); } #[test] fn stdin_shutdown_reports_wait_failure() { let invalid = OwnedHandle::new(HANDLE::default()); - let error = wait_for_stdin_relay(Some(&invalid), 0).unwrap_err(); - assert!(error.to_string().contains("waiting for stdin relay")); - assert!(wait_for_stdin_relay(None, 0).is_ok()); - } - - #[test] - fn stdin_shutdown_propagates_shutdown_failure_after_preparation() { - let prepared = std::cell::Cell::new(false); - let result = complete_wait( - WaitPlan::KillThenReportTimeout, - || { - prepared.set(true); - Ok(()) - }, - || { - assert!(prepared.get()); - Err(lifecycle_err("shutdown failed")) - }, - ); - assert!(result.unwrap_err().to_string().contains("shutdown failed")); - } - - #[test] - fn pipe_stdin_eof_closes_only_non_conpty_input() { - use super::super::process_options::build_process_options; - use crate::mxc_common::models::ExecutionRequest; - use std::sync::atomic::{AtomicBool, Ordering}; - use std::sync::Arc; - - for interactive in [false, true] { - let options = build_process_options(&ExecutionRequest::default(), interactive); - let closed = Arc::new(AtomicBool::new(false)); - let closed_by_callback = Arc::clone(&closed); - let on_eof = pipe_stdin_eof_callback(&options, move || { - closed_by_callback.store(true, Ordering::SeqCst); - }); - - assert_eq!(on_eof.is_some(), !interactive); - assert!(!closed.load(Ordering::SeqCst)); - if let Some(on_eof) = on_eof { - on_eof(); - } - assert_eq!(closed.load(Ordering::SeqCst), !interactive); - } + let error = wait_for_relay(Some(&invalid), 0).unwrap_err(); + assert!(error.to_string().contains("waiting for relay failed")); + assert!(wait_for_relay(None, 0).unwrap()); } #[test] diff --git a/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs b/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs index c4c8d7d888..fcc18de347 100644 --- a/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs +++ b/src/mxc-sdk/src/backends/isolation_session/common/pipe_relay.rs @@ -52,7 +52,6 @@ pub(super) fn duplicate_handle(handle: HANDLE) -> Result struct PipeRelayParams { reader: InterruptiblePipeReader, h_write: OwnedHandle, - on_eof: Option>, } /// Thread procedure for relaying data between two handles. @@ -67,13 +66,9 @@ unsafe extern "system" fn pipe_relay_thread_proc(param: *mut core::ffi::c_void) let mut params = unsafe { Box::from_raw(param as *mut PipeRelayParams) }; let mut buffer = [0u8; BUFFER_SIZE as usize]; - let reached_source_eof = loop { + loop { let bytes_read = match params.reader.read(&mut buffer) { - // `Ok(0)` is reported for a genuine source EOF *and* for a fired - // canceller, so disambiguate with `was_cancelled` to determine if - // this is a genuine EOF. - Ok(0) => break !params.reader.was_cancelled(), - Err(_) => break false, + Ok(0) | Err(_) => break, Ok(n) => n, }; @@ -87,16 +82,10 @@ unsafe extern "system" fn pipe_relay_thread_proc(param: *mut core::ffi::c_void) .is_err() || bytes_written as usize != bytes_read { - break false; + break; } let _ = FlushFileBuffers(params.h_write.get()); - }; - - if reached_source_eof { - if let Some(on_eof) = params.on_eof.take() { - on_eof(); - } } 0 @@ -112,26 +101,12 @@ unsafe extern "system" fn pipe_relay_thread_proc(param: *mut core::ffi::c_void) pub(super) unsafe fn create_relay_thread( h_read: HANDLE, h_write: HANDLE, -) -> Result<(OwnedHandle, PipeReadCanceller), WxcError> { - unsafe { create_relay_thread_with_eof(h_read, h_write, None) } -} - -/// As [`create_relay_thread`], but runs `on_eof` once if the relay ends because -/// the source reached a genuine, uncancelled EOF. Pipe/file-backed stdin uses -/// this to propagate end-of-input to the agent (`CloseStandardInput`); a fired -/// canceller or a write error does **not** run it, so the forced-teardown path -/// stays distinct from natural end-of-input. -pub(super) unsafe fn create_relay_thread_with_eof( - h_read: HANDLE, - h_write: HANDLE, - on_eof: Option>, ) -> Result<(OwnedHandle, PipeReadCanceller), WxcError> { let reader = InterruptiblePipeReader::new(duplicate_handle(h_read)?); let canceller = reader.canceller(); let raw = Box::into_raw(Box::new(PipeRelayParams { reader, h_write: duplicate_handle(h_write)?, - on_eof, })); match unsafe { CreateThread( @@ -173,7 +148,7 @@ pub(super) unsafe fn create_relay_thread_with_eof( // also work). Anonymous pipe handles are NOT supported: they appear "always // signalled when open", so the wait returns immediately, the relay enters // `ReadFile`, and from there the stop event cannot interrupt it. For -// pipe-backed stdin (the non-TTY case), use the simpler EOF-driven +// pipe-backed stdin, use the simpler EOF-driven // `create_relay_thread` and rely on natural EOF or its canceller for cleanup. /// Parameters for a stop-event-aware relay thread. The thread loops @@ -465,92 +440,7 @@ mod tests { assert_eq!(read_from_pipe(dest_send.get()), ""); } - /// A genuine, uncancelled source EOF must run the relay's `on_eof` hook — - /// this is what closes the agent's stdin so a read-to-EOF workload exits. - #[test] - fn a_pipe_relay_at_source_eof_runs_its_eof_callback() { - use std::sync::atomic::{AtomicBool, Ordering}; - use std::sync::Arc; - - let (source_read, source_write) = create_test_pipes(); - let (mut dest_read, dest_write) = create_test_pipes(); - - let fired = Arc::new(AtomicBool::new(false)); - let fired_cb = Arc::clone(&fired); - let on_eof: Box = - Box::new(move || fired_cb.store(true, Ordering::SeqCst)); - - let (relay_thread, _canceller) = unsafe { - create_relay_thread_with_eof(source_read.get(), dest_write.get(), Some(on_eof)).unwrap() - }; - - // Drain concurrently so the relay's post-write FlushFileBuffers returns. - let dest_send = SendOwnedHandle::take(&mut dest_read); - let reader = std::thread::spawn(move || read_from_pipe(dest_send.get())); - - let mut written = 0u32; - unsafe { - WriteFile( - source_write.get(), - Some(b"payload"), - Some(&mut written), - None, - ) - .unwrap(); - } - // Source EOF: the relay copies the payload, reads `Ok(0)`, runs on_eof. - drop(source_write); - - assert_eq!( - unsafe { WaitForSingleObject(relay_thread.get(), 5000) }, - WAIT_OBJECT_0, - "the relay should reach EOF and exit" - ); - // The hook runs inside the thread proc before it returns, so by the time - // the thread is signalled the flag is already set. - assert!( - fired.load(Ordering::SeqCst), - "a genuine source EOF must run the EOF callback" - ); - - drop(dest_write); - assert_eq!(reader.join().unwrap(), "payload"); - } - - /// A cancellation also surfaces as `Ok(0)`, but it is forced teardown, not - /// end-of-input — the `on_eof` hook must not run (otherwise shutdown would - /// spuriously close the agent's stdin). - #[test] - fn a_cancelled_pipe_relay_does_not_run_its_eof_callback() { - use std::sync::atomic::{AtomicBool, Ordering}; - use std::sync::Arc; - - let (source_read, _source_write) = create_test_pipes(); - let (_dest_read, dest_write) = create_test_pipes(); - - let fired = Arc::new(AtomicBool::new(false)); - let fired_cb = Arc::clone(&fired); - let on_eof: Box = - Box::new(move || fired_cb.store(true, Ordering::SeqCst)); - - let (relay_thread, canceller) = unsafe { - create_relay_thread_with_eof(source_read.get(), dest_write.get(), Some(on_eof)).unwrap() - }; - - // Parked in `ReadFile` (`_source_write` keeps it from reaching EOF). - canceller.close(); - - assert_eq!( - unsafe { WaitForSingleObject(relay_thread.get(), 5000) }, - WAIT_OBJECT_0, - "the cancelled relay should exit" - ); - assert!( - !fired.load(Ordering::SeqCst), - "cancellation is not EOF: the callback must not run" - ); - } - + /// Closing the source handle the relay was built from does not disturb it: /// it reads a duplicate. This is what lets `IsoSessionProcess::Close` run /// unconditionally. #[test] diff --git a/src/mxc-sdk/src/core/mxc_common/process_util.rs b/src/mxc-sdk/src/core/mxc_common/process_util.rs index 37854065ed..2567d10719 100644 --- a/src/mxc-sdk/src/core/mxc_common/process_util.rs +++ b/src/mxc-sdk/src/core/mxc_common/process_util.rs @@ -163,13 +163,6 @@ impl InterruptiblePipeReader { PipeReadCanceller(Arc::downgrade(&self.0)) } - /// Whether a [`PipeReadCanceller`] has already fired for this reader. A - /// cancelled read reports `Ok(0)` just like a genuine EOF, so a relay that - /// must act only on real end-of-input checks this to tell them apart. - pub fn was_cancelled(&self) -> bool { - self.0.lock().cancelled - } - pub fn try_clone_owned_handle(&self) -> std::io::Result { self.0.handle.try_clone_owned_handle() }