From 967a336a71f667b3229ce21441e1c6580542a761 Mon Sep 17 00:00:00 2001 From: Soham Das Date: Mon, 10 Aug 2026 17:10:56 -0700 Subject: [PATCH 1/7] WSLC state-aware: live exec output streaming + fs-delegation gate --- .../wslc/common/src/container_steps.rs | 52 +++- src/backends/wslc/common/src/daemon_client.rs | 54 +++-- src/backends/wslc/common/src/state_aware.rs | 107 +++++--- .../wslc/daemon/src/control_server.rs | 228 ++++++++++++++++-- .../wslc/daemon/src/session_manager.rs | 98 ++++++-- tests/configs/wslc_state_aware_exec_drip.json | 9 + tests/scripts/run_wslc_state_aware_tests.ps1 | 136 ++++++++--- 7 files changed, 565 insertions(+), 119 deletions(-) create mode 100644 tests/configs/wslc_state_aware_exec_drip.json diff --git a/src/backends/wslc/common/src/container_steps.rs b/src/backends/wslc/common/src/container_steps.rs index 2392c15ec..e4c703ff6 100644 --- a/src/backends/wslc/common/src/container_steps.rs +++ b/src/backends/wslc/common/src/container_steps.rs @@ -109,6 +109,20 @@ fn cstr_bytes(field: &str, value: &str) -> Result, ScriptResponse> { // after adoption ownership passes to `exit_callback`; the kill path trades a // bounded, one-time leak for memory safety. +/// Which standard stream a live-output chunk came from. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum OutStream { + Stdout, + Stderr, +} + +/// Optional live-output sink invoked from the SDK's stdout/stderr callbacks in +/// addition to the capped capture buffers. The daemon supplies one to stream a +/// container's output to the client as bytes arrive; paths that only need the +/// final captured blob (one-shot, detached init) leave it unset. It receives +/// the full callback bytes, independent of the capped buffers' truncation. +pub type OutputSink = Box; + /// Shared buffer for capturing process I/O via SDK callbacks. Fields are /// `pub(crate)` so the one-shot runner's wait/collect helpers can read the /// captured bytes and exit signal. @@ -116,6 +130,11 @@ pub struct IoContext { pub(crate) stdout: Arc>>, pub(crate) stderr: Arc>>, pub(crate) exited: Arc<(Mutex, Condvar)>, + /// Live sink for streaming output alongside the capped buffers; `None` when + /// only the final captured blob is needed. Its owning `Arc` is + /// released on the same schedule as the capture buffers, so on the + /// deliberate kill-path leak the sink (and its sender) is leaked too. + sink: Option, } /// Per-stream cap on captured stdout/stderr, in bytes. The WSLc SDK streams @@ -166,12 +185,22 @@ unsafe extern "C" fn io_callback( let bytes = std::slice::from_raw_parts(data, data_size as usize); match io_handle { WslcProcessIOHandle::WSLC_PROCESS_IO_HANDLE_STDOUT => { - let mut buf = ctx.stdout.lock().unwrap_or_else(|e| e.into_inner()); - append_capped(&mut buf, bytes); + { + let mut buf = ctx.stdout.lock().unwrap_or_else(|e| e.into_inner()); + append_capped(&mut buf, bytes); + } + if let Some(sink) = ctx.sink.as_ref() { + sink(OutStream::Stdout, bytes); + } } WslcProcessIOHandle::WSLC_PROCESS_IO_HANDLE_STDERR => { - let mut buf = ctx.stderr.lock().unwrap_or_else(|e| e.into_inner()); - append_capped(&mut buf, bytes); + { + let mut buf = ctx.stderr.lock().unwrap_or_else(|e| e.into_inner()); + append_capped(&mut buf, bytes); + } + if let Some(sink) = ctx.sink.as_ref() { + sink(OutStream::Stderr, bytes); + } } _ => {} } @@ -274,8 +303,9 @@ impl ProcessSettings { script_code: &str, env: &[String], working_directory: &str, + sink: Option, ) -> Result { - Self::build_inner(sdk, script_code, env, working_directory, true) + Self::build_inner(sdk, script_code, env, working_directory, true, sink) } /// Like [`build`](Self::build) but registers no stdio callbacks and shares no @@ -290,7 +320,7 @@ impl ProcessSettings { env: &[String], working_directory: &str, ) -> Result { - Self::build_inner(sdk, script_code, env, working_directory, false) + Self::build_inner(sdk, script_code, env, working_directory, false, None) } unsafe fn build_inner( @@ -299,6 +329,7 @@ impl ProcessSettings { env: &[String], working_directory: &str, register_callbacks: bool, + sink: Option, ) -> Result { let mut raw = std::mem::zeroed::(); let hr = sdk.WslcInitProcessSettings(&mut raw); @@ -310,6 +341,7 @@ impl ProcessSettings { stdout: Arc::new(Mutex::new(Vec::new())), stderr: Arc::new(Mutex::new(Vec::new())), exited: Arc::new((Mutex::new(false), Condvar::new())), + sink, }); // Callbacks are registered only for the streamed path; a detached init @@ -858,6 +890,10 @@ pub struct ExecOutcome { /// # Safety /// `sdk` must hold valid function pointers and `container` must be a live, /// started handle. +// A thin FFI primitive whose parameters mirror the SDK's process inputs plus +// the optional live-output sink; grouping them into a struct would only add an +// indirection for a single call site. +#[allow(clippy::too_many_arguments)] pub unsafe fn exec_in_container( sdk: &WslcSdk, container: WslcContainer, @@ -865,12 +901,14 @@ pub unsafe fn exec_in_container( env: &[String], working_directory: &str, timeout_ms: u32, + sink: Option, logger: &mut Logger, ) -> Result { // `process_settings` owns every buffer the SDK reads at // `WslcCreateContainerProcess` time plus the I/O-capture context; it is held // as a stationary local until after the process exits below. - let mut process_settings = ProcessSettings::build(sdk, script_code, env, working_directory)?; + let mut process_settings = + ProcessSettings::build(sdk, script_code, env, working_directory, sink)?; let mut process: WslcProcess = ptr::null_mut(); let mut err_msg = CoTaskMemPWSTR::null(); diff --git a/src/backends/wslc/common/src/daemon_client.rs b/src/backends/wslc/common/src/daemon_client.rs index 94a4688c9..707af3e14 100644 --- a/src/backends/wslc/common/src/daemon_client.rs +++ b/src/backends/wslc/common/src/daemon_client.rs @@ -38,6 +38,7 @@ use anyhow::{bail, Context, Result}; use serde::de::DeserializeOwned; use serde::Serialize; +use crate::container_steps::OutStream; use crate::daemon_protocol::{ encode_frame, DaemonRequest, DaemonResponse, DeprovisionConfig, ErrKind, ExecConfig, ProvisionConfig, StartConfig, StopConfig, StreamFrame, MAX_FRAME_SIZE, PROTOCOL_VERSION, @@ -122,10 +123,10 @@ const OPEN_RETRY: Duration = Duration::from_millis(20); const ERROR_FILE_NOT_FOUND: i32 = 2; const ERROR_PIPE_BUSY: i32 = 231; -/// The captured result of an exec run. Live stdout/stderr framing is a daemon -/// fill-in; today the daemon returns only the terminal exit code, so these -/// buffers are usually empty, but the client already accumulates any -/// [`StreamFrame::Stdout`] / [`StreamFrame::Stderr`] the daemon sends. +/// The captured result of an exec run: the exit code plus fully-buffered +/// stdout/stderr. Prefer [`DaemonClient::exec_streaming`] to relay output as it +/// arrives; this buffering variant is for callers (and tests) that want the +/// whole capture at once. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct ExecResult { pub exit_code: i32, @@ -285,12 +286,39 @@ impl DaemonClient { } /// Run a command in a started container to completion, returning its exit - /// code and any streamed output. + /// code and fully-buffered output. /// - /// After the daemon admits the exec with `Ok`, this reads the - /// [`StreamFrame`] data phase — accumulating stdout/stderr — until the - /// terminal [`StreamFrame::Exit`] (or [`StreamFrame::Error`]). + /// A convenience wrapper over [`exec_streaming`](Self::exec_streaming) that + /// accumulates every stdout/stderr chunk into an [`ExecResult`]. Prefer + /// `exec_streaming` when output should be relayed live rather than buffered. pub fn exec(&self, config: ExecConfig) -> DaemonResult { + let mut stdout = Vec::new(); + let mut stderr = Vec::new(); + let exit_code = self.exec_streaming(config, |stream, data| match stream { + OutStream::Stdout => stdout.extend_from_slice(data), + OutStream::Stderr => stderr.extend_from_slice(data), + })?; + Ok(ExecResult { + exit_code, + stdout, + stderr, + }) + } + + /// Run a command in a started container, invoking `on_output` for each live + /// stdout/stderr chunk **as it arrives**, and returning the exit code once + /// the run completes. + /// + /// After the daemon admits the exec with `Ok`, this reads the + /// [`StreamFrame`] data phase, dispatching `Stdout`/`Stderr` chunks to the + /// callback until the terminal [`StreamFrame::Exit`] (or + /// [`StreamFrame::Error`]). Unlike [`exec`](Self::exec), nothing is buffered + /// here — the caller decides what to do with each chunk. + pub fn exec_streaming( + &self, + config: ExecConfig, + mut on_output: impl FnMut(OutStream, &[u8]), + ) -> DaemonResult { let mut pipe = self.open_pipe()?; write_frame(&mut pipe, &DaemonRequest::Exec(config))?; @@ -306,15 +334,11 @@ impl DaemonClient { } } - let mut result = ExecResult::default(); loop { match read_frame::(&mut pipe)? { - StreamFrame::Stdout { data } => result.stdout.extend_from_slice(&data), - StreamFrame::Stderr { data } => result.stderr.extend_from_slice(&data), - StreamFrame::Exit { code } => { - result.exit_code = code; - return Ok(result); - } + StreamFrame::Stdout { data } => on_output(OutStream::Stdout, &data), + StreamFrame::Stderr { data } => on_output(OutStream::Stderr, &data), + StreamFrame::Exit { code } => return Ok(code), StreamFrame::Error { message } => { return Err(DaemonError::transport(format!("exec failed: {message}"))) } diff --git a/src/backends/wslc/common/src/state_aware.rs b/src/backends/wslc/common/src/state_aware.rs index 729554fa7..cba6cc7c2 100644 --- a/src/backends/wslc/common/src/state_aware.rs +++ b/src/backends/wslc/common/src/state_aware.rs @@ -13,7 +13,8 @@ //! //! Windows-only: the daemon and its pipe transport are a Windows feature. -use std::io::Write; +use std::io::{IsTerminal, Write}; +use std::time::{Duration, Instant}; use wxc_common::logger::{Logger, Mode}; use wxc_common::models::{ContainerPolicy, ExecutionRequest, NetworkPolicy}; @@ -24,6 +25,7 @@ use wxc_common::state_aware_backend::{ }; use wxc_common::wire::WslcProvisionPhase; +use crate::container_steps::OutStream; use crate::daemon_client::{DaemonClient, DaemonError}; use crate::daemon_protocol::{ DeprovisionConfig, ErrKind, ExecConfig, NetworkMode, ProvisionConfig, StartConfig, StopConfig, @@ -36,6 +38,42 @@ use crate::policy::{ /// Default image when a provision request omits `experimental.wslc.provision.image`. const DEFAULT_IMAGE: &str = "alpine:latest"; +/// Bytes accumulated on a non-TTY exec output stream before [`FlushGate`] forces +/// a flush, bounding how much newline-free output can sit buffered. +const NON_TTY_FLUSH_BYTES: usize = 32 * 1024; +/// Max wall-clock between flushes on a non-TTY exec output stream, so slow +/// carriage-return progress output reaches a pipe consumer promptly. +const NON_TTY_FLUSH_INTERVAL: Duration = Duration::from_millis(200); + +/// Bounded flush policy for non-TTY exec output: flush once enough bytes have +/// accumulated or enough wall-clock has elapsed, so a pipe consumer sees +/// newline-free progress output promptly without a syscall per chunk. +struct FlushGate { + last: Instant, + pending: usize, +} + +impl FlushGate { + fn new() -> Self { + Self { + last: Instant::now(), + pending: 0, + } + } + + /// Record `n` freshly-written bytes and report whether to flush now. + fn should_flush(&mut self, n: usize) -> bool { + self.pending += n; + if self.pending >= NON_TTY_FLUSH_BYTES || self.last.elapsed() >= NON_TTY_FLUSH_INTERVAL { + self.pending = 0; + self.last = Instant::now(); + true + } else { + false + } + } +} + /// State-aware WSLc backend. Zero-sized: every phase opens a fresh /// [`DaemonClient`] connection (the daemon holds all persistent state). #[derive(Debug, Default, Clone, Copy)] @@ -126,12 +164,11 @@ impl StatefulSandboxBackend for WslcStateAwareRunner { Ok(DeprovisionResult { metadata: None }) } - /// Runs one command in the warm container. The daemon buffers the run to - /// completion and returns the captured stdout/stderr plus the exit code; - /// this relays the buffers to the executor's own stdio, then hands back an + /// Runs one command in the warm container, relaying its stdout/stderr to the + /// executor's own stdio **live** as the daemon streams it, then hands back an /// [`ExecHandle`] with sentinel pipe handles and a waiter that yields the - /// already-captured exit code (so the dispatcher's `relay_exec_to_stdio` is - /// a thin call-through, mirroring the IsolationSession backend). + /// captured exit code (so the dispatcher's `relay_exec_to_stdio` is a thin + /// call-through, mirroring the IsolationSession and Windows Sandbox backends). fn exec( &mut self, sandbox_id: &str, @@ -152,30 +189,46 @@ impl StatefulSandboxBackend for WslcStateAwareRunner { }; let client = connect_daemon()?; - let result = client - .exec(ExecConfig { - sandbox_id: sandbox_id.to_string(), - script_code: request.script_code.clone(), - working_directory: request.working_directory.clone(), - env, - timeout_ms: request.script_timeout, - }) + + // Relay each chunk to our own stdio as it arrives. Best-effort: a failed + // local write must not mask the container's exit code. A `FlushGate` + // bounds latency on a non-TTY pipe consumer; a TTY flushes every chunk. + let mut stdout = std::io::stdout(); + let mut stderr = std::io::stderr(); + let stdout_is_tty = stdout.is_terminal(); + let stderr_is_tty = stderr.is_terminal(); + let mut stdout_gate = FlushGate::new(); + let mut stderr_gate = FlushGate::new(); + + let exit_code = client + .exec_streaming( + ExecConfig { + sandbox_id: sandbox_id.to_string(), + script_code: request.script_code.clone(), + working_directory: request.working_directory.clone(), + env, + timeout_ms: request.script_timeout, + }, + |stream, bytes| match stream { + OutStream::Stdout => { + let _ = stdout.write_all(bytes); + if stdout_is_tty || stdout_gate.should_flush(bytes.len()) { + let _ = stdout.flush(); + } + } + OutStream::Stderr => { + let _ = stderr.write_all(bytes); + if stderr_is_tty || stderr_gate.should_flush(bytes.len()) { + let _ = stderr.flush(); + } + } + }, + ) .map_err(map_daemon_error)?; - // Relay the daemon-captured buffers to our own stdio. Best-effort: - // a failed local write must not mask the container's exit code. - if !result.stdout.is_empty() { - let mut out = std::io::stdout(); - let _ = out.write_all(&result.stdout); - let _ = out.flush(); - } - if !result.stderr.is_empty() { - let mut err = std::io::stderr(); - let _ = err.write_all(&result.stderr); - let _ = err.flush(); - } + let _ = stdout.flush(); + let _ = stderr.flush(); - let exit_code = result.exit_code; Ok(ExecHandle { stdout: null_pipe_handle(), stderr: null_pipe_handle(), diff --git a/src/backends/wslc/daemon/src/control_server.rs b/src/backends/wslc/daemon/src/control_server.rs index f28df9518..09256bb43 100644 --- a/src/backends/wslc/daemon/src/control_server.rs +++ b/src/backends/wslc/daemon/src/control_server.rs @@ -40,11 +40,12 @@ use windows::Win32::Security::{ }; use windows::Win32::System::Threading::{GetCurrentProcess, OpenProcessToken}; +use wslc_common::container_steps::OutStream; use wslc_common::daemon_protocol::{ encode_frame, DaemonRequest, DaemonResponse, StreamFrame, MAX_FRAME_SIZE, }; -use crate::session_manager::{SessionHandle, WorkerError}; +use crate::session_manager::{ExecStream, SessionHandle, WorkerError}; /// Upper bound on concurrently-serviced client connections. At capacity the /// accept loop applies backpressure (a new connection waits for a slot) instead @@ -407,7 +408,8 @@ async fn handle_client(mut pipe: NamedPipeServer, session: SessionHandle) -> Res Ok(()) } -/// Exec: validate-then-admit, then stream the run's outcome as [`StreamFrame`]s. +/// Exec: validate-then-admit, then stream the run's stdout/stderr live as +/// [`StreamFrame`]s, followed by a terminal frame. /// /// The sandbox is validated (exists + started) *before* the `Ok` admission is /// written, and — critically — admission is **atomic** with the start of the @@ -417,9 +419,13 @@ async fn handle_client(mut pipe: NamedPipeServer, session: SessionHandle) -> Res /// unknown/not-started sandbox therefore comes back as a pre-admission typed /// [`DaemonResponse::Err`] rather than a post-admission stream `Error` frame. /// -/// TODO(fill-in): bidirectional live stdio (client `Stdin` frames -> process, -/// process stdout/stderr -> `Stdout`/`Stderr` frames). The skeleton runs the -/// command to completion and emits only the terminal frame. +/// Output streaming (process -> `Stdout`/`Stderr`) is live. Client `Stdin` +/// frames are NOT forwarded: the WSLc SDK consumes all process IO handles once +/// any `WslcSetProcessSettingsCallbacks` is registered (the callback path this +/// live output streaming depends on), so `WslcGetProcessIOHandle(STDIN)` is +/// unavailable. Piped stdin would require a handle-mode rearchitecture (no +/// callbacks; `ReadFile` threads for stdout/stderr + `WriteFile` for stdin) and +/// is deferred to Tier 2 (see the 2c plan). async fn handle_exec( mut pipe: NamedPipeServer, session: SessionHandle, @@ -433,22 +439,78 @@ async fn handle_exec( /// Turn an exec **admission** outcome into the client's frame sequence, generic /// over the transport so the protocol can be exercised over an in-memory duplex /// in tests. On rejection it writes a single typed [`DaemonResponse::Err`]; on -/// admission it writes `Ok` then awaits completion and writes exactly one -/// terminal [`StreamFrame`] — `Exit` on success, `Error` on a run failure or a -/// dropped completion channel. +/// admission it writes `Ok`, then pumps live `Stdout`/`Stderr` frames as output +/// arrives, and finally writes exactly one terminal [`StreamFrame`] — `Exit` on +/// success, `Error` on a run failure or a dropped completion channel. async fn write_exec_result( pipe: &mut S, - admission: Result>, WorkerError>, + admission: Result, ) -> Result<()> { - let done = match admission { - Ok(done) => done, + let ExecStream { done, mut output } = match admission { + Ok(stream) => stream, Err(e) => { write_frame(pipe, &worker_err_response(e)).await?; return Ok(()); } }; write_frame(pipe, &DaemonResponse::Ok).await?; - let terminal = match done.await { + + // Pump live output until the run completes, then write the terminal frame. + // + // Two completion signals, because the output channel does not always close: + // - Normal path: the worker's `IoContext` (and the sink's sender) drop when + // the run returns, so `output` closes *before* `done` fires; the biased + // select drains every queued chunk first, then observes the close. + // - Kill/leak path: a container killed without its exit callback leaks the + // `IoContext` (deliberately), so the sink's sender never drops and + // `output` never closes. There `done` is the only completion signal, so + // we drain whatever is already queued and stop. + let mut done = done; + let terminal = loop { + tokio::select! { + biased; + chunk = output.recv() => match chunk { + Some(chunk) => { + write_frame(pipe, &output_frame(chunk)).await?; + } + // Senders dropped: the run has completed and every chunk is + // flushed. Await the (already-resolved) exit code. + None => break exit_terminal(done.await), + }, + result = &mut done => { + // Run finished but the output channel may still be open (leak + // path: a killed container whose exit callback never fired keeps + // the sink's sender alive). Close the receiver first so any + // leaked callback can enqueue nothing further — otherwise a + // container still producing output could keep `try_recv` + // returning chunks forever and the terminal frame would never be + // written. Then flush what is already queued and stop. + output.close(); + while let Ok(chunk) = output.try_recv() { + write_frame(pipe, &output_frame(chunk)).await?; + } + break exit_terminal(result); + } + } + }; + write_frame(pipe, &terminal).await?; + Ok(()) +} + +/// Map a live-output chunk to its wire frame. +fn output_frame((kind, data): (OutStream, Vec)) -> StreamFrame { + match kind { + OutStream::Stdout => StreamFrame::Stdout { data }, + OutStream::Stderr => StreamFrame::Stderr { data }, + } +} + +/// Map a completed exec's result (or a dropped completion channel) to its +/// terminal [`StreamFrame`]. +fn exit_terminal( + result: Result, oneshot::error::RecvError>, +) -> StreamFrame { + match result { Ok(Ok(code)) => StreamFrame::Exit { code }, Ok(Err(e)) => StreamFrame::Error { message: e.to_string(), @@ -456,9 +518,7 @@ async fn write_exec_result( Err(_) => StreamFrame::Error { message: "WSLc worker dropped the exec reply channel".to_string(), }, - }; - write_frame(pipe, &terminal).await?; - Ok(()) + } } /// Map a worker `Result<()>` to an `Ok` / typed `Err` response. @@ -502,8 +562,19 @@ async fn write_frame(pipe: &mut S, msg: &T) mod tests { use super::*; use tokio::io::duplex; + use tokio::sync::mpsc; use wslc_common::daemon_protocol::ErrKind; + /// Admit an exec whose output channel is already closed (no live output), + /// so `write_exec_result` goes straight from `Ok` to the terminal frame. + fn admitted_no_output( + done: oneshot::Receiver>, + ) -> Result { + let (tx, output) = mpsc::channel(16); + drop(tx); + Ok(ExecStream { done, output }) + } + /// A rejected admission (unknown sandbox) round-trips as a single typed /// `DaemonResponse::Err { NotProvisioned }` through frame encode → transport /// → decode, with no terminal frame following it. @@ -556,7 +627,9 @@ mod tests { let (done_tx, done_rx) = oneshot::channel::>(); drop(done_tx); - write_exec_result(&mut server, Ok(done_rx)).await.unwrap(); + write_exec_result(&mut server, admitted_no_output(done_rx)) + .await + .unwrap(); drop(server); let admit: DaemonResponse = read_frame(&mut client).await.unwrap(); @@ -583,7 +656,9 @@ mod tests { let (done_tx, done_rx) = oneshot::channel::>(); done_tx.send(Ok(7)).unwrap(); - write_exec_result(&mut server, Ok(done_rx)).await.unwrap(); + write_exec_result(&mut server, admitted_no_output(done_rx)) + .await + .unwrap(); drop(server); let admit: DaemonResponse = read_frame(&mut client).await.unwrap(); @@ -591,4 +666,123 @@ mod tests { let terminal: StreamFrame = read_frame(&mut client).await.unwrap(); assert_eq!(terminal, StreamFrame::Exit { code: 7 }); } + + /// Live output is streamed as `Stdout`/`Stderr` frames — in the order the + /// worker enqueued them — before the terminal `Exit`, so the client sees the + /// run's output incrementally rather than as one buffered blob. + #[tokio::test] + async fn live_output_streams_before_exit() { + let (mut server, mut client) = duplex(64 * 1024); + let (done_tx, done_rx) = oneshot::channel::>(); + let (out_tx, output) = mpsc::channel(16); + + // Enqueue interleaved output, then the exit code, then close the channel + // (mirrors the worker: the sink's sender drops as the run returns). + out_tx + .try_send((OutStream::Stdout, b"hello ".to_vec())) + .unwrap(); + out_tx + .try_send((OutStream::Stderr, b"warn".to_vec())) + .unwrap(); + out_tx + .try_send((OutStream::Stdout, b"world".to_vec())) + .unwrap(); + done_tx.send(Ok(0)).unwrap(); + drop(out_tx); + + write_exec_result( + &mut server, + Ok(ExecStream { + done: done_rx, + output, + }), + ) + .await + .unwrap(); + drop(server); + + let admit: DaemonResponse = read_frame(&mut client).await.unwrap(); + assert_eq!(admit, DaemonResponse::Ok); + assert_eq!( + read_frame::<_, StreamFrame>(&mut client).await.unwrap(), + StreamFrame::Stdout { + data: b"hello ".to_vec() + } + ); + assert_eq!( + read_frame::<_, StreamFrame>(&mut client).await.unwrap(), + StreamFrame::Stderr { + data: b"warn".to_vec() + } + ); + assert_eq!( + read_frame::<_, StreamFrame>(&mut client).await.unwrap(), + StreamFrame::Stdout { + data: b"world".to_vec() + } + ); + assert_eq!( + read_frame::<_, StreamFrame>(&mut client).await.unwrap(), + StreamFrame::Exit { code: 0 } + ); + assert!(read_frame::<_, StreamFrame>(&mut client).await.is_err()); + } + + /// Leak path: the run completes (`done` fires) while the sink's sender is + /// still alive and would keep producing. `write_exec_result` must close the + /// receiver, flush only what was already queued, and write the terminal + /// frame — it must not stream chunks enqueued after completion nor hang. + #[tokio::test] + async fn leak_path_drains_queued_then_terminates() { + let (mut server, mut client) = duplex(64 * 1024); + let (done_tx, done_rx) = oneshot::channel::>(); + let (out_tx, output) = mpsc::channel(16); + + // Two chunks already queued, the run reports its exit, and the sender is + // deliberately kept alive (the leaked `IoContext`). + out_tx + .try_send((OutStream::Stdout, b"queued".to_vec())) + .unwrap(); + out_tx + .try_send((OutStream::Stderr, b"tail".to_vec())) + .unwrap(); + done_tx.send(Ok(3)).unwrap(); + + write_exec_result( + &mut server, + Ok(ExecStream { + done: done_rx, + output, + }), + ) + .await + .unwrap(); + + // A post-completion enqueue attempt must fail because the receiver was + // closed, proving a leaked producer cannot extend the stream. + assert!(out_tx + .try_send((OutStream::Stdout, b"after".to_vec())) + .is_err()); + drop(server); + + let admit: DaemonResponse = read_frame(&mut client).await.unwrap(); + assert_eq!(admit, DaemonResponse::Ok); + assert_eq!( + read_frame::<_, StreamFrame>(&mut client).await.unwrap(), + StreamFrame::Stdout { + data: b"queued".to_vec() + } + ); + assert_eq!( + read_frame::<_, StreamFrame>(&mut client).await.unwrap(), + StreamFrame::Stderr { + data: b"tail".to_vec() + } + ); + assert_eq!( + read_frame::<_, StreamFrame>(&mut client).await.unwrap(), + StreamFrame::Exit { code: 3 } + ); + assert!(read_frame::<_, StreamFrame>(&mut client).await.is_err()); + } } diff --git a/src/backends/wslc/daemon/src/session_manager.rs b/src/backends/wslc/daemon/src/session_manager.rs index 460db98f1..817e75795 100644 --- a/src/backends/wslc/daemon/src/session_manager.rs +++ b/src/backends/wslc/daemon/src/session_manager.rs @@ -19,17 +19,17 @@ //! Each phase drives the real WSLc SDK via the reusable steps in //! [`wslc_common::container_steps`]: `provision` ensures the session + resolves //! the image + creates a container with a keepalive init process; `start` boots -//! it; `exec` runs a fresh `WslcCreateContainerProcess` to completion; `stop` / -//! `deprovision` stop + delete. Live bidirectional stdio streaming over the -//! control pipe is still a later fill-in — `exec` currently returns the exit -//! code only. +//! it; `exec` runs a fresh `WslcCreateContainerProcess` to completion, streaming +//! its stdout/stderr live to the pipe handler via an [`OutputSink`]; `stop` / +//! `deprovision` stop + delete. The completion reply carries the exit code; +//! output flows over the sink. (Client `Stdin` forwarding is a later fill-in.) use std::collections::HashMap; use anyhow::Result; use tokio::sync::{mpsc, oneshot}; -use wslc_common::container_steps::{self, ProcessSettings}; +use wslc_common::container_steps::{self, OutStream, OutputSink, ProcessSettings}; use wslc_common::daemon_protocol::{ DeprovisionConfig, ErrKind, ExecConfig, NetworkMode, ProvisionConfig, StartConfig, StopConfig, }; @@ -119,6 +119,10 @@ pub enum WorkerCommand { /// post-admission stream `Error`); `done` carries the run's exit code. Exec { config: ExecConfig, + /// Live-output sink the worker hands to `exec_in_container`; the SDK's + /// stdout/stderr callbacks push chunks through it to the pipe handler as + /// bytes arrive, alongside the capped capture buffers. + sink: OutputSink, admit: oneshot::Sender>, done: oneshot::Sender>, }, @@ -136,6 +140,27 @@ pub enum WorkerCommand { Shutdown { reply: oneshot::Sender<()> }, } +/// A chunk of live process output streamed from the worker to the pipe handler: +/// which stream it came from and the bytes (owned, so it can cross the channel). +pub type OutputChunk = (OutStream, Vec); + +/// Bound on the number of unconsumed live-output chunks buffered between the +/// SDK's I/O callback threads and the pipe handler. The channel is bounded (not +/// unbounded) so a container emitting output faster than a slow client drains it +/// cannot grow the queue without limit and OOM the persistent daemon, taking +/// down every sandbox it owns — the same hazard the capture-buffer cap guards +/// against. When the queue is full the sink applies backpressure (see +/// [`SessionHandle::exec`]) rather than dropping bytes. +const LIVE_OUTPUT_CHANNEL_CAPACITY: usize = 256; + +/// An admitted exec: the completion receiver (the run's exit code) plus the +/// live-output receiver the pipe handler drains into `Stdout`/`Stderr` frames. +#[derive(Debug)] +pub struct ExecStream { + pub done: oneshot::Receiver>, + pub output: mpsc::Receiver, +} + /// A cheap, clonable handle async tasks use to drive the worker thread. #[derive(Clone)] pub struct SessionHandle { @@ -160,23 +185,37 @@ impl SessionHandle { /// Admit and run a command in a started container. Awaits the worker's /// **admission** decision first: on rejection (unknown/not-started sandbox) /// this returns the typed error *before* the caller writes any admission to - /// the client. On admission it returns the completion receiver, which - /// resolves to the run's exit code. Admission and the start of the run are - /// atomic on the worker thread, so no lifecycle command can invalidate the - /// checked state between the two. - pub async fn exec( - &self, - config: ExecConfig, - ) -> Result>, WorkerError> { + /// the client. On admission it returns an [`ExecStream`] — the completion + /// receiver (the run's exit code) plus the live-output receiver, which the + /// caller drains into `Stdout`/`Stderr` frames as bytes arrive. Admission and + /// the start of the run are atomic on the worker thread, so no lifecycle + /// command can invalidate the checked state between the two. + pub async fn exec(&self, config: ExecConfig) -> Result { let (admit, admit_rx) = oneshot::channel(); let (done, done_rx) = oneshot::channel(); + let (stream_tx, output) = mpsc::channel::(LIVE_OUTPUT_CHANNEL_CAPACITY); + // The sink is invoked from the SDK's I/O callback threads (native OS + // threads, never the async runtime). `blocking_send` on the bounded + // channel parks that thread when the queue is full, propagating + // backpressure to the container's stdout/stderr pipe — the container + // blocks producing until the pipe handler drains, so memory stays + // bounded and no output bytes are dropped. A closed receiver (client + // gone, or the leak-path `close()` in the pipe handler) makes the send + // return `Err`, so a leaked callback never parks forever. + let sink: OutputSink = Box::new(move |kind, bytes| { + let _ = stream_tx.blocking_send((kind, bytes.to_vec())); + }); self.send(WorkerCommand::Exec { config, + sink, admit, done, })?; admit_rx.await.map_err(worker_gone)??; - Ok(done_rx) + Ok(ExecStream { + done: done_rx, + output, + }) } /// Stop a running container. @@ -377,7 +416,13 @@ impl Worker { /// Run a command in a sandbox whose existence/started state was already /// confirmed by [`validate_exec`]; `container` is that validated handle. - fn exec(&mut self, config: ExecConfig, container: WslcContainer) -> Result { + /// `sink` streams the run's stdout/stderr live to the pipe handler. + fn exec( + &mut self, + config: ExecConfig, + container: WslcContainer, + sink: OutputSink, + ) -> Result { let sdk = self .sdk .as_ref() @@ -399,6 +444,7 @@ impl Worker { &env, &config.working_directory, config.timeout_ms, + Some(sink), &mut self.logger, ) } @@ -429,9 +475,8 @@ impl Worker { config.timeout_ms ))); } - // NOTE: outcome.stdout/stderr are captured but not yet forwarded — live - // stdio streaming over the control pipe is a later fill-in; the PR1 - // contract returns the exit code only. + // outcome.stdout/stderr were captured (and already streamed live via the + // sink); the completion reply carries only the exit code. Ok(outcome.exit_code) } @@ -541,6 +586,7 @@ pub fn spawn() -> Result { } WorkerCommand::Exec { config, + sink, admit, done, } => { @@ -557,7 +603,7 @@ pub fn spawn() -> Result { // every other lifecycle command for its full timeout. Ok(container) if admit.send(Ok(())).is_ok() => { let sandbox_id = config.sandbox_id.clone(); - let outcome = worker.exec(config, container); + let outcome = worker.exec(config, container, sink); if let Err(orphaned) = done.send(outcome) { // The client handler is gone (e.g. its // post-admission Ok write failed) but the run @@ -753,7 +799,7 @@ mod tests { .await .unwrap(); - let code = handle + let mut exec = handle .exec(ExecConfig { sandbox_id: id.clone(), script_code: "echo hi".to_string(), @@ -762,11 +808,17 @@ mod tests { timeout_ms: 30_000, }) .await - .unwrap() - .await - .unwrap() .unwrap(); + // Drain the live output stream, then await the exit code. + let mut stdout = Vec::new(); + while let Some((kind, data)) = exec.output.recv().await { + if kind == OutStream::Stdout { + stdout.extend_from_slice(&data); + } + } + let code = exec.done.await.unwrap().unwrap(); assert_eq!(code, 0); + assert_eq!(String::from_utf8_lossy(&stdout).trim(), "hi"); handle .stop(StopConfig { diff --git a/tests/configs/wslc_state_aware_exec_drip.json b/tests/configs/wslc_state_aware_exec_drip.json new file mode 100644 index 000000000..7ca158cdb --- /dev/null +++ b/tests/configs/wslc_state_aware_exec_drip.json @@ -0,0 +1,9 @@ +{ + "version": "0.8.0-alpha", + "phase": "exec", + "sandboxId": "{{SANDBOX_ID}}", + "process": { + "commandLine": "sh -c 'echo DRIP-PART1; sleep 2; echo DRIP-PART2'", + "timeout": 30000 + } +} diff --git a/tests/scripts/run_wslc_state_aware_tests.ps1 b/tests/scripts/run_wslc_state_aware_tests.ps1 index 184fd31c4..2585636f4 100644 --- a/tests/scripts/run_wslc_state_aware_tests.ps1 +++ b/tests/scripts/run_wslc_state_aware_tests.ps1 @@ -197,6 +197,76 @@ function Invoke-StateAware { } } +# Like Invoke-StateAware but records the wall-clock arrival time of each stdout +# line, so a test can prove output is streamed incrementally (an early line lands +# well before a later one) rather than buffered and dumped together at process +# exit. `ReadLineAsync` returns the moment the child flushes a newline-terminated +# line, so a streamed line is observed immediately; a buffer-then-dump impl would +# surface every line at once only when the process exits. Returns +# @{ ExitCode; Stdout; Stderr; Lines = @(@{ Text; At }) } (At = UTC DateTime). +function Invoke-StateAwareStreaming { + param( + [string]$ConfigFile, + [hashtable]$Request, + [string]$SandboxId + ) + + if ($ConfigFile) { + $path = Join-Path $ConfigDir $ConfigFile + if (-not (Test-Path $path)) { throw "Config fixture not found: $path" } + $json = Get-Content $path -Raw + if ($json -match '\{\{SANDBOX_ID\}\}') { + if (-not $SandboxId) { + throw "Fixture $ConfigFile contains {{SANDBOX_ID}} but -SandboxId was not supplied" + } + $json = $json -replace '\{\{SANDBOX_ID\}\}', $SandboxId + } + } elseif ($Request) { + $json = $Request | ConvertTo-Json -Compress -Depth 12 + } else { + throw "Invoke-StateAwareStreaming requires either -Request or -ConfigFile" + } + + $b64 = [Convert]::ToBase64String([System.Text.Encoding]::UTF8.GetBytes($json)) + $argList = @('--experimental') + if ($Debug) { $argList += '--debug' } + $argList += @('--config-base64', $b64) + + $psi = [System.Diagnostics.ProcessStartInfo]::new() + $psi.FileName = $WxcExec + foreach ($a in $argList) { $psi.ArgumentList.Add($a) } + $psi.UseShellExecute = $false + $psi.RedirectStandardOutput = $true + $psi.RedirectStandardError = $true + $psi.CreateNoWindow = $true + + $proc = [System.Diagnostics.Process]::new() + $proc.StartInfo = $psi + $null = $proc.Start() + + # Drain stderr async so a large stderr can never deadlock the stdout read. + $stderrTask = $proc.StandardError.ReadToEndAsync() + + $lines = New-Object System.Collections.Generic.List[object] + $sb = [System.Text.StringBuilder]::new() + while ($true) { + $line = $proc.StandardOutput.ReadLineAsync().GetAwaiter().GetResult() + if ($null -eq $line) { break } + $null = $lines.Add(@{ Text = $line; At = [DateTime]::UtcNow }) + $null = $sb.AppendLine($line) + } + $proc.WaitForExit() + $stderrText = $stderrTask.GetAwaiter().GetResult() + $exitCode = $proc.ExitCode + $proc.Dispose() + @{ + ExitCode = $exitCode + Stdout = $sb.ToString() + Stderr = if ($null -eq $stderrText) { "" } else { [string]$stderrText } + Lines = $lines + } +} + # Parse the wxc-exec stdout envelope; $null if not valid JSON. function Parse-Envelope { param([string]$Stdout) @@ -233,22 +303,6 @@ function Assert-True { } } -# Non-failing probe for assertions that depend on exec STDOUT/STDERR content. -# Live exec output forwarding is deferred: the daemon's handle_exec runs the -# command to completion and emits only the terminal Exit frame, so container -# output is captured but not yet framed back to the caller (exit codes, timeouts, -# and the phase state machine all work). These probes print PASS the moment the -# output path lands and an INFO note until then, keeping the lifecycle-mechanic -# coverage green in the interim without masking a hard-assert regression. -function Probe-ExecOutput { - param([bool]$Condition, [string]$Message) - if ($Condition) { - Write-Host " PASS: $Message" -ForegroundColor Green - } else { - Write-Host " INFO (exec-output deferred): $Message" -ForegroundColor DarkYellow - } -} - function Run-StateAwareTest { param([string]$Name, [scriptblock]$Body) Write-Host "" @@ -356,7 +410,7 @@ try { $execedOk = Run-StateAwareTest "A: exec (basic)" { $r = Invoke-StateAware -ConfigFile 'wslc_state_aware_exec_basic.json' -SandboxId $script:sandboxId Assert-True ($r.ExitCode -eq 0) "exit code = 0 on success" - Probe-ExecOutput ($r.Stdout -match 'wslc-state-aware-exec-marker') ` + Assert-True ($r.Stdout -match 'wslc-state-aware-exec-marker') ` "stdout contains the script's output (relayed live, not enveloped)" $maybeEnv = Parse-Envelope -Stdout $r.Stdout Assert-True ($null -eq $maybeEnv -or $null -eq $maybeEnv.error) ` @@ -364,6 +418,28 @@ try { } } + # A3b: LIVE DRIP -- prove output is streamed incrementally, not buffered and + # dumped at exit. The script prints PART1, sleeps 2s, then prints PART2. With + # live streaming the harness observes PART1 well before PART2 (~the sleep); + # a buffer-then-dump impl would surface both lines together at process exit + # (gap ~0). Asserting the inter-line arrival gap is the airtight liveness + # proof a content-only "stdout contains X" check cannot give. + if ($execedOk) { + Run-StateAwareTest "A: exec (live drip -- incremental streaming)" { + $r = Invoke-StateAwareStreaming -ConfigFile 'wslc_state_aware_exec_drip.json' -SandboxId $script:sandboxId + Assert-True ($r.ExitCode -eq 0) "drip exec exit code = 0" + $p1 = $r.Lines | Where-Object { $_.Text -match 'DRIP-PART1' } | Select-Object -First 1 + $p2 = $r.Lines | Where-Object { $_.Text -match 'DRIP-PART2' } | Select-Object -First 1 + Assert-True ($null -ne $p1) "PART1 line observed on stdout" + Assert-True ($null -ne $p2) "PART2 line observed on stdout" + if ($p1 -and $p2) { + $gapSec = ($p2.At - $p1.At).TotalSeconds + Assert-True ($gapSec -ge 1.0) ` + ("PART1 arrived >=1.0s before PART2 (gap {0:N2}s) -- streamed live, not buffered" -f $gapSec) + } + } | Out-Null + } + # A4: WARM REUSE -- in-container state continuity across separate wxc-exec # invocations. exec #1 writes /tmp/wslc_sa_marker; exec #2 (a fresh # wxc-exec process) reads it back. This only succeeds if the daemon kept @@ -375,7 +451,7 @@ try { Assert-True ($w.ExitCode -eq 0) "exec #1 (write /tmp marker) exit 0" $rd = Invoke-StateAware -ConfigFile 'wslc_state_aware_exec_read_marker.json' -SandboxId $script:sandboxId Assert-True ($rd.ExitCode -eq 0) "exec #2 (read /tmp marker) exit 0" - Probe-ExecOutput ($rd.Stdout -match 'wslc-warm-marker-content') ` + Assert-True ($rd.Stdout -match 'wslc-warm-marker-content') ` "exec #2 sees the marker exec #1 wrote (container stayed warm across wxc-exec processes)" } | Out-Null } @@ -402,7 +478,7 @@ try { Run-StateAwareTest "A: multi-exec (per-invocation env)" { $r = Invoke-StateAware -ConfigFile 'wslc_state_aware_exec_env.json' -SandboxId $script:sandboxId Assert-True ($r.ExitCode -eq 0) "exit code = 0" - Probe-ExecOutput ($r.Stdout -match 'MY_SA_VAR=state-aware-env-value') ` + Assert-True ($r.Stdout -match 'MY_SA_VAR=state-aware-env-value') ` "wire env block reaches the container ($($r.Stdout.Trim()))" } | Out-Null } @@ -511,7 +587,7 @@ try { } $r = Invoke-StateAware -Request $req Assert-True ($r.ExitCode -eq 0) "container read of ro mount exit 0" - Probe-ExecOutput ($r.Stdout -match 'ro-seed-content') "container reads host-seeded ro content" + Assert-True ($r.Stdout -match 'ro-seed-content') "container reads host-seeded ro content" } | Out-Null Run-StateAwareTest "B: ro mount write denied" { @@ -521,7 +597,7 @@ try { process = @{ commandLine = "sh -c 'echo x > /mnt/c/mxc_wslc_sa_test/ro/should_fail.txt && echo WROTE || echo BLOCKED'"; timeout = 30000 } } $r = Invoke-StateAware -Request $req - Probe-ExecOutput ($r.Stdout -match 'BLOCKED') "write to ro mount is blocked" + Assert-True ($r.Stdout -match 'BLOCKED') "write to ro mount is blocked" Assert-True (-not (Test-Path "$script:BTestRoot\ro\should_fail.txt")) "no file created on host ro path" } | Out-Null } @@ -573,9 +649,9 @@ try { Run-StateAwareTest "C: exec injects cooperative proxy env" { $r = Invoke-StateAware -ConfigFile 'wslc_state_aware_exec_proxy.json' -SandboxId $script:netSandboxId Assert-True ($r.ExitCode -eq 0) "exit code = 0" - Probe-ExecOutput ($r.Stdout -match 'HTTP_PROXY=\[http://127\.0\.0\.1:8888\]') ` + Assert-True ($r.Stdout -match 'HTTP_PROXY=\[http://127\.0\.0\.1:8888\]') ` "HTTP_PROXY injected into the container ($($r.Stdout.Trim()))" - Probe-ExecOutput ($r.Stdout -match 'https_proxy=\[http://127\.0\.0\.1:8888\]') ` + Assert-True ($r.Stdout -match 'https_proxy=\[http://127\.0\.0\.1:8888\]') ` "https_proxy injected into the container" } | Out-Null } @@ -692,7 +768,7 @@ try { $req = @{ phase = 'exec'; sandboxId = $script:reSandboxId; process = @{ commandLine = 'echo pre-restart-ok'; timeout = 30000 } } $r = Invoke-StateAware -Request $req Assert-True ($r.ExitCode -eq 0) "exec #1 exit 0" - Probe-ExecOutput ($r.Stdout -match 'pre-restart-ok') "exec #1 produces output" + Assert-True ($r.Stdout -match 'pre-restart-ok') "exec #1 produces output" } | Out-Null $reStoppedOk = Run-StateAwareTest "E: stop" { @@ -723,7 +799,7 @@ try { $req = @{ phase = 'exec'; sandboxId = $script:reSandboxId; process = @{ commandLine = 'echo post-restart-ok'; timeout = 30000 } } $r = Invoke-StateAware -Request $req Assert-True ($r.ExitCode -eq 0) "exec after re-start exit 0" - Probe-ExecOutput ($r.Stdout -match 'post-restart-ok') "exec after re-start produces output" + Assert-True ($r.Stdout -match 'post-restart-ok') "exec after re-start produces output" } | Out-Null } } @@ -795,7 +871,7 @@ try { $after = @{ phase = 'exec'; sandboxId = $script:edgeSandboxId; process = @{ commandLine = 'echo survived-timeout'; timeout = 30000 } } $r2 = Invoke-StateAware -Request $after Assert-True ($r2.ExitCode -eq 0) "next exec after a timeout succeeds (container stayed warm)" - Probe-ExecOutput ($r2.Stdout -match 'survived-timeout') "warm container still executes commands" + Assert-True ($r2.Stdout -match 'survived-timeout') "warm container still executes commands" } | Out-Null } @@ -805,7 +881,7 @@ try { $req = @{ phase = 'exec'; sandboxId = $script:edgeSandboxId; process = @{ commandLine = 'pwd'; cwd = '/tmp'; timeout = 30000 } } $r = Invoke-StateAware -Request $req Assert-True ($r.ExitCode -eq 0) "exit code = 0" - Probe-ExecOutput ($r.Stdout -match '(^|\s)/tmp\s*$') "pwd reports the requested cwd (/tmp) ($($r.Stdout.Trim()))" + Assert-True ($r.Stdout -match '(^|\s)/tmp\s*$') "pwd reports the requested cwd (/tmp) ($($r.Stdout.Trim()))" } | Out-Null } @@ -890,7 +966,7 @@ try { $readB = @{ phase = 'exec'; sandboxId = $script:mcSandboxB; process = @{ commandLine = "sh -c 'cat /tmp/iso_marker 2>/dev/null || echo NO_MARKER'"; timeout = 30000 } } $rb = Invoke-StateAware -Request $readB Assert-True ($rb.ExitCode -eq 0) "read attempt in B exit 0" - Probe-ExecOutput ($rb.Stdout -match 'NO_MARKER') "B does not see A's marker (isolated /tmp)" + Assert-True ($rb.Stdout -match 'NO_MARKER') "B does not see A's marker (isolated /tmp)" Assert-True (-not ($rb.Stdout -match 'A-secret-content')) "A's content never leaks into B" } | Out-Null } @@ -910,7 +986,7 @@ try { $req = @{ phase = 'exec'; sandboxId = $script:mcSandboxB; process = @{ commandLine = 'echo B-still-alive'; timeout = 30000 } } $r = Invoke-StateAware -Request $req Assert-True ($r.ExitCode -eq 0) "B exec after A deprovision exit 0 (daemon stayed up)" - Probe-ExecOutput ($r.Stdout -match 'B-still-alive') "B remains fully usable after A is gone" + Assert-True ($r.Stdout -match 'B-still-alive') "B remains fully usable after A is gone" } | Out-Null } @@ -990,7 +1066,7 @@ try { $req = @{ phase = 'exec'; sandboxId = $script:recSandboxId; process = @{ commandLine = 'echo recovered-ok'; timeout = 30000 } } $r = Invoke-StateAware -Request $req Assert-True ($r.ExitCode -eq 0) "exec on the respawned daemon exit 0" - Probe-ExecOutput ($r.Stdout -match 'recovered-ok') "respawned daemon executes commands normally" + Assert-True ($r.Stdout -match 'recovered-ok') "respawned daemon executes commands normally" } | Out-Null } From 7639a28a0e69b0a1f4994817ba38e1b665d1c2e1 Mon Sep 17 00:00:00 2001 From: Soham Das Date: Tue, 11 Aug 2026 17:20:29 -0700 Subject: [PATCH 2/7] chunk state-aware live-output frames, flush exec relay per chunk, and add provision delegation regression tests --- src/backends/wslc/common/src/state_aware.rs | 115 ++++++++++-------- .../wslc/daemon/src/session_manager.rs | 53 +++++++- 2 files changed, 115 insertions(+), 53 deletions(-) diff --git a/src/backends/wslc/common/src/state_aware.rs b/src/backends/wslc/common/src/state_aware.rs index cba6cc7c2..2d8b5de5c 100644 --- a/src/backends/wslc/common/src/state_aware.rs +++ b/src/backends/wslc/common/src/state_aware.rs @@ -13,8 +13,7 @@ //! //! Windows-only: the daemon and its pipe transport are a Windows feature. -use std::io::{IsTerminal, Write}; -use std::time::{Duration, Instant}; +use std::io::Write; use wxc_common::logger::{Logger, Mode}; use wxc_common::models::{ContainerPolicy, ExecutionRequest, NetworkPolicy}; @@ -38,42 +37,6 @@ use crate::policy::{ /// Default image when a provision request omits `experimental.wslc.provision.image`. const DEFAULT_IMAGE: &str = "alpine:latest"; -/// Bytes accumulated on a non-TTY exec output stream before [`FlushGate`] forces -/// a flush, bounding how much newline-free output can sit buffered. -const NON_TTY_FLUSH_BYTES: usize = 32 * 1024; -/// Max wall-clock between flushes on a non-TTY exec output stream, so slow -/// carriage-return progress output reaches a pipe consumer promptly. -const NON_TTY_FLUSH_INTERVAL: Duration = Duration::from_millis(200); - -/// Bounded flush policy for non-TTY exec output: flush once enough bytes have -/// accumulated or enough wall-clock has elapsed, so a pipe consumer sees -/// newline-free progress output promptly without a syscall per chunk. -struct FlushGate { - last: Instant, - pending: usize, -} - -impl FlushGate { - fn new() -> Self { - Self { - last: Instant::now(), - pending: 0, - } - } - - /// Record `n` freshly-written bytes and report whether to flush now. - fn should_flush(&mut self, n: usize) -> bool { - self.pending += n; - if self.pending >= NON_TTY_FLUSH_BYTES || self.last.elapsed() >= NON_TTY_FLUSH_INTERVAL { - self.pending = 0; - self.last = Instant::now(); - true - } else { - false - } - } -} - /// State-aware WSLc backend. Zero-sized: every phase opens a fresh /// [`DaemonClient`] connection (the daemon holds all persistent state). #[derive(Debug, Default, Clone, Copy)] @@ -190,15 +153,11 @@ impl StatefulSandboxBackend for WslcStateAwareRunner { let client = connect_daemon()?; - // Relay each chunk to our own stdio as it arrives. Best-effort: a failed - // local write must not mask the container's exit code. A `FlushGate` - // bounds latency on a non-TTY pipe consumer; a TTY flushes every chunk. + // Relay each chunk to our own stdio as it arrives, flushing every chunk + // so newline-free progress output reaches the consumer promptly. + // Best-effort: a failed local write must not mask the container's exit code. let mut stdout = std::io::stdout(); let mut stderr = std::io::stderr(); - let stdout_is_tty = stdout.is_terminal(); - let stderr_is_tty = stderr.is_terminal(); - let mut stdout_gate = FlushGate::new(); - let mut stderr_gate = FlushGate::new(); let exit_code = client .exec_streaming( @@ -212,15 +171,11 @@ impl StatefulSandboxBackend for WslcStateAwareRunner { |stream, bytes| match stream { OutStream::Stdout => { let _ = stdout.write_all(bytes); - if stdout_is_tty || stdout_gate.should_flush(bytes.len()) { - let _ = stdout.flush(); - } + let _ = stdout.flush(); } OutStream::Stderr => { let _ = stderr.write_all(bytes); - if stderr_is_tty || stderr_gate.should_flush(bytes.len()) { - let _ = stderr.flush(); - } + let _ = stderr.flush(); } }, ) @@ -645,6 +600,64 @@ mod tests { assert_eq!(tightened.readonly_paths, vec![d]); } + #[cfg(windows)] + #[test] + fn provision_delegation_tightens_rw_alias_of_denied_and_drops_mount() { + // A writable path that resolves to the same object as a `denied` entry + // (here a case-variant string on case-insensitive NTFS) must tighten to + // denied, and the daemon volumes must be built from that tightened policy + // — otherwise the writable alias would still be mounted, granting access + // the deny was meant to block. + let dir = tempfile::tempdir().unwrap(); + let denied = dir.path().to_str().unwrap().to_string(); + let rw_alias = denied.to_uppercase(); + let raw = ExecutionRequest { + policy: ContainerPolicy { + readwrite_paths: vec![rw_alias], + denied_paths: vec![denied], + ..Default::default() + }, + ..Default::default() + }; + // Pre-fix behaviour the bug relied on: the raw request WOULD mount the + // alias writable. + let raw_mounts = build_daemon_volumes(&raw).unwrap(); + assert_eq!(raw_mounts.len(), 1); + assert!(!raw_mounts[0].read_only); + + let tightened = normalize_and_check_delegation(&raw) + .unwrap() + .expect("rw alias of a denied object should tighten"); + assert!(tightened.readwrite_paths.is_empty()); + assert!(!tightened.denied_paths.is_empty()); + let tightened_req = ExecutionRequest { + policy: tightened, + ..raw.clone() + }; + assert!(build_daemon_volumes(&tightened_req).unwrap().is_empty()); + } + + #[cfg(windows)] + #[test] + fn provision_delegation_rejects_inaccessible_path() { + // A delegated path the invoking user cannot access must fail closed + // before provisioning mounts anything. `C:\mxc_invalid); /// [`SessionHandle::exec`]) rather than dropping bytes. const LIVE_OUTPUT_CHANNEL_CAPACITY: usize = 256; +/// Max bytes per enqueued live-output chunk. A single SDK callback can deliver +/// an arbitrarily large buffer; splitting it here bounds each queue entry's +/// allocation and keeps the resulting `Stdout`/`Stderr` frame well under the +/// protocol's `MAX_FRAME_SIZE` (a `Vec` serializes as a JSON number array, +/// ~4x expansion), so a large callback can never overflow a frame and abort the +/// stream before its terminal frame. +const LIVE_OUTPUT_MAX_CHUNK_BYTES: usize = 64 * 1024; + +/// Enqueue an SDK output callback, splitting it into `LIVE_OUTPUT_MAX_CHUNK_BYTES` +/// pieces so each queue entry and its resulting frame stay bounded regardless of +/// the callback's buffer size. Stops early once the receiver is gone (client +/// left / leak-path `close()`), so a leaked callback never parks forever. +fn enqueue_output(tx: &mpsc::Sender, kind: OutStream, bytes: &[u8]) { + for chunk in bytes.chunks(LIVE_OUTPUT_MAX_CHUNK_BYTES) { + if tx.blocking_send((kind, chunk.to_vec())).is_err() { + break; + } + } +} + /// An admitted exec: the completion receiver (the run's exit code) plus the /// live-output receiver the pipe handler drains into `Stdout`/`Stderr` frames. #[derive(Debug)] @@ -203,7 +223,7 @@ impl SessionHandle { // gone, or the leak-path `close()` in the pipe handler) makes the send // return `Err`, so a leaked callback never parks forever. let sink: OutputSink = Box::new(move |kind, bytes| { - let _ = stream_tx.blocking_send((kind, bytes.to_vec())); + enqueue_output(&stream_tx, kind, bytes); }); self.send(WorkerCommand::Exec { config, @@ -767,7 +787,36 @@ mod tests { handle.shutdown().await.unwrap(); } - // ---- Full lifecycle integration test (WSL2 host only) ---- + #[tokio::test] + async fn large_callback_split_into_frame_safe_chunks() { + use wslc_common::daemon_protocol::{encode_frame, StreamFrame, MAX_FRAME_SIZE}; + + let (tx, mut rx) = mpsc::channel::(LIVE_OUTPUT_CHANNEL_CAPACITY); + // A single callback far larger than one chunk (and, unsplit, larger than + // one frame once number-array-encoded): must arrive as many bounded chunks. + let payload = vec![b'x'; LIVE_OUTPUT_MAX_CHUNK_BYTES * 3 + 7]; + let producer = tokio::task::spawn_blocking({ + let payload = payload.clone(); + move || enqueue_output(&tx, OutStream::Stdout, &payload) + }); + + let mut reassembled = Vec::new(); + let mut chunks = 0usize; + while let Some((kind, data)) = rx.recv().await { + assert_eq!(kind, OutStream::Stdout); + assert!(data.len() <= LIVE_OUTPUT_MAX_CHUNK_BYTES); + let encoded = encode_frame(&StreamFrame::Stdout { data: data.clone() }).unwrap(); + assert!(encoded.len() <= MAX_FRAME_SIZE); + reassembled.extend_from_slice(&data); + chunks += 1; + } + producer.await.unwrap(); + assert_eq!(reassembled, payload); + assert!( + chunks >= 4, + "expected the callback to be split, got {chunks}" + ); + } // // Exercises the real SDK path end to end: provision (boot VM + create // container) → start → exec → stop → deprovision → refcount back to 0. It From dac74ad5386f0dab27860d03d9ca02b9a10e85cc Mon Sep 17 00:00:00 2001 From: Soham Das Date: Wed, 12 Aug 2026 13:07:12 -0700 Subject: [PATCH 3/7] WSLc 2c review: non-blocking callback + biased-toward-done terminal Address the two blocking review findings on the live exec-output stream: - enqueue_output now uses try_send with an overflowed latch instead of blocking_send, so the WSLc SDK I/O callback thread (which also delivers the exit callback) is never parked when a slow pipe client stops draining. - write_exec_result biases the select toward the done branch so a completed or leaked exec always writes its terminal frame; a latched overflow turns a clean Exit into a truncation Error via terminal_frame. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: f3ae1af2-7b79-4340-a5ce-a5402e7ede3d --- .../wslc/daemon/src/control_server.rs | 202 +++++++++++++++--- .../wslc/daemon/src/session_manager.rs | 77 +++++-- 2 files changed, 233 insertions(+), 46 deletions(-) diff --git a/src/backends/wslc/daemon/src/control_server.rs b/src/backends/wslc/daemon/src/control_server.rs index 09256bb43..0612e3449 100644 --- a/src/backends/wslc/daemon/src/control_server.rs +++ b/src/backends/wslc/daemon/src/control_server.rs @@ -17,7 +17,7 @@ //! connect to the control plane. use std::ffi::c_void; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -446,7 +446,11 @@ async fn write_exec_result( pipe: &mut S, admission: Result, ) -> Result<()> { - let ExecStream { done, mut output } = match admission { + let ExecStream { + done, + mut output, + overflowed, + } = match admission { Ok(stream) => stream, Err(e) => { write_frame(pipe, &worker_err_response(e)).await?; @@ -455,42 +459,40 @@ async fn write_exec_result( }; write_frame(pipe, &DaemonResponse::Ok).await?; - // Pump live output until the run completes, then write the terminal frame. + // Pump live output until the run completes, then write exactly one terminal + // frame. The select is **biased toward `done`** so a completed run always + // makes progress to termination: on the deliberate kill/leak path the sink's + // sender stays alive and can keep producing forever, and a biased-toward- + // output loop would let that continuously non-empty queue starve the ready + // `done` branch — the terminal frame would never be written and the client + // would stream forever. While the run is in flight `done` is not ready, so + // the fall-through streams live output as it arrives. // - // Two completion signals, because the output channel does not always close: - // - Normal path: the worker's `IoContext` (and the sink's sender) drop when - // the run returns, so `output` closes *before* `done` fires; the biased - // select drains every queued chunk first, then observes the close. - // - Kill/leak path: a container killed without its exit callback leaks the - // `IoContext` (deliberately), so the sink's sender never drops and - // `output` never closes. There `done` is the only completion signal, so - // we drain whatever is already queued and stop. + // Once `done` resolves we `close()` the receiver (so any leaked producer can + // enqueue nothing further), drain only the already-queued bounded tail with + // non-blocking `try_recv`, and terminate. On the normal path the run has + // already stopped producing, so that tail is exactly the remaining real + // output; the sink's sender also drops as the run returns, so `output` may + // instead close first — the `None` arm handles that and awaits the exit code. let mut done = done; let terminal = loop { tokio::select! { biased; - chunk = output.recv() => match chunk { - Some(chunk) => { - write_frame(pipe, &output_frame(chunk)).await?; - } - // Senders dropped: the run has completed and every chunk is - // flushed. Await the (already-resolved) exit code. - None => break exit_terminal(done.await), - }, result = &mut done => { - // Run finished but the output channel may still be open (leak - // path: a killed container whose exit callback never fired keeps - // the sink's sender alive). Close the receiver first so any - // leaked callback can enqueue nothing further — otherwise a - // container still producing output could keep `try_recv` - // returning chunks forever and the terminal frame would never be - // written. Then flush what is already queued and stop. output.close(); while let Ok(chunk) = output.try_recv() { write_frame(pipe, &output_frame(chunk)).await?; } - break exit_terminal(result); + break terminal_frame(result, &overflowed); } + chunk = output.recv() => match chunk { + Some(chunk) => { + write_frame(pipe, &output_frame(chunk)).await?; + } + // Senders dropped before `done` fired (normal path): the run has + // completed and every chunk is flushed. Await the exit code. + None => break terminal_frame(done.await, &overflowed), + }, } }; write_frame(pipe, &terminal).await?; @@ -505,6 +507,27 @@ fn output_frame((kind, data): (OutStream, Vec)) -> StreamFrame { } } +/// Choose the exec's terminal [`StreamFrame`]. A latched `overflowed` means the +/// sink had to drop live output because the client did not drain the daemon's +/// bounded queue fast enough, so a would-be clean [`StreamFrame::Exit`] is +/// reported as a truncation [`StreamFrame::Error`] instead — the client must not +/// treat a short stream as a successful, complete run. A genuine run failure +/// (already an `Error`) is strictly more informative and passes through +/// unchanged. +fn terminal_frame( + result: Result, oneshot::error::RecvError>, + overflowed: &AtomicBool, +) -> StreamFrame { + match exit_terminal(result) { + StreamFrame::Exit { .. } if overflowed.load(Ordering::Relaxed) => StreamFrame::Error { + message: "WSLc: live output was truncated — the client did not read the exec stream \ + fast enough and the daemon's bounded output queue overflowed" + .to_string(), + }, + other => other, + } +} + /// Map a completed exec's result (or a dropped completion channel) to its /// terminal [`StreamFrame`]. fn exit_terminal( @@ -572,7 +595,11 @@ mod tests { ) -> Result { let (tx, output) = mpsc::channel(16); drop(tx); - Ok(ExecStream { done, output }) + Ok(ExecStream { + done, + output, + overflowed: Arc::new(AtomicBool::new(false)), + }) } /// A rejected admission (unknown sandbox) round-trips as a single typed @@ -695,6 +722,7 @@ mod tests { Ok(ExecStream { done: done_rx, output, + overflowed: Arc::new(AtomicBool::new(false)), }), ) .await @@ -753,6 +781,7 @@ mod tests { Ok(ExecStream { done: done_rx, output, + overflowed: Arc::new(AtomicBool::new(false)), }), ) .await @@ -785,4 +814,121 @@ mod tests { ); assert!(read_frame::<_, StreamFrame>(&mut client).await.is_err()); } + + /// Starvation regression: completion (`done`) must terminate the stream even + /// when a leaked producer keeps the queue continuously non-empty. A task + /// enqueues forever while `done` is already resolved; because the select is + /// biased toward `done`, the handler closes the receiver, drains the bounded + /// tail, and writes the terminal frame instead of streaming the infinite + /// producer forever. (A biased-toward-output loop would hang here.) + #[tokio::test] + async fn continuous_producer_does_not_starve_terminal() { + let (mut server, mut client) = duplex(64 * 1024); + let (done_tx, done_rx) = oneshot::channel::>(); + // Small queue so a fast producer keeps it perpetually non-empty. + let (out_tx, output) = mpsc::channel(4); + + // Completion is already ready before the handler runs. + done_tx.send(Ok(5)).unwrap(); + // A producer that keeps enqueuing (the leaked `IoContext` on the kill + // path). It races the handler; `send` errors once the receiver closes, + // ending the task — so this never leaks past the test. + let producer = tokio::spawn(async move { + loop { + if out_tx + .send((OutStream::Stdout, b"x".to_vec())) + .await + .is_err() + { + break; + } + } + }); + + // Must complete (not hang). A generous timeout guards a regression to a + // starving loop, which would never return. + let result = tokio::time::timeout( + Duration::from_secs(5), + write_exec_result( + &mut server, + Ok(ExecStream { + done: done_rx, + output, + overflowed: Arc::new(AtomicBool::new(false)), + }), + ), + ) + .await + .expect("write_exec_result must terminate, not starve on continuous output"); + result.unwrap(); + drop(server); + let _ = producer.await; + + // The stream ends with a single `Exit` terminal after some prefix of + // `Stdout` frames; it must not stream indefinitely. + let admit: DaemonResponse = read_frame(&mut client).await.unwrap(); + assert_eq!(admit, DaemonResponse::Ok); + let mut saw_exit = false; + while let Ok(frame) = read_frame::<_, StreamFrame>(&mut client).await { + match frame { + StreamFrame::Stdout { .. } => {} + StreamFrame::Exit { code } => { + assert_eq!(code, 5); + saw_exit = true; + break; + } + other => panic!("unexpected frame: {other:?}"), + } + } + assert!(saw_exit, "the stream must end with an Exit terminal"); + } + + /// Overflow regression: when the sink latched `overflowed` (it dropped live + /// output because the client did not drain fast enough), the terminal frame + /// must be a truncation `Error` — never a clean `Exit` the client would + /// mistake for a complete run. + #[tokio::test] + async fn overflow_yields_truncation_error_terminal() { + let (mut server, mut client) = duplex(64 * 1024); + let (done_tx, done_rx) = oneshot::channel::>(); + let (out_tx, output) = mpsc::channel(16); + + // Some output made it through before the drop, then a clean exit, but the + // sink signalled truncation. + out_tx + .try_send((OutStream::Stdout, b"partial".to_vec())) + .unwrap(); + done_tx.send(Ok(0)).unwrap(); + drop(out_tx); + let overflowed = Arc::new(AtomicBool::new(true)); + + write_exec_result( + &mut server, + Ok(ExecStream { + done: done_rx, + output, + overflowed, + }), + ) + .await + .unwrap(); + drop(server); + + let admit: DaemonResponse = read_frame(&mut client).await.unwrap(); + assert_eq!(admit, DaemonResponse::Ok); + assert_eq!( + read_frame::<_, StreamFrame>(&mut client).await.unwrap(), + StreamFrame::Stdout { + data: b"partial".to_vec() + } + ); + let terminal: StreamFrame = read_frame(&mut client).await.unwrap(); + match terminal { + StreamFrame::Error { message } => { + assert!(message.contains("truncated"), "message was {message:?}"); + } + other => panic!("expected a truncation Error terminal, got {other:?}"), + } + assert!(read_frame::<_, StreamFrame>(&mut client).await.is_err()); + } } diff --git a/src/backends/wslc/daemon/src/session_manager.rs b/src/backends/wslc/daemon/src/session_manager.rs index 19d542240..4da8f3ed7 100644 --- a/src/backends/wslc/daemon/src/session_manager.rs +++ b/src/backends/wslc/daemon/src/session_manager.rs @@ -25,8 +25,11 @@ //! output flows over the sink. (Client `Stdin` forwarding is a later fill-in.) use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; use anyhow::Result; +use tokio::sync::mpsc::error::TrySendError; use tokio::sync::{mpsc, oneshot}; use wslc_common::container_steps::{self, OutStream, OutputSink, ProcessSettings}; @@ -149,8 +152,11 @@ pub type OutputChunk = (OutStream, Vec); /// unbounded) so a container emitting output faster than a slow client drains it /// cannot grow the queue without limit and OOM the persistent daemon, taking /// down every sandbox it owns — the same hazard the capture-buffer cap guards -/// against. When the queue is full the sink applies backpressure (see -/// [`SessionHandle::exec`]) rather than dropping bytes. +/// against. When the queue is full the sink does **not** block (see +/// [`enqueue_output`]): it latches a truncation flag and drops further bytes so +/// the SDK callback thread — which also delivers the process-exit callback — is +/// never parked. Stalling that thread could otherwise block exit delivery and +/// wedge teardown for every sandbox sharing the daemon. const LIVE_OUTPUT_CHANNEL_CAPACITY: usize = 256; /// Max bytes per enqueued live-output chunk. A single SDK callback can deliver @@ -163,22 +169,50 @@ const LIVE_OUTPUT_MAX_CHUNK_BYTES: usize = 64 * 1024; /// Enqueue an SDK output callback, splitting it into `LIVE_OUTPUT_MAX_CHUNK_BYTES` /// pieces so each queue entry and its resulting frame stay bounded regardless of -/// the callback's buffer size. Stops early once the receiver is gone (client -/// left / leak-path `close()`), so a leaked callback never parks forever. -fn enqueue_output(tx: &mpsc::Sender, kind: OutStream, bytes: &[u8]) { +/// the callback's buffer size. +/// +/// This runs **synchronously on the SDK's I/O callback thread**, which also +/// delivers the process-exit callback. It must therefore never block: a +/// [`try_send`](mpsc::Sender::try_send) that finds the bounded queue full does +/// **not** apply backpressure (that would park this thread and could deadlock +/// exit delivery and teardown for every sandbox on the daemon). Instead it +/// latches `overflowed` and stops enqueuing; the pipe handler turns a latched +/// overflow into a terminal `Error` frame so the client sees an explicit +/// truncation rather than a silently short stream. Once latched, subsequent +/// callbacks return immediately so the client receives a clean truncated prefix +/// rather than a gapped stream. A closed receiver (client left / leak-path +/// `close()`) likewise stops enqueuing. +fn enqueue_output( + tx: &mpsc::Sender, + overflowed: &AtomicBool, + kind: OutStream, + bytes: &[u8], +) { + if overflowed.load(Ordering::Relaxed) { + return; + } for chunk in bytes.chunks(LIVE_OUTPUT_MAX_CHUNK_BYTES) { - if tx.blocking_send((kind, chunk.to_vec())).is_err() { - break; + match tx.try_send((kind, chunk.to_vec())) { + Ok(()) => {} + Err(TrySendError::Full(_)) => { + overflowed.store(true, Ordering::Relaxed); + return; + } + Err(TrySendError::Closed(_)) => return, } } } -/// An admitted exec: the completion receiver (the run's exit code) plus the -/// live-output receiver the pipe handler drains into `Stdout`/`Stderr` frames. +/// An admitted exec: the completion receiver (the run's exit code), the +/// live-output receiver the pipe handler drains into `Stdout`/`Stderr` frames, +/// and the `overflowed` latch the sink sets when it had to drop output because a +/// slow client let the bounded queue fill. The pipe handler reports a set latch +/// as a terminal `Error` frame. #[derive(Debug)] pub struct ExecStream { pub done: oneshot::Receiver>, pub output: mpsc::Receiver, + pub overflowed: Arc, } /// A cheap, clonable handle async tasks use to drive the worker thread. @@ -215,15 +249,18 @@ impl SessionHandle { let (done, done_rx) = oneshot::channel(); let (stream_tx, output) = mpsc::channel::(LIVE_OUTPUT_CHANNEL_CAPACITY); // The sink is invoked from the SDK's I/O callback threads (native OS - // threads, never the async runtime). `blocking_send` on the bounded - // channel parks that thread when the queue is full, propagating - // backpressure to the container's stdout/stderr pipe — the container - // blocks producing until the pipe handler drains, so memory stays - // bounded and no output bytes are dropped. A closed receiver (client - // gone, or the leak-path `close()` in the pipe handler) makes the send - // return `Err`, so a leaked callback never parks forever. + // threads, never the async runtime). Those same threads deliver the + // process-exit callback, so the sink must never block: [`enqueue_output`] + // uses a non-blocking `try_send` and, on a full queue, latches + // `overflowed` and drops further bytes instead of parking the thread. + // Memory stays bounded by the channel capacity; a slow/non-reading + // client can no longer wedge exit delivery or teardown for every sandbox + // on the daemon. The pipe handler reports a set latch as a terminal + // `Error` frame so truncation is explicit rather than silent. + let overflowed = Arc::new(AtomicBool::new(false)); + let sink_overflowed = Arc::clone(&overflowed); let sink: OutputSink = Box::new(move |kind, bytes| { - enqueue_output(&stream_tx, kind, bytes); + enqueue_output(&stream_tx, &sink_overflowed, kind, bytes); }); self.send(WorkerCommand::Exec { config, @@ -235,6 +272,7 @@ impl SessionHandle { Ok(ExecStream { done: done_rx, output, + overflowed, }) } @@ -797,7 +835,10 @@ mod tests { let payload = vec![b'x'; LIVE_OUTPUT_MAX_CHUNK_BYTES * 3 + 7]; let producer = tokio::task::spawn_blocking({ let payload = payload.clone(); - move || enqueue_output(&tx, OutStream::Stdout, &payload) + move || { + let overflowed = AtomicBool::new(false); + enqueue_output(&tx, &overflowed, OutStream::Stdout, &payload) + } }); let mut reassembled = Vec::new(); From 7ac155999bc083836a0231c15942928520d8a44e Mon Sep 17 00:00:00 2001 From: Soham Das Date: Wed, 12 Aug 2026 14:13:04 -0700 Subject: [PATCH 4/7] Addressed PR comments Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: f3ae1af2-7b79-4340-a5ce-a5402e7ede3d --- src/Cargo.lock | 1 + src/backends/wslc/common/Cargo.toml | 1 + .../wslc/common/src/container_steps.rs | 15 +++ src/backends/wslc/common/src/daemon_client.rs | 11 ++- .../wslc/common/src/daemon_protocol.rs | 69 +++++++++++++- .../wslc/common/src/policy_mapping.rs | 42 +++++++++ src/backends/wslc/common/src/state_aware.rs | 92 +++++++++---------- .../wslc/common/src/wsl_container_runner.rs | 48 +++------- .../wslc/daemon/src/control_server.rs | 2 +- tests/configs/wslc_state_aware_exec_drip.json | 2 +- tests/scripts/run_wslc_state_aware_tests.ps1 | 25 +++-- 11 files changed, 209 insertions(+), 99 deletions(-) diff --git a/src/Cargo.lock b/src/Cargo.lock index 43fa7e83c..fda78c92a 100644 --- a/src/Cargo.lock +++ b/src/Cargo.lock @@ -3120,6 +3120,7 @@ name = "wslc_common" version = "0.7.0" dependencies = [ "anyhow", + "base64", "libloading", "serde", "serde_json", diff --git a/src/backends/wslc/common/Cargo.toml b/src/backends/wslc/common/Cargo.toml index c130b9890..7d69b232b 100644 --- a/src/backends/wslc/common/Cargo.toml +++ b/src/backends/wslc/common/Cargo.toml @@ -13,6 +13,7 @@ wxc_common = { workspace = true } windows = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } +base64 = { workspace = true } anyhow = { workspace = true } uuid = { workspace = true } tar = "0.4" diff --git a/src/backends/wslc/common/src/container_steps.rs b/src/backends/wslc/common/src/container_steps.rs index e4c703ff6..b0be01703 100644 --- a/src/backends/wslc/common/src/container_steps.rs +++ b/src/backends/wslc/common/src/container_steps.rs @@ -121,6 +121,21 @@ pub enum OutStream { /// container's output to the client as bytes arrive; paths that only need the /// final captured blob (one-shot, detached init) leave it unset. It receives /// the full callback bytes, independent of the capped buffers' truncation. +/// +/// **Two distinct live-output architectures — why they don't share plumbing.** +/// The one-shot runner streams via `OutputMode::Stream` (in `wsl_container_runner`) +/// plus a synchronous in-process `stream_pair`/`StreamReader` (in `stream_buffer`), +/// drained in the same process (caller pipes, or an `inherit_pump` thread). The +/// daemon cannot reuse that: it must relay output across a named pipe to a +/// *separate* client process through an async (tokio) control server, so it +/// bridges the synchronous SDK callback thread to the async pipe writer with a +/// bounded mpsc channel (see `session_manager::enqueue_output`). Their +/// backpressure invariants also differ deliberately: the one-shot `StreamReader` +/// is drained synchronously in-process, whereas this callback **must never +/// block** — the same SDK thread also delivers the process-exit callback, so the +/// daemon path uses a non-blocking `try_send` that drops on overflow rather than +/// stalling teardown. Keep the two paths separate for those reasons; share only +/// the leaf primitives ([`OutStream`], the capped capture buffers). pub type OutputSink = Box; /// Shared buffer for capturing process I/O via SDK callbacks. Fields are diff --git a/src/backends/wslc/common/src/daemon_client.rs b/src/backends/wslc/common/src/daemon_client.rs index 707af3e14..a3d0418b3 100644 --- a/src/backends/wslc/common/src/daemon_client.rs +++ b/src/backends/wslc/common/src/daemon_client.rs @@ -289,8 +289,15 @@ impl DaemonClient { /// code and fully-buffered output. /// /// A convenience wrapper over [`exec_streaming`](Self::exec_streaming) that - /// accumulates every stdout/stderr chunk into an [`ExecResult`]. Prefer - /// `exec_streaming` when output should be relayed live rather than buffered. + /// accumulates every stdout/stderr chunk into an [`ExecResult`]. + /// + /// **Unbounded capture — trusted/bounded output only.** This buffers the + /// entire stream in memory with no cap, so container-controlled output can + /// exhaust the caller's process memory. It is intended for tests and callers + /// that already know the command's output is small and bounded. Any path that + /// relays or handles live/untrusted output must use + /// [`exec_streaming`](Self::exec_streaming) (the production state-aware runner + /// does), which buffers nothing and lets the caller apply its own policy. pub fn exec(&self, config: ExecConfig) -> DaemonResult { let mut stdout = Vec::new(); let mut stderr = Vec::new(); diff --git a/src/backends/wslc/common/src/daemon_protocol.rs b/src/backends/wslc/common/src/daemon_protocol.rs index ceae696fa..def14663d 100644 --- a/src/backends/wslc/common/src/daemon_protocol.rs +++ b/src/backends/wslc/common/src/daemon_protocol.rs @@ -37,7 +37,7 @@ pub const MAX_FRAME_SIZE: usize = 16 * 1024 * 1024; /// from the same build, so in normal operation both sides always match; the /// version guards against a stale daemon left running by a different mxc /// install. Bump only for incompatible changes to framing or message shape. -pub const PROTOCOL_VERSION: u32 = 1; +pub const PROTOCOL_VERSION: u32 = 2; // --------------------------------------------------------------------------- // Per-phase config structs (daemon-internal; NOT the public wire schema) @@ -194,16 +194,32 @@ pub enum ErrKind { /// [`DaemonRequest::Exec`]). Client→daemon carries [`StreamFrame::Stdin`]; /// daemon→client carries [`StreamFrame::Stdout`] / [`StreamFrame::Stderr`] and /// a terminal [`StreamFrame::Exit`] (or [`StreamFrame::Error`]). +/// +/// The raw byte payloads are base64-encoded on the wire (see [`base64_bytes`]). +/// serde_json renders a `Vec` as a JSON array of decimal integers (`[104, +/// 105, ...]`), roughly **4 bytes of wire per payload byte**, and since +/// [`MAX_FRAME_SIZE`] is measured against the *encoded* frame that would cut +/// effective throughput to ~1/4. base64 is ~1.33x instead, while keeping the +/// single uniform JSON framing (no separate binary path to test). #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum StreamFrame { /// Client→daemon: bytes to write to the process stdin. An empty payload /// signals stdin EOF. - Stdin { data: Vec }, + Stdin { + #[serde(with = "base64_bytes")] + data: Vec, + }, /// Daemon→client: bytes read from the process stdout. - Stdout { data: Vec }, + Stdout { + #[serde(with = "base64_bytes")] + data: Vec, + }, /// Daemon→client: bytes read from the process stderr. - Stderr { data: Vec }, + Stderr { + #[serde(with = "base64_bytes")] + data: Vec, + }, /// Daemon→client: terminal frame; the process exited with `code`. No more /// stream frames follow. Exit { code: i32 }, @@ -211,6 +227,23 @@ pub enum StreamFrame { Error { message: String }, } +/// serde adapter that (de)serializes a `Vec` as a base64 string rather than +/// a JSON integer array. Used for the [`StreamFrame`] byte payloads to keep the +/// exec data phase compact over the wire. +mod base64_bytes { + use base64::{engine::general_purpose::STANDARD, Engine as _}; + use serde::{Deserialize, Deserializer, Serializer}; + + pub fn serialize(bytes: &[u8], serializer: S) -> Result { + serializer.serialize_str(&STANDARD.encode(bytes)) + } + + pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result, D::Error> { + let encoded = String::deserialize(deserializer)?; + STANDARD.decode(&encoded).map_err(serde::de::Error::custom) + } +} + // --------------------------------------------------------------------------- // Framing helpers (generic over any serde message type) // --------------------------------------------------------------------------- @@ -360,6 +393,34 @@ mod tests { }); } + #[test] + fn stream_frame_bytes_roundtrip_including_non_utf8() { + // Arbitrary bytes (incl. non-UTF8, NUL, 0xFF) must survive the base64 + // wire encoding exactly. + let raw: Vec = (0u16..=255).map(|b| b as u8).collect(); + roundtrip(StreamFrame::Stdout { data: raw.clone() }); + roundtrip(StreamFrame::Stderr { data: raw.clone() }); + roundtrip(StreamFrame::Stdin { data: raw }); + } + + #[test] + fn stream_frame_payload_is_base64_not_integer_array() { + // Guards the compact wire encoding: the payload is a base64 string, not + // serde_json's default `[104, 105, ...]` integer array (~4x larger). + let json = serde_json::to_string(&StreamFrame::Stdout { + data: b"hi".to_vec(), + }) + .unwrap(); + assert!( + json.contains("\"data\":\"aGk=\""), + "unexpected wire: {json}" + ); + assert!( + !json.contains('['), + "payload must not be an integer array: {json}" + ); + } + #[test] fn network_mode_defaults_to_none() { assert_eq!(NetworkMode::default(), NetworkMode::None); diff --git a/src/backends/wslc/common/src/policy_mapping.rs b/src/backends/wslc/common/src/policy_mapping.rs index d96b4e4b5..f177b332b 100644 --- a/src/backends/wslc/common/src/policy_mapping.rs +++ b/src/backends/wslc/common/src/policy_mapping.rs @@ -11,6 +11,8 @@ use crate::wslc_bindings::WslcContainerNetworkingMode; use wxc_common::filesystem_canonical::{canonicalize_allowing_absent_tail, PathCanonical}; +use wxc_common::logger::Logger; +use wxc_common::models::{ContainerPolicy, ExecutionRequest}; /// A resolved volume mount ready to be passed to `WslcSetContainerSettingsVolumes`. #[derive(Debug, Clone, PartialEq, Eq)] @@ -276,6 +278,46 @@ pub fn validate_denied_path_overlap( ) } +/// The complete WSLc provision-time filesystem-policy gate, shared by the +/// one-shot runner (`wsl_container_runner::start_container`) and the state-aware +/// provision path (`state_aware::build_provision_config`). Keeping the three +/// ordered steps in one place stops the two runners drifting apart when a policy +/// check or its ordering changes. +/// +/// The steps, in the order both paths must run them: +/// 1. **Object-identity normalization (D6)** — tighten rw/ro/denied aliases of +/// the same host object to the strictest intent (deny > ro > rw) via +/// [`wxc_common::filesystem_object::normalize_object_conflicts`]. A path moved +/// to `denied` is simply not mounted (unmounted = invisible). +/// 2. **Delegation (D3)** — reject any path the invoking user cannot access via +/// [`wxc_common::filesystem_access::check_delegation`], evaluated against the +/// already-tightened intents so the sandbox never gains access the caller lacks. +/// 3. **Denied-path overlap** — reject a `denied` entry nested under a still-mounted +/// parent (WSLc's flat volume surface has no overlay primitive), again against +/// the tightened lists, so a deny that only becomes nested after normalization +/// cannot slip through unenforced. +/// +/// Returns the tightened policy when normalization changed something (the caller +/// rebuilds the request around it) or `None` when the policy was already +/// conflict-free. Any failure is returned as a `String`; callers map it to their +/// own error envelope (`ScriptResponse` / `MxcError`). Normalization diagnostics +/// are written to `logger`; callers decide how to surface them. +pub fn apply_provision_policy_gate( + request: &ExecutionRequest, + logger: &mut Logger, +) -> Result, String> { + let normalized = + wxc_common::filesystem_object::normalize_object_conflicts(&request.policy, logger)?; + let effective = normalized.as_ref().unwrap_or(&request.policy); + wxc_common::filesystem_access::check_delegation(effective)?; + validate_denied_path_overlap( + &effective.readwrite_paths, + &effective.readonly_paths, + &effective.denied_paths, + )?; + Ok(normalized) +} + /// Overlap message for a denied path nested under a mounted parent. fn overlap_error(denied: &str, mounted: &str, list_name: &str, via_alias: bool) -> String { let lead = if via_alias { diff --git a/src/backends/wslc/common/src/state_aware.rs b/src/backends/wslc/common/src/state_aware.rs index 2d8b5de5c..9039e5aff 100644 --- a/src/backends/wslc/common/src/state_aware.rs +++ b/src/backends/wslc/common/src/state_aware.rs @@ -153,14 +153,20 @@ impl StatefulSandboxBackend for WslcStateAwareRunner { let client = connect_daemon()?; - // Relay each chunk to our own stdio as it arrives, flushing every chunk - // so newline-free progress output reaches the consumer promptly. + // Relay each chunk to our own stdio as it arrives. Hold the stdout/stderr + // locks for the whole relay so we don't reacquire the handle per chunk, + // and coalesce flushes: `std::io::Stdout`/`Stderr` are line-buffered, so + // newline-terminated output already reaches the consumer promptly; we only + // force a flush for a chunk that does *not* end in a newline (progress + // output — prompts, spinners) so it isn't stranded in the line buffer. + // This avoids a flush syscall per bulk chunk while preserving low latency. // Best-effort: a failed local write must not mask the container's exit code. - let mut stdout = std::io::stdout(); - let mut stderr = std::io::stderr(); - - let exit_code = client - .exec_streaming( + let stdout = std::io::stdout(); + let stderr = std::io::stderr(); + let exit_code = { + let mut out = stdout.lock(); + let mut err = stderr.lock(); + let result = client.exec_streaming( ExecConfig { sandbox_id: sandbox_id.to_string(), script_code: request.script_code.clone(), @@ -170,19 +176,26 @@ impl StatefulSandboxBackend for WslcStateAwareRunner { }, |stream, bytes| match stream { OutStream::Stdout => { - let _ = stdout.write_all(bytes); - let _ = stdout.flush(); + let _ = out.write_all(bytes); + if bytes.last() != Some(&b'\n') { + let _ = out.flush(); + } } OutStream::Stderr => { - let _ = stderr.write_all(bytes); - let _ = stderr.flush(); + let _ = err.write_all(bytes); + if bytes.last() != Some(&b'\n') { + let _ = err.flush(); + } } }, - ) - .map_err(map_daemon_error)?; - - let _ = stdout.flush(); - let _ = stderr.flush(); + ); + let _ = out.flush(); + let _ = err.flush(); + // Drop the locks before mapping the error so error conversion never + // contends with the writers we just held. + drop((out, err)); + result.map_err(map_daemon_error)? + }; Ok(ExecHandle { stdout: null_pipe_handle(), @@ -304,12 +317,12 @@ fn build_provision_config( .unwrap_or_else(|| DEFAULT_IMAGE.to_string()); let image_tar_path = config.and_then(|c| c.image_tar_path); - // Object-identity normalization (D6) + delegation (D3), mirroring the - // one-shot runner: tighten rw/ro/denied aliases of the same host object to - // the strictest intent, then reject any path the caller cannot access. The - // daemon must mount the tightened policy, so a writable alias of a readonly - // object never leaks and a persistent daemon never mounts a path the phase - // caller could not delegate. + // WSLc provision-time filesystem-policy gate (D6 normalization → D3 + // delegation → denied-path overlap), shared verbatim with the one-shot + // runner via `policy_mapping::apply_provision_policy_gate`. The daemon must + // mount the tightened policy, so a writable alias of a readonly object never + // leaks and a persistent daemon never mounts a path the phase caller could + // not delegate. let normalized = normalize_and_check_delegation(request)?; let normalized_request; let request = match normalized { @@ -323,19 +336,6 @@ fn build_provision_config( None => request, }; - // Re-run the denied-path overlap check on the *normalized* lists, mirroring - // the one-shot runner order (wsl_container_runner.rs). Normalization can - // tighten a mounted alias into `deniedPaths`; if that alias is nested under - // another mounted parent the raw pre-normalization check in - // `validate_provision_policy` cannot see it, and the daemon would mount the - // parent leaving the deny reachable through it. Fail closed here. - crate::policy_mapping::validate_denied_path_overlap( - &request.policy.readwrite_paths, - &request.policy.readonly_paths, - &request.policy.denied_paths, - ) - .map_err(MxcError::policy_validation)?; - let volumes = build_daemon_volumes(request)?; let network = map_network(request); Ok(ProvisionConfig { @@ -346,27 +346,25 @@ fn build_provision_config( }) } -/// Object-identity normalization (D6) then delegation check (D3) for the -/// provision phase, mirroring the one-shot runner order. Returns the tightened -/// policy when aliasing required a change, else `None`; either check maps its -/// `String` error to a `policy_validation` envelope. +/// State-aware adapter over the shared WSLc provision policy gate +/// ([`crate::policy_mapping::apply_provision_policy_gate`]): runs the full +/// three-step gate (D6 normalization → D3 delegation → denied-path overlap), +/// buffering normalization diagnostics and surfacing them on stderr (stdout +/// carries the phase envelope), and maps the gate's `String` error to a +/// `policy_validation` [`MxcError`]. Returns the tightened policy when +/// normalization changed something, else `None`. fn normalize_and_check_delegation( request: &ExecutionRequest, ) -> Result, MxcError> { let mut logger = Logger::new(Mode::Buffer); - let normalized = - wxc_common::filesystem_object::normalize_object_conflicts(&request.policy, &mut logger) - .map_err(MxcError::policy_validation)?; + let result = crate::policy_mapping::apply_provision_policy_gate(request, &mut logger); // Surface any normalization notes (policy tightening / unresolved paths) on - // stderr rather than dropping the buffer: stdout carries the phase envelope, - // so these diagnostics must not go there. + // stderr even when the gate then fails, rather than dropping the buffer. let notes = logger.get_buffer(); if !notes.is_empty() { eprint!("{notes}"); } - let policy = normalized.as_ref().unwrap_or(&request.policy); - wxc_common::filesystem_access::check_delegation(policy).map_err(MxcError::policy_validation)?; - Ok(normalized) + result.map_err(MxcError::policy_validation) } /// Build daemon volume mounts from the request's filesystem policy. Overlapping diff --git a/src/backends/wslc/common/src/wsl_container_runner.rs b/src/backends/wslc/common/src/wsl_container_runner.rs index ab4e14e95..9be3f570d 100644 --- a/src/backends/wslc/common/src/wsl_container_runner.rs +++ b/src/backends/wslc/common/src/wsl_container_runner.rs @@ -1279,17 +1279,14 @@ impl WSLContainerRunner { ) -> Result { let _ = writeln!(logger, "[WSLC] Starting WSL Container runner"); - // Object-based FS-policy normalization (D6): tighten aliases of the same - // host object to the strictest intent (deny > ro > rw) before mapping to - // volume mounts. See `wxc_common::filesystem_object`. (A path moved to - // `denied` is simply not mounted by WSLC — unmounted = invisible.) Only - // clone the request when an aliasing conflict actually needs tightening; - // an unresolvable path with deniedPaths present fails closed. + // WSLc provision-time filesystem-policy gate (D6 normalization → D3 + // delegation → denied-path overlap), shared verbatim with the + // state-aware provision path via `policy_mapping::apply_provision_policy_gate` + // so the two runners cannot drift. Only clone the request when + // normalization actually tightened something; any failure is surfaced on + // the streaming logger and returned as a `ScriptResponse` error. let normalized; - let request = match wxc_common::filesystem_object::normalize_object_conflicts( - &request.policy, - logger, - ) { + let request = match policy_mapping::apply_provision_policy_gate(request, logger) { Ok(Some(policy)) => { normalized = ExecutionRequest { policy, @@ -1298,34 +1295,11 @@ impl WSLContainerRunner { &normalized } Ok(None) => request, - Err(msg) => return Err(ScriptResponse::error(&msg)), + Err(msg) => { + let _ = writeln!(logger, "[WSLC] {}", msg); + return Err(ScriptResponse::error(&msg)); + } }; - // Delegation check (D3): reject any policy path the invoking user cannot - // access, so the sandbox never gains access the caller lacks. Runs AFTER - // object normalization so it is evaluated against the already-tightened - // intents. On Windows this covers directory readwrite paths (the common - // WSLC case). - if let Err(msg) = wxc_common::filesystem_access::check_delegation(&request.policy) { - return Err(ScriptResponse::error(&msg)); - } - - // Denied-path overlap validation: WSLC's flat volume-mount surface has no - // overlay primitive, so a deniedPaths entry nested under a mounted - // (readwrite/readonly) parent cannot be masked and would stay accessible - // through the parent mount. A lexical tier catches `..`/case/whole-drive - // spellings; a canonicalizing tier resolves symlink/junction/8.3/`\\?\` - // aliases (including a not-yet-created deny under an aliased parent) and - // fails closed on unresolvable paths. Reject such configs rather than - // silently leaving the subtree exposed. Runs after object normalization - // so it sees the already-tightened intents. - if let Err(msg) = policy_mapping::validate_denied_path_overlap( - &request.policy.readwrite_paths, - &request.policy.readonly_paths, - &request.policy.denied_paths, - ) { - let _ = writeln!(logger, "[WSLC] {}", msg); - return Err(ScriptResponse::error(&msg)); - } // -- Init: COM + SDK + preflight -- let sdk = Self::init_and_load_sdk(logger)?; diff --git a/src/backends/wslc/daemon/src/control_server.rs b/src/backends/wslc/daemon/src/control_server.rs index 0612e3449..910ce7d0b 100644 --- a/src/backends/wslc/daemon/src/control_server.rs +++ b/src/backends/wslc/daemon/src/control_server.rs @@ -425,7 +425,7 @@ async fn handle_client(mut pipe: NamedPipeServer, session: SessionHandle) -> Res /// live output streaming depends on), so `WslcGetProcessIOHandle(STDIN)` is /// unavailable. Piped stdin would require a handle-mode rearchitecture (no /// callbacks; `ReadFile` threads for stdout/stderr + `WriteFile` for stdin) and -/// is deferred to Tier 2 (see the 2c plan). +/// is deferred; stdin forwarding is tracked in issue #804. async fn handle_exec( mut pipe: NamedPipeServer, session: SessionHandle, diff --git a/tests/configs/wslc_state_aware_exec_drip.json b/tests/configs/wslc_state_aware_exec_drip.json index 7ca158cdb..7a6830130 100644 --- a/tests/configs/wslc_state_aware_exec_drip.json +++ b/tests/configs/wslc_state_aware_exec_drip.json @@ -3,7 +3,7 @@ "phase": "exec", "sandboxId": "{{SANDBOX_ID}}", "process": { - "commandLine": "sh -c 'echo DRIP-PART1; sleep 2; echo DRIP-PART2'", + "commandLine": "sh -c 'echo DRIP-PART1; sleep 3; echo DRIP-PART2'", "timeout": 30000 } } diff --git a/tests/scripts/run_wslc_state_aware_tests.ps1 b/tests/scripts/run_wslc_state_aware_tests.ps1 index 2585636f4..2f2faa034 100644 --- a/tests/scripts/run_wslc_state_aware_tests.ps1 +++ b/tests/scripts/run_wslc_state_aware_tests.ps1 @@ -419,13 +419,24 @@ try { } # A3b: LIVE DRIP -- prove output is streamed incrementally, not buffered and - # dumped at exit. The script prints PART1, sleeps 2s, then prints PART2. With - # live streaming the harness observes PART1 well before PART2 (~the sleep); - # a buffer-then-dump impl would surface both lines together at process exit - # (gap ~0). Asserting the inter-line arrival gap is the airtight liveness - # proof a content-only "stdout contains X" check cannot give. + # dumped at exit. The script prints PART1, sleeps 3s, then prints PART2. With + # live streaming the harness observes PART1 ~3s before PART2; a buffer-then- + # dump impl would surface both lines together at process exit (gap ~0). + # Asserting the inter-line arrival gap is the airtight liveness proof a + # content-only "stdout contains X" check cannot give. + # + # De-flake / tolerance: the two outcomes are cleanly separated -- buffered + # ~0s vs streamed ~$dripSleepSec. We assert the observed gap is at least + # $dripMinGapSec, chosen as roughly half the sleep so it stays far above the + # buffered-dump floor while absorbing up to ~$($dripSleepSec - $dripMinGapSec)s + # of scheduler / pipe-relay jitter (first-byte latency on PART1 shrinks the + # measured gap, so a generous margin below the full sleep is what keeps this + # from false-failing under CI load). Widen $dripSleepSec, not $dripMinGapSec, + # if a slower host is ever observed. if ($execedOk) { Run-StateAwareTest "A: exec (live drip -- incremental streaming)" { + $dripSleepSec = 3.0 + $dripMinGapSec = 1.5 $r = Invoke-StateAwareStreaming -ConfigFile 'wslc_state_aware_exec_drip.json' -SandboxId $script:sandboxId Assert-True ($r.ExitCode -eq 0) "drip exec exit code = 0" $p1 = $r.Lines | Where-Object { $_.Text -match 'DRIP-PART1' } | Select-Object -First 1 @@ -434,8 +445,8 @@ try { Assert-True ($null -ne $p2) "PART2 line observed on stdout" if ($p1 -and $p2) { $gapSec = ($p2.At - $p1.At).TotalSeconds - Assert-True ($gapSec -ge 1.0) ` - ("PART1 arrived >=1.0s before PART2 (gap {0:N2}s) -- streamed live, not buffered" -f $gapSec) + Assert-True ($gapSec -ge $dripMinGapSec) ` + ("PART1 arrived >={0:N1}s before PART2 (gap {1:N2}s, sleep {2:N1}s) -- streamed live, not buffered" -f $dripMinGapSec, $gapSec, $dripSleepSec) } } | Out-Null } From afdce45586100fc8d2171e5d7a0be2ff5c1f320e Mon Sep 17 00:00:00 2001 From: Soham Das Date: Tue, 11 Aug 2026 12:26:01 -0700 Subject: [PATCH 5/7] WSLC state-aware: TypeScript SDK per-phase policy configs, wslc: routing, and 0.8.0-alpha default --- sdk/node/README.md | 8 +- sdk/node/src/index.ts | 5 + sdk/node/src/state-aware-helper.ts | 21 ++- sdk/node/src/state-aware-types.ts | 86 +++++++++++- sdk/node/tests/unit/state-aware-types.test.ts | 112 ++++++++++++++++ sdk/node/tests/unit/state-aware.test.ts | 123 ++++++++++++++++++ 6 files changed, 350 insertions(+), 5 deletions(-) diff --git a/sdk/node/README.md b/sdk/node/README.md index 47ca684e3..0c995e7a3 100644 --- a/sdk/node/README.md +++ b/sdk/node/README.md @@ -232,7 +232,7 @@ capability names are reserved and must not be added directly to For long-lived sandboxes where you provision once, exec many times, and tear down at the end (e.g. agentic loops), use the state-aware lifecycle. -> **Backend support:** the state-aware lifecycle is currently implemented for `isolation_session` and `windows_sandbox` (both Windows-only; both still experimental, so every call must pass `{ experimental: true }`). The one-shot spawn APIs (`spawnSandbox` / `spawnSandboxFromConfig`) are the supported path for every other backend. +> **Backend support:** the state-aware lifecycle is currently implemented for `isolation_session`, `windows_sandbox`, and `wslc` (all Windows-only; all still experimental, so every call must pass `{ experimental: true }`). The one-shot spawn APIs (`spawnSandbox` / `spawnSandboxFromConfig`) are the supported path for every other backend. ```typescript import { @@ -265,6 +265,8 @@ await deprovisionSandbox(sandboxId, undefined, opts); `windows_sandbox` follows the same shape (substitute the containment string and provide `filesystem.readwritePaths` / `readonlyPaths` at provision if needed). See [`docs/windows-sandbox/windows-sandbox.md`](https://github.com/microsoft/mxc/blob/main/docs/windows-sandbox/windows-sandbox.md) for the per-phase config matrix. +`wslc` follows the same shape and needs no provision config at all (it defaults to an `alpine:latest` container with no network). Provide `filesystem.readwritePaths` / `readonlyPaths` (mounted for the sandbox's lifetime), `network.defaultPolicy: 'allow'` (a bridged container; the default `'block'` gives no network), and/or a backend-specific `image` / `imageTarPath` at provision; inject a cooperative `network.proxy: { url }` per-exec. WSLc state-aware requests default to schema `0.8.0-alpha`. See [`docs/wsl/`](https://github.com/microsoft/mxc/tree/main/docs/wsl/) for details. + **Handling failures.** Every lifecycle call rejects with a typed `MxcError`. Branch on `code` first; when the failure came from an underlying platform API, the error also carries discrete diagnostic fields rather than a prose blob: ```typescript @@ -397,10 +399,10 @@ spawnSandboxFromConfig(config, options?, workingDirectory?, env?) → IPty | Chi spawnSandbox(script, policy, options?, workingDirectory?, containerName?, env?) → IPty spawnSandboxAsync(script, policy, ...) → Promise<{ stdout, stderr, exitCode }> -// State-aware lifecycle (currently `isolation_session` and `windows_sandbox` — both Windows-only) +// State-aware lifecycle (currently `isolation_session`, `windows_sandbox`, and `wslc` — all Windows-only) // `config` on provisionSandbox is required for backends whose provision config // has a required member (isolation_session: the network acknowledgment) and -// optional otherwise (windows_sandbox). +// optional otherwise (windows_sandbox, wslc). provisionSandbox(containment, config, options?) → Promise startSandbox(sandboxId, config?, options?) → Promise execInSandbox(sandboxId, config, options?) → IPty // streaming diff --git a/sdk/node/src/index.ts b/sdk/node/src/index.ts index 4432ba7a0..d3b50fa9b 100644 --- a/sdk/node/src/index.ts +++ b/sdk/node/src/index.ts @@ -98,6 +98,11 @@ export { WindowsSandboxExecConfig, WindowsSandboxStopConfig, WindowsSandboxDeprovisionConfig, + WslcProvisionConfig, + WslcStartConfig, + WslcExecConfig, + WslcStopConfig, + WslcDeprovisionConfig, ConfigsForBackend, ProvisionConfigFor, StartConfigFor, diff --git a/sdk/node/src/state-aware-helper.ts b/sdk/node/src/state-aware-helper.ts index 0fc072981..9ecdfca3c 100644 --- a/sdk/node/src/state-aware-helper.ts +++ b/sdk/node/src/state-aware-helper.ts @@ -10,6 +10,12 @@ import { Phase, StateAwareContainmentBackend } from './state-aware-types.js'; export const STATE_AWARE_VERSION = '0.6.0-alpha'; +// WSLc's state-aware surface shipped at a later schema version than the +// isolation_session default above. It is intentionally NOT gate-locked to the +// canonical `stateAware` constant (which tracks isolation_session): the two +// backends were promoted independently. See `DEFAULT_STATE_AWARE_VERSION`. +export const WSLC_STATE_AWARE_VERSION = '0.8.0-alpha'; + // Wire-format cross-cutting fields that live at the envelope's top level. // Anything else on a per-(backend, phase) Config is backend-specific and is // nested under `experimental..`. @@ -21,12 +27,24 @@ export const CROSS_CUTTING_FIELDS = ['filesystem', 'network', 'ui', 'process'] a // declares its own `_ID_PREFIX` const here. export const ISOLATION_SESSION_ID_PREFIX = 'iso'; export const WINDOWS_SANDBOX_ID_PREFIX = 'wsb'; +export const WSLC_ID_PREFIX = 'wslc'; + +// Per-backend default schema version stamped onto an envelope when the caller +// supplies none. Each backend's state-aware surface was promoted at its own +// schema version, so the default is backend-specific rather than a single +// global constant. +const DEFAULT_STATE_AWARE_VERSION: Record = { + isolation_session: STATE_AWARE_VERSION, + windows_sandbox: STATE_AWARE_VERSION, + wslc: WSLC_STATE_AWARE_VERSION, +}; // Mapping from a sandboxId's leading prefix segment to the wire-format // backend key. Extended as more state-aware backends opt in. export const PREFIX_TO_BACKEND: Record = { [ISOLATION_SESSION_ID_PREFIX]: 'isolation_session', [WINDOWS_SANDBOX_ID_PREFIX]: 'windows_sandbox', + [WSLC_ID_PREFIX]: 'wslc', }; /** @@ -67,7 +85,8 @@ export function buildStateAwareEnvelope(args: BuildEnvelopeArgs): Record.. const backendSpecific: Record = { ...(config ?? {}) }; - const version = (typeof backendSpecific.version === 'string' && backendSpecific.version) || STATE_AWARE_VERSION; + const defaultVersion = DEFAULT_STATE_AWARE_VERSION[backendKey] ?? STATE_AWARE_VERSION; + const version = (typeof backendSpecific.version === 'string' && backendSpecific.version) || defaultVersion; delete backendSpecific.version; const envelope: Record = { version, phase }; diff --git a/sdk/node/src/state-aware-types.ts b/sdk/node/src/state-aware-types.ts index e8f7093a5..4310340a7 100644 --- a/sdk/node/src/state-aware-types.ts +++ b/sdk/node/src/state-aware-types.ts @@ -4,6 +4,7 @@ import { ContainmentBackend, FilesystemConfig, + NetworkConfig, ProcessConfig, } from './types.js'; @@ -18,7 +19,7 @@ export type Phase = 'provision' | 'start' | 'exec' | 'stop' | 'deprovision'; */ export type StateAwareContainmentBackend = Extract< ContainmentBackend, - 'isolation_session' | 'windows_sandbox' + 'isolation_session' | 'windows_sandbox' | 'wslc' >; /** @@ -146,6 +147,78 @@ export interface WindowsSandboxDeprovisionConfig { version?: string; } +// WSLc per-(backend, phase) Configs. WSLc runs each sandbox as a warm +// container behind a persistent host-side daemon (one amortized WSL session +// shared across sandboxes). Filesystem mounts and network mode are applied at +// provision and frozen for the sandbox's lifetime; a cooperative env-var proxy +// may be injected per-exec. + +export interface WslcProvisionConfig { + /** Schema version (semver). When omitted, the SDK fills in `0.8.0-alpha`. */ + version?: string; + /** + * Filesystem policy applied at provision and frozen for the life of the + * sandbox. `readwritePaths` / `readonlyPaths` become container volume mounts + * at the same absolute host path. The backend runs the same object-identity + * normalization + delegation gate as the one-shot runner and rejects a + * `deniedPath` equal to or nested within a mounted share (WSLc has no Deny + * mount primitive) with `code: 'policy_validation'`. + */ + filesystem?: FilesystemConfig; + /** + * Network mode applied at provision and frozen thereafter. Only + * `defaultPolicy` is honored: `'allow'` provisions a bridged container, + * `'block'` (the default when omitted) provisions with no network. Per-host + * filtering (`allowedHosts` / `blockedHosts`) and a `proxy` are rejected at + * provision (`code: 'policy_validation'`) — WSLc has no in-kernel iptables, + * and the cooperative proxy is an exec-phase concern (see + * {@link WslcExecConfig.network}). + */ + network?: NetworkConfig; + /** + * Container image reference (e.g. `alpine:latest`). Defaults to + * `alpine:latest` when omitted. Nested under + * `experimental.wslc.provision.image` on the wire. + */ + image?: string; + /** + * Path to a local image tarball to import instead of pulling. Nested under + * `experimental.wslc.provision.imageTarPath` on the wire. + */ + imageTarPath?: string; +} + +export interface WslcStartConfig { + /** Schema version (semver). */ + version?: string; +} + +export interface WslcExecConfig { + /** Schema version (semver). */ + version?: string; + process: ProcessConfig; + /** + * Per-exec network overrides. Only `proxy` is honored: it injects a + * cooperative `HTTP_PROXY` / `HTTPS_PROXY` into the command's environment + * (well-behaved HTTP clients honor it; raw-socket clients can bypass it). + * WSLc accepts only the `{ url }` proxy form — its containers run in their + * own network namespace, so the `localhost` / `builtinTestServer` loopback + * forms are unreachable and rejected. All other network fields are ignored + * at exec (network mode is fixed at provision). + */ + network?: NetworkConfig; +} + +export interface WslcStopConfig { + /** Schema version (semver). */ + version?: string; +} + +export interface WslcDeprovisionConfig { + /** Schema version (semver). */ + version?: string; +} + /** * The five per-phase Config slots every state-aware backend must declare. * `object` (not `Record`) is the slot base: interfaces have @@ -184,6 +257,13 @@ type StateAwareConfigRegistry = DefineStateAwareConfigRegistry<{ stop: WindowsSandboxStopConfig; deprovision: WindowsSandboxDeprovisionConfig; }; + wslc: { + provision: WslcProvisionConfig; + start: WslcStartConfig; + exec: WslcExecConfig; + stop: WslcStopConfig; + deprovision: WslcDeprovisionConfig; + }; }>; /** Compile-time guard: catches a backend with no registry entry. */ @@ -274,6 +354,10 @@ export type StateAwareMetadata = DefineStateAwareMetadataRegistry<{ // checks for `C = 'windows_sandbox'`. `Record` has `keyof = // never`, so every `*MetadataFor<'windows_sandbox'>` resolves to `undefined`. windows_sandbox: Record; + // WSLc returns no metadata for any phase (provision yields only the sandbox + // id). `Record` has `keyof = never`, so every + // `*MetadataFor<'wslc'>` resolves to `undefined`. + wslc: Record; // Future state-aware-capable backends add typed entries here. }>; diff --git a/sdk/node/tests/unit/state-aware-types.test.ts b/sdk/node/tests/unit/state-aware-types.test.ts index b10b03481..4e027d7ae 100644 --- a/sdk/node/tests/unit/state-aware-types.test.ts +++ b/sdk/node/tests/unit/state-aware-types.test.ts @@ -17,6 +17,11 @@ import { StopConfigFor, WindowsSandboxProvisionConfig, WindowsSandboxStartConfig, + WslcProvisionConfig, + WslcStartConfig, + WslcExecConfig, + WslcStopConfig, + WslcDeprovisionConfig, } from '../../src/state-aware-types.js'; import { backendForSandboxId } from '../../src/state-aware-helper.js'; @@ -364,3 +369,110 @@ describe('ProvisionResult', () => { assert.strictEqual(result.metadata?.ephemeralWorkspacePath, 'C:\\ProgramData\\ws'); }); }); + +describe('WslcProvisionConfig', () => { + it('accepts version, filesystem, network, and the backend-specific image knobs', () => { + const cfg: WslcProvisionConfig = { + version: '0.8.0-alpha', + filesystem: { readwritePaths: ['C:\\ws\\rw'], readonlyPaths: ['C:\\ws\\ro'] }, + network: { defaultPolicy: 'allow' }, + image: 'alpine:latest', + imageTarPath: 'C:\\images\\alpine.tar', + }; + assert.strictEqual(cfg.image, 'alpine:latest'); + assert.strictEqual(cfg.imageTarPath, 'C:\\images\\alpine.tar'); + assert.strictEqual(cfg.filesystem?.readwritePaths?.[0], 'C:\\ws\\rw'); + }); + + it('is entirely optional (every member optional)', () => { + const empty: WslcProvisionConfig = {}; + assert.ok(empty); + }); + + it('rejects an undeclared backend-specific field', () => { + const cfg: WslcProvisionConfig = { + // @ts-expect-error — wslc provision declares no such field. + unsupportedSetting: { nested: true }, + }; + assert.ok(cfg); + }); + + it('rejects ui at provision', () => { + const cfg: WslcProvisionConfig = { + // @ts-expect-error — ui is not exposed on the wslc provision config. + ui: { disable: true, clipboard: 'none', injection: false }, + }; + assert.ok(cfg); + }); +}); + +describe('WslcStartConfig / WslcStopConfig / WslcDeprovisionConfig', () => { + it('carry only version', () => { + const start: WslcStartConfig = { version: '0.8.0-alpha' }; + const stop: WslcStopConfig = {}; + const deprov: WslcDeprovisionConfig = {}; + assert.strictEqual(start.version, '0.8.0-alpha'); + assert.ok(stop); + assert.ok(deprov); + + const wrongStart: WslcStartConfig = { + // @ts-expect-error — start accepts no backend-specific config. + image: 'alpine:latest', + }; + assert.ok(wrongStart); + }); +}); + +describe('WslcExecConfig', () => { + it('requires process and accepts an optional cooperative proxy', () => { + const cfg: WslcExecConfig = { + process: { commandLine: 'echo hi' }, + network: { proxy: { url: 'http://127.0.0.1:8888' } }, + }; + assert.strictEqual(cfg.process.commandLine, 'echo hi'); + + // @ts-expect-error — exec config requires process. + const missing: WslcExecConfig = { network: { proxy: { url: 'http://127.0.0.1:8888' } } }; + assert.ok(missing); + }); +}); + +describe('Wslc metadata resolves to undefined for every phase', () => { + it('ProvisionResult carries no metadata and the id brands distinctly', () => { + const provMeta: ProvisionMetadataFor<'wslc'> = undefined; + const startMeta: StartMetadataFor<'wslc'> = undefined; + assert.strictEqual(provMeta, undefined); + assert.strictEqual(startMeta, undefined); + + const result: ProvisionResult<'wslc'> = { + sandboxId: 'wslc:abcd' as SandboxId<'wslc'>, + }; + assert.strictEqual(result.metadata, undefined); + + function takesWslcId(_id: SandboxId<'wslc'>): void { + // body unused + } + // @ts-expect-error — an isolation_session id is not a wslc id. + takesWslcId('iso:abcd' as SandboxId<'isolation_session'>); + assert.ok(true); + }); + + it('routes a wslc: id to the wslc backend by prefix', () => { + const id = 'wslc:0123abcd' as SandboxId<'wslc'>; + assert.strictEqual(backendForSandboxId(id), 'wslc'); + }); +}); + +describe('ConfigsForBackend selects the wslc bundle', () => { + it('selects the Wslc bundle for the wslc backend', () => { + const bundle: ConfigsForBackend<'wslc'> = { + provision: { image: 'alpine:latest' }, + start: {}, + exec: { process: { commandLine: 'echo' } }, + stop: {}, + deprovision: {}, + }; + assert.strictEqual(bundle.provision.image, 'alpine:latest'); + assert.strictEqual(bundle.exec.process.commandLine, 'echo'); + }); +}); diff --git a/sdk/node/tests/unit/state-aware.test.ts b/sdk/node/tests/unit/state-aware.test.ts index 566c6be6e..57c9e3d0a 100644 --- a/sdk/node/tests/unit/state-aware.test.ts +++ b/sdk/node/tests/unit/state-aware.test.ts @@ -546,3 +546,126 @@ describe('windows_sandbox state-aware lifecycle', () => { }); }); }); + +describe('wslc state-aware lifecycle', () => { + it('defaults the version to 0.8.0-alpha (not the isolation_session default)', () => { + const env = buildStateAwareEnvelope({ + phase: 'provision', + backendKey: 'wslc', + containment: 'wslc', + config: { image: 'alpine:latest' }, + }); + assert.strictEqual(env.version, '0.8.0-alpha'); + }); + + it('still honors a caller-supplied version over the wslc default', () => { + const env = buildStateAwareEnvelope({ + phase: 'provision', + backendKey: 'wslc', + containment: 'wslc', + config: { version: '0.8.1-alpha', image: 'alpine:latest' }, + }); + assert.strictEqual(env.version, '0.8.1-alpha'); + }); + + it('lifts filesystem + network and nests image under experimental.wslc.provision', () => { + const env = buildStateAwareEnvelope({ + phase: 'provision', + backendKey: 'wslc', + containment: 'wslc', + config: { + filesystem: { readwritePaths: ['C:\\ws\\rw'] }, + network: { defaultPolicy: 'allow' }, + image: 'alpine:latest', + imageTarPath: 'C:\\images\\alpine.tar', + }, + }); + assert.strictEqual(env.containment, 'wslc'); + assert.deepStrictEqual(env.filesystem, { readwritePaths: ['C:\\ws\\rw'] }); + assert.deepStrictEqual(env.network, { defaultPolicy: 'allow' }); + const wire = JSON.parse(JSON.stringify(env)); + assert.deepStrictEqual(wire.experimental, { + wslc: { provision: { image: 'alpine:latest', imageTarPath: 'C:\\images\\alpine.tar' } }, + }); + }); + + it('omits the experimental block when provision carries no backend-specific field', () => { + const env = buildStateAwareEnvelope({ + phase: 'provision', + backendKey: 'wslc', + containment: 'wslc', + config: { network: { defaultPolicy: 'block' } }, + }); + assert.strictEqual(env.experimental, undefined); + assert.deepStrictEqual(env.network, { defaultPolicy: 'block' }); + }); + + it('lifts exec process + cooperative proxy network to top-level with no experimental block', () => { + const env = buildStateAwareEnvelope({ + phase: 'exec', + backendKey: 'wslc', + sandboxId: 'wslc:abc', + config: { + process: { commandLine: 'echo hi' }, + network: { proxy: { url: 'http://127.0.0.1:8888' } }, + }, + }); + assert.deepStrictEqual(env.process, { commandLine: 'echo hi' }); + assert.deepStrictEqual(env.network, { proxy: { url: 'http://127.0.0.1:8888' } }); + assert.strictEqual(env.experimental, undefined); + }); + + describe('round-trip via the typed API', { skip: platformSkip }, () => { + afterEach(() => { _resetSpawnImpl(); }); + + it('provisionSandbox builds a wslc envelope and routes back via the wslc: prefix', async () => { + const fake = fakeSpawn({ stdout: '{"result":{"sandboxId":"wslc:0123abcd"}}', exitCode: 0 }); + _setSpawnImpl(fake.spawn); + const result = await provisionSandbox( + 'wslc', + { image: 'alpine:latest', network: { defaultPolicy: 'block' } }, + testOptions(), + ); + assert.strictEqual(result.sandboxId, 'wslc:0123abcd'); + assert.strictEqual(fake.captured.envelope?.phase, 'provision'); + assert.strictEqual(fake.captured.envelope?.containment, 'wslc'); + assert.strictEqual(fake.captured.envelope?.version, '0.8.0-alpha'); + }); + + it('startSandbox infers wslc from the wslc: prefix', async () => { + const fake = fakeSpawn({ stdout: '{"result":{}}', exitCode: 0 }); + _setSpawnImpl(fake.spawn); + const id = 'wslc:0123abcd' as SandboxId<'wslc'>; + await startSandbox(id, undefined, testOptions()); + assert.strictEqual(fake.captured.envelope?.phase, 'start'); + assert.strictEqual(fake.captured.envelope?.sandboxId, 'wslc:0123abcd'); + assert.strictEqual(fake.captured.envelope?.experimental, undefined); + }); + + it('execInSandboxAsync places process at top-level for a wslc: id', async () => { + const fake = fakeSpawn({ stdout: 'hello-from-wslc\n', stderr: '', exitCode: 0 }); + _setSpawnImpl(fake.spawn); + const id = 'wslc:0123abcd' as SandboxId<'wslc'>; + const result = await execInSandboxAsync( + id, + { process: { commandLine: 'echo hello-from-wslc' } }, + testOptions(), + ); + assert.deepStrictEqual(result, { stdout: 'hello-from-wslc\n', stderr: '', exitCode: 0 }); + assert.deepStrictEqual(fake.captured.envelope?.process, { commandLine: 'echo hello-from-wslc' }); + }); + + it('stopSandbox and deprovisionSandbox build minimal envelopes for a wslc: id', async () => { + for (const phase of ['stop', 'deprovision'] as const) { + const fake = fakeSpawn({ stdout: '{"result":{}}', exitCode: 0 }); + _setSpawnImpl(fake.spawn); + const id = 'wslc:0123abcd' as SandboxId<'wslc'>; + const call = phase === 'stop' ? stopSandbox : deprovisionSandbox; + await call(id, undefined, testOptions()); + assert.strictEqual(fake.captured.envelope?.phase, phase); + assert.strictEqual(fake.captured.envelope?.sandboxId, 'wslc:0123abcd'); + _resetSpawnImpl(); + } + }); + }); +}); From 5e42f6f141b0be6b2a5e22118af9b7aa69c8154f Mon Sep 17 00:00:00 2001 From: Soham Das Date: Tue, 11 Aug 2026 20:28:55 -0700 Subject: [PATCH 6/7] Addressed PR comments --- sdk/node/src/state-aware-types.ts | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/sdk/node/src/state-aware-types.ts b/sdk/node/src/state-aware-types.ts index 4310340a7..850091aef 100644 --- a/sdk/node/src/state-aware-types.ts +++ b/sdk/node/src/state-aware-types.ts @@ -120,8 +120,8 @@ export interface WindowsSandboxProvisionConfig { * sandbox. `readwritePaths` / `readonlyPaths` are mapped into the guest at * the same absolute host path; `deniedPaths` name HOST paths the contained * code must not reach. The SDK forwards this policy as-is; the backend - * enforces it at provision and rejects a `deniedPath` equal to or nested - * within a mapped share (`.wsb` has no Deny primitive). + * enforces it at provision and rejects a `deniedPaths` entry equal to or + * nested within a mapped share (`.wsb` has no Deny primitive). */ filesystem?: FilesystemConfig; } @@ -161,8 +161,8 @@ export interface WslcProvisionConfig { * sandbox. `readwritePaths` / `readonlyPaths` become container volume mounts * at the same absolute host path. The backend runs the same object-identity * normalization + delegation gate as the one-shot runner and rejects a - * `deniedPath` equal to or nested within a mounted share (WSLc has no Deny - * mount primitive) with `code: 'policy_validation'`. + * `deniedPaths` entry equal to or nested within a mounted share (WSLc has no + * Deny mount primitive) with `code: 'policy_validation'`. */ filesystem?: FilesystemConfig; /** @@ -203,8 +203,9 @@ export interface WslcExecConfig { * (well-behaved HTTP clients honor it; raw-socket clients can bypass it). * WSLc accepts only the `{ url }` proxy form — its containers run in their * own network namespace, so the `localhost` / `builtinTestServer` loopback - * forms are unreachable and rejected. All other network fields are ignored - * at exec (network mode is fixed at provision). + * forms are unreachable and rejected. Every other network field — host + * filters, a `defaultPolicy` change, and `allowLocalNetwork` — is rejected + * with `code: 'policy_validation'` (network mode is fixed at provision). */ network?: NetworkConfig; } From 527a731fcd31d65583c44933a691dc121aa38fbd Mon Sep 17 00:00:00 2001 From: Soham Das Date: Thu, 13 Aug 2026 13:15:02 -0700 Subject: [PATCH 7/7] Addressed PR comments Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: f3ae1af2-7b79-4340-a5ce-a5402e7ede3d --- sdk/node/README.md | 2 +- sdk/node/src/state-aware-helper.ts | 32 +++++++--- sdk/node/tests/integration/test-helpers.ts | 6 ++ .../unit/wire-conformance-state-aware.test.ts | 62 ++++++++++++++++++- 4 files changed, 91 insertions(+), 11 deletions(-) diff --git a/sdk/node/README.md b/sdk/node/README.md index 0c995e7a3..8da746ea8 100644 --- a/sdk/node/README.md +++ b/sdk/node/README.md @@ -265,7 +265,7 @@ await deprovisionSandbox(sandboxId, undefined, opts); `windows_sandbox` follows the same shape (substitute the containment string and provide `filesystem.readwritePaths` / `readonlyPaths` at provision if needed). See [`docs/windows-sandbox/windows-sandbox.md`](https://github.com/microsoft/mxc/blob/main/docs/windows-sandbox/windows-sandbox.md) for the per-phase config matrix. -`wslc` follows the same shape and needs no provision config at all (it defaults to an `alpine:latest` container with no network). Provide `filesystem.readwritePaths` / `readonlyPaths` (mounted for the sandbox's lifetime), `network.defaultPolicy: 'allow'` (a bridged container; the default `'block'` gives no network), and/or a backend-specific `image` / `imageTarPath` at provision; inject a cooperative `network.proxy: { url }` per-exec. WSLc state-aware requests default to schema `0.8.0-alpha`. See [`docs/wsl/`](https://github.com/microsoft/mxc/tree/main/docs/wsl/) for details. +`wslc` follows the same shape and needs no provision config at all (it defaults to an `alpine:latest` container with no network). Provide `filesystem.readwritePaths` / `readonlyPaths` (mounted for the sandbox's lifetime), `network.defaultPolicy: 'allow'` (a bridged container; the default `'block'` gives no network), and/or a backend-specific `image` / `imageTarPath` at provision; inject a cooperative `network.proxy: { url }` per-exec. WSLc state-aware requests default to schema `0.8.0-alpha`. See [`docs/wsl/wslc-state-aware.md`](https://github.com/microsoft/mxc/blob/main/docs/wsl/wslc-state-aware.md) for the per-phase config matrix. **Handling failures.** Every lifecycle call rejects with a typed `MxcError`. Branch on `code` first; when the failure came from an underlying platform API, the error also carries discrete diagnostic fields rather than a prose blob: diff --git a/sdk/node/src/state-aware-helper.ts b/sdk/node/src/state-aware-helper.ts index 9ecdfca3c..5174ce424 100644 --- a/sdk/node/src/state-aware-helper.ts +++ b/sdk/node/src/state-aware-helper.ts @@ -11,9 +11,10 @@ import { Phase, StateAwareContainmentBackend } from './state-aware-types.js'; export const STATE_AWARE_VERSION = '0.6.0-alpha'; // WSLc's state-aware surface shipped at a later schema version than the -// isolation_session default above. It is intentionally NOT gate-locked to the -// canonical `stateAware` constant (which tracks isolation_session): the two -// backends were promoted independently. See `DEFAULT_STATE_AWARE_VERSION`. +// `STATE_AWARE_VERSION` default above (the shared default for IsolationSession +// and Windows Sandbox). WSLc is intentionally NOT gate-locked to it: the +// backends were promoted independently, so WSLc carries its own later default. +// See `DEFAULT_STATE_AWARE_VERSION`. export const WSLC_STATE_AWARE_VERSION = '0.8.0-alpha'; // Wire-format cross-cutting fields that live at the envelope's top level. @@ -39,14 +40,27 @@ const DEFAULT_STATE_AWARE_VERSION: Record wslc: WSLC_STATE_AWARE_VERSION, }; -// Mapping from a sandboxId's leading prefix segment to the wire-format -// backend key. Extended as more state-aware backends opt in. -export const PREFIX_TO_BACKEND: Record = { - [ISOLATION_SESSION_ID_PREFIX]: 'isolation_session', - [WINDOWS_SANDBOX_ID_PREFIX]: 'windows_sandbox', - [WSLC_ID_PREFIX]: 'wslc', +// Exhaustive backend→prefix map. Typed `Record` so adding a backend to the union without registering a prefix here +// is a compile error — the same exhaustiveness guarantee the config, metadata, +// and default-version registries carry. Without it a new backend would compile +// with no prefix and fail every non-provision call at runtime with +// `malformed_id`. +export const BACKEND_TO_PREFIX: Record = { + isolation_session: ISOLATION_SESSION_ID_PREFIX, + windows_sandbox: WINDOWS_SANDBOX_ID_PREFIX, + wslc: WSLC_ID_PREFIX, }; +// Reverse lookup (prefix → backend), derived from the exhaustive map above so +// the two can never drift. Used to route a sandboxId's leading prefix segment +// to its wire-format backend key. +export const PREFIX_TO_BACKEND: Record = Object.fromEntries( + (Object.entries(BACKEND_TO_PREFIX) as [StateAwareContainmentBackend, string][]).map( + ([backend, prefix]) => [prefix, backend], + ), +); + /** * Resolves the wire-format backend key for a sandbox id by reading its * leading prefix segment. Throws an `MxcError` with `code: 'malformed_id'` diff --git a/sdk/node/tests/integration/test-helpers.ts b/sdk/node/tests/integration/test-helpers.ts index 77da91bd0..f353f666e 100644 --- a/sdk/node/tests/integration/test-helpers.ts +++ b/sdk/node/tests/integration/test-helpers.ts @@ -262,6 +262,12 @@ export async function probeStateAwareRuntime>; // per-phase rather than a single pooled key set. That is strictly stronger: a // field legal only on provision cannot satisfy the oracle by appearing // on the start config, or vice versa. -type LiftedPhaseKey = 'version' | 'process' | 'network'; +// +// `filesystem` is a lifted top-level wire field (like `network`): WSLc provision +// surfaces it publicly but it maps to the envelope's top-level `filesystem`, not +// under `experimental.wslc.provision`. Listing it here keeps the backend-key set +// limited to genuinely per-phase wire fields. +type LiftedPhaseKey = 'version' | 'process' | 'network' | 'filesystem'; type BackendKeys = Exclude; type WireKeys = keyof StripIndex; @@ -123,6 +134,44 @@ type _ProvisionWireKeysNonVacuous = AssertTrue< Equivalent, 'appId'> >; +// --- WSLc per-phase wire field-set conformance ----------------------------- + +// WSLc is the second state-aware backend, so the oracle must cover it too or a +// wire-model change to the WSLc surface would regenerate `wire.ts`, pass the +// codegen gate, and leave the SDK silently lagging with no CI signal. WSLc's +// only per-phase wire object is provision (`image` / `imageTarPath`); start, +// exec, stop, and deprovision have wire associated type `()` and must expose no +// backend-specific field. `filesystem` and `network` are lifted top-level wire +// fields (see `LiftedPhaseKey`), so they are excluded from the backend-key sets. +type _WslcProvisionPublicKeys = AssertTrue< + Equivalent, WireKeys>, never> +>; +type _WslcProvisionWireKeys = AssertTrue< + Equivalent, BackendKeys>, never> +>; +type _WslcProvisionFieldValueTypes = AssertTrue< + Equivalent< + PublicFieldValues, + WireFieldValues + > +>; + +type _WslcStartNoBackendKeys = AssertTrue, never>>; +type _WslcExecNoBackendKeys = AssertTrue, never>>; +type _WslcStopNoBackendKeys = AssertTrue, never>>; +type _WslcDeprovisionNoBackendKeys = AssertTrue< + Equivalent, never> +>; + +// Non-vacuity guards (see the isolation_session pins above): pin the derived key +// sets so a derivation bug fails the oracle rather than silently disabling it. +type _WslcProvisionKeysNonVacuous = AssertTrue< + Equivalent, 'image' | 'imageTarPath'> +>; +type _WslcProvisionWireKeysNonVacuous = AssertTrue< + Equivalent, 'image' | 'imageTarPath'> +>; + // --- delegation to the one-shot oracle (documented, asserted) -------------- // The per-phase configs must REUSE the public one-shot leaf types for their @@ -130,6 +179,7 @@ type _ProvisionWireKeysNonVacuous = AssertTrue< // re-declared an inline shape instead, it would escape that coverage — these // assertions fail if that ever happens. type _ExecProcessReuse = AssertTrue>; +type _WslcExecProcessReuse = AssertTrue>; // Reference the assertion aliases so they read as intentionally load-bearing. export type StateAwareWireConformanceAssertions = [ @@ -144,6 +194,16 @@ export type StateAwareWireConformanceAssertions = [ _ProvisionKeysNonVacuous, _ProvisionWireKeysNonVacuous, _ExecProcessReuse, + _WslcProvisionPublicKeys, + _WslcProvisionWireKeys, + _WslcProvisionFieldValueTypes, + _WslcStartNoBackendKeys, + _WslcExecNoBackendKeys, + _WslcStopNoBackendKeys, + _WslcDeprovisionNoBackendKeys, + _WslcProvisionKeysNonVacuous, + _WslcProvisionWireKeysNonVacuous, + _WslcExecProcessReuse, ]; test('public state-aware SDK types conform to the generated wire schema (compile-time)', () => {