Add Windows sandbox unified exec runtime support (#15578)

## Summary

This is the runtime/foundation half of the Windows sandbox unified-exec
work.

- add Windows sandbox `unified_exec` session support in
`windows-sandbox-rs` for both:
  - the legacy restricted-token backend
  - the elevated runner backend
- extend the PTY/process runtime so driver-backed sessions can support:
  - stdin streaming
  - stdout/stderr separation
  - exit propagation
  - PTY resize hooks
- add Windows sandbox runtime coverage in `codex-windows-sandbox` /
`codex-utils-pty`

This PR does **not** enable Windows sandbox `UnifiedExec` for product
callers yet because hooking this up to app-server comes in the next PR.

Windows sandbox advertising is intentionally kept aligned with `main`,
so sandboxed Windows callers still fall back to `ShellCommand`.

This PR isolates the runtime/session layer so it can be reviewed
independently from product-surface enablement.

---------

Co-authored-by: jif-oai <jif@openai.com>
Co-authored-by: Codex <noreply@openai.com>

codex-rs/utils/pty/src/lib.rs

@@ -13,6 +13,8 @@ pub const DEFAULT_OUTPUT_BYTES_CAP: usize = 1024 * 1024;
 pub use pipe::spawn_process as spawn_pipe_process;
 /// Spawn a non-interactive process using regular pipes, but close stdin immediately.
 pub use pipe::spawn_process_no_stdin as spawn_pipe_process_no_stdin;
+/// Driver-backed process adapter used by integrations with their own process transport.
+pub use process::ProcessDriver;
 /// Handle for interacting with a spawned process (PTY or pipe).
 pub use process::ProcessHandle;
 /// Bundle of process handles plus split output and exit receivers returned by spawn helpers.
@@ -21,6 +23,8 @@ pub use process::SpawnedProcess;
 pub use process::TerminalSize;
 /// Combine stdout/stderr receivers into a single broadcast receiver.
 pub use process::combine_output_receivers;
+/// Adapt an externally-driven process into the standard spawned-process handle.
+pub use process::spawn_from_driver;
 /// Backwards-compatible alias for ProcessHandle.
 pub type ExecCommandSession = ProcessHandle;
 /// Backwards-compatible alias for SpawnedProcess.

codex-rs/utils/pty/src/pipe.rs

@@ -234,6 +234,7 @@ async fn spawn_process_with_stdin_mode(
         exit_status,
         exit_code,
         /*pty_handles*/ None,
+        /*resizer*/ None,
     );
 
     Ok(SpawnedProcess {

codex-rs/utils/pty/src/process.rs

@@ -13,6 +13,7 @@ use portable_pty::SlavePty;
 use tokio::sync::broadcast;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
+use tokio::sync::watch;
 use tokio::task::AbortHandle;
 use tokio::task::JoinHandle;
 
@@ -69,6 +70,10 @@ impl fmt::Debug for PtyHandles {
     }
 }
 
+/// Callback used by driver-backed sessions to resize a PTY-like backend when
+/// there is no local `PtyHandles` instance to resize directly.
+type ResizeFn = Box<dyn FnMut(TerminalSize) -> anyhow::Result<()> + Send>;
+
 /// Handle for driving an interactive process (PTY or pipe).
 pub struct ProcessHandle {
     writer_tx: StdMutex<Option<mpsc::Sender<Vec<u8>>>>,
@@ -82,6 +87,9 @@ pub struct ProcessHandle {
     // PtyHandles must be preserved because the process will receive Control+C if the
     // slave is closed
     _pty_handles: StdMutex<Option<PtyHandles>>,
+    // Optional resize hook for driver-backed sessions that proxy PTY control to
+    // another backend instead of owning local PTY handles.
+    resizer: StdMutex<Option<ResizeFn>>,
 }
 
 impl fmt::Debug for ProcessHandle {
@@ -102,6 +110,7 @@ impl ProcessHandle {
         exit_status: Arc<AtomicBool>,
         exit_code: Arc<StdMutex<Option<i32>>>,
         pty_handles: Option<PtyHandles>,
+        resizer: Option<ResizeFn>,
     ) -> Self {
         Self {
             writer_tx: StdMutex::new(Some(writer_tx)),
@@ -113,6 +122,7 @@ impl ProcessHandle {
             exit_status,
             exit_code,
             _pty_handles: StdMutex::new(pty_handles),
+            resizer: StdMutex::new(resizer),
         }
     }
 
@@ -141,17 +151,28 @@ impl ProcessHandle {
 
     /// Resize the PTY in character cells.
     pub fn resize(&self, size: TerminalSize) -> anyhow::Result<()> {
-        let handles = self
-            ._pty_handles
+        {
+            let handles = self
+                ._pty_handles
+                .lock()
+                .map_err(|_| anyhow!("failed to lock PTY handles"))?;
+            if let Some(handles) = handles.as_ref() {
+                return match &handles._master {
+                    PtyMasterHandle::Resizable(master) => master.resize(size.into()),
+                    #[cfg(unix)]
+                    PtyMasterHandle::Opaque { raw_fd, .. } => resize_raw_pty(*raw_fd, size),
+                };
+            }
+        }
+
+        let mut resizer = self
+            .resizer
             .lock()
-            .map_err(|_| anyhow!("failed to lock PTY handles"))?;
-        let handles = handles
-            .as_ref()
-            .ok_or_else(|| anyhow!("process is not attached to a PTY"))?;
-        match &handles._master {
-            PtyMasterHandle::Resizable(master) => master.resize(size.into()),
-            #[cfg(unix)]
-            PtyMasterHandle::Opaque { raw_fd, .. } => resize_raw_pty(*raw_fd, size),
+            .map_err(|_| anyhow!("failed to lock PTY resizer"))?;
+        if let Some(resizer) = resizer.as_mut() {
+            resizer(size)
+        } else {
+            Err(anyhow!("process is not attached to a PTY"))
         }
     }
 
@@ -205,6 +226,20 @@ impl Drop for ProcessHandle {
     }
 }
 
+/// Adapts a closure into a `ChildTerminator` implementation.
+struct ClosureTerminator {
+    inner: Option<Box<dyn FnMut() + Send + Sync>>,
+}
+
+impl ChildTerminator for ClosureTerminator {
+    fn kill(&mut self) -> io::Result<()> {
+        if let Some(inner) = self.inner.as_mut() {
+            (inner)();
+        }
+        Ok(())
+    }
+}
+
 #[cfg(unix)]
 fn resize_raw_pty(raw_fd: RawFd, size: TerminalSize) -> anyhow::Result<()> {
     let mut winsize = libc::winsize {
@@ -263,3 +298,113 @@ pub struct SpawnedProcess {
     pub stderr_rx: mpsc::Receiver<Vec<u8>>,
     pub exit_rx: oneshot::Receiver<i32>,
 }
+
+/// Driver-backed process handles for non-standard spawn backends.
+pub struct ProcessDriver {
+    pub writer_tx: mpsc::Sender<Vec<u8>>,
+    pub stdout_rx: broadcast::Receiver<Vec<u8>>,
+    pub stderr_rx: Option<broadcast::Receiver<Vec<u8>>>,
+    pub exit_rx: oneshot::Receiver<i32>,
+    pub terminator: Option<Box<dyn FnMut() + Send + Sync>>,
+    pub writer_handle: Option<JoinHandle<()>>,
+    pub resizer: Option<ResizeFn>,
+}
+
+/// Build a `SpawnedProcess` from a driver that supplies stdin/output/exit channels.
+pub fn spawn_from_driver(driver: ProcessDriver) -> SpawnedProcess {
+    let ProcessDriver {
+        writer_tx,
+        stdout_rx: stdout_driver_rx,
+        stderr_rx: mut stderr_driver_rx,
+        exit_rx,
+        terminator,
+        writer_handle,
+        resizer,
+    } = driver;
+
+    let (stdout_tx, stdout_rx) = mpsc::channel::<Vec<u8>>(256);
+    let (stderr_tx, stderr_rx) = mpsc::channel::<Vec<u8>>(256);
+    let (exit_seen_tx, exit_seen_rx) = watch::channel(false);
+    let spawn_stream_reader =
+        |mut output_rx: broadcast::Receiver<Vec<u8>>,
+         output_tx: mpsc::Sender<Vec<u8>>,
+         mut exit_seen_rx: watch::Receiver<bool>| {
+            tokio::spawn(async move {
+                let mut process_exited = false;
+                loop {
+                    let recv_result = if process_exited {
+                        match tokio::time::timeout(
+                            std::time::Duration::from_millis(200),
+                            output_rx.recv(),
+                        )
+                        .await
+                        {
+                            Ok(result) => result,
+                            Err(_) => break,
+                        }
+                    } else {
+                        tokio::select! {
+                            _ = exit_seen_rx.changed() => {
+                                process_exited = *exit_seen_rx.borrow();
+                                continue;
+                            }
+                            result = output_rx.recv() => result,
+                        }
+                    };
+                    match recv_result {
+                        Ok(chunk) => {
+                            if output_tx.send(chunk).await.is_err() {
+                                break;
+                            }
+                        }
+                        Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+                        Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+                    }
+                }
+            })
+        };
+    let reader_handle = spawn_stream_reader(stdout_driver_rx, stdout_tx, exit_seen_rx.clone());
+    let stderr_reader_handle = stderr_driver_rx
+        .take()
+        .map(|rx| spawn_stream_reader(rx, stderr_tx, exit_seen_rx));
+
+    let writer_handle = writer_handle.unwrap_or_else(|| tokio::spawn(async {}));
+
+    let (exit_tx, exit_rx_out) = oneshot::channel::<i32>();
+    let exit_status = Arc::new(AtomicBool::new(false));
+    let wait_exit_status = Arc::clone(&exit_status);
+    let exit_code = Arc::new(StdMutex::new(None));
+    let wait_exit_code = Arc::clone(&exit_code);
+    let wait_handle = tokio::spawn(async move {
+        let code = exit_rx.await.unwrap_or(-1);
+        wait_exit_status.store(true, std::sync::atomic::Ordering::SeqCst);
+        if let Ok(mut guard) = wait_exit_code.lock() {
+            *guard = Some(code);
+        }
+        let _ = exit_seen_tx.send(true);
+        let _ = exit_tx.send(code);
+    });
+
+    let handle = ProcessHandle::new(
+        writer_tx,
+        Box::new(ClosureTerminator { inner: terminator }),
+        reader_handle,
+        stderr_reader_handle
+            .map(|handle| handle.abort_handle())
+            .into_iter()
+            .collect(),
+        writer_handle,
+        wait_handle,
+        exit_status,
+        exit_code,
+        /*pty_handles*/ None,
+        resizer,
+    );
+
+    SpawnedProcess {
+        session: handle,
+        stdout_rx,
+        stderr_rx,
+        exit_rx: exit_rx_out,
+    }
+}

codex-rs/utils/pty/src/pty.rs

@@ -242,6 +242,7 @@ async fn spawn_process_portable(
         exit_status,
         exit_code,
         Some(handles),
+        /*resizer*/ None,
     );
 
     Ok(SpawnedProcess {
@@ -395,6 +396,7 @@ async fn spawn_process_preserving_fds(
         exit_status,
         exit_code,
         Some(handles),
+        /*resizer*/ None,
     );
 
     Ok(SpawnedProcess {

codex-rs/utils/pty/src/tests.rs

@@ -3,6 +3,7 @@ use std::path::Path;
 
 use pretty_assertions::assert_eq;
 
+use crate::ProcessDriver;
 use crate::SpawnedProcess;
 use crate::TerminalSize;
 use crate::combine_output_receivers;
@@ -10,6 +11,7 @@ use crate::combine_output_receivers;
 use crate::pipe::spawn_process_no_stdin_with_inherited_fds;
 #[cfg(unix)]
 use crate::pty::spawn_process_with_inherited_fds;
+use crate::spawn_from_driver;
 use crate::spawn_pipe_process;
 use crate::spawn_pipe_process_no_stdin;
 use crate::spawn_pty_process;
@@ -589,6 +591,103 @@ async fn pipe_process_can_expose_split_stdout_and_stderr() -> anyhow::Result<()>
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn driver_backed_process_can_expose_split_stdout_and_stderr() -> anyhow::Result<()> {
+    let (writer_tx, _writer_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(1);
+    let (stdout_tx, stdout_driver_rx) = tokio::sync::broadcast::channel::<Vec<u8>>(8);
+    let (stderr_tx, stderr_driver_rx) = tokio::sync::broadcast::channel::<Vec<u8>>(8);
+    let (exit_tx, exit_rx) = tokio::sync::oneshot::channel::<i32>();
+
+    let spawned = spawn_from_driver(ProcessDriver {
+        writer_tx,
+        stdout_rx: stdout_driver_rx,
+        stderr_rx: Some(stderr_driver_rx),
+        exit_rx,
+        terminator: None,
+        writer_handle: None,
+        resizer: None,
+    });
+
+    let SpawnedProcess {
+        session: _session,
+        stdout_rx,
+        stderr_rx,
+        exit_rx,
+    } = spawned;
+    let stdout_task = tokio::spawn(async move { collect_split_output(stdout_rx).await });
+    let stderr_task = tokio::spawn(async move { collect_split_output(stderr_rx).await });
+
+    stdout_tx.send(b"driver-out".to_vec())?;
+    stderr_tx.send(b"driver-err".to_vec())?;
+    drop(stdout_tx);
+    drop(stderr_tx);
+    exit_tx.send(0).expect("send exit code");
+
+    let timeout = tokio::time::Duration::from_secs(2);
+    let code = tokio::time::timeout(timeout, exit_rx)
+        .await
+        .map_err(|_| anyhow::anyhow!("timed out waiting for driver exit"))?
+        .unwrap_or(-1);
+    let stdout = tokio::time::timeout(timeout, stdout_task)
+        .await
+        .map_err(|_| anyhow::anyhow!("timed out waiting to drain driver stdout"))??;
+    let stderr = tokio::time::timeout(timeout, stderr_task)
+        .await
+        .map_err(|_| anyhow::anyhow!("timed out waiting to drain driver stderr"))??;
+
+    assert_eq!(stdout, b"driver-out".to_vec());
+    assert_eq!(stderr, b"driver-err".to_vec());
+    assert_eq!(code, 0);
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn driver_backed_process_can_resize_via_resizer_hook() -> anyhow::Result<()> {
+    let (writer_tx, _writer_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(1);
+    let (_stdout_tx, stdout_driver_rx) = tokio::sync::broadcast::channel::<Vec<u8>>(8);
+    let (exit_tx, exit_rx) = tokio::sync::oneshot::channel::<i32>();
+    let (size_tx, size_rx) = tokio::sync::oneshot::channel::<TerminalSize>();
+
+    let size_tx = std::sync::Arc::new(std::sync::Mutex::new(Some(size_tx)));
+    let spawned = spawn_from_driver(ProcessDriver {
+        writer_tx,
+        stdout_rx: stdout_driver_rx,
+        stderr_rx: None,
+        exit_rx,
+        terminator: None,
+        writer_handle: None,
+        resizer: Some(Box::new(move |size| {
+            if let Ok(mut guard) = size_tx.lock()
+                && let Some(size_tx) = guard.take()
+            {
+                let _ = size_tx.send(size);
+            }
+            Ok(())
+        })),
+    });
+
+    spawned.session.resize(TerminalSize {
+        rows: 40,
+        cols: 120,
+    })?;
+    exit_tx.send(0).expect("send exit code");
+
+    let resized = tokio::time::timeout(tokio::time::Duration::from_secs(2), size_rx)
+        .await
+        .map_err(|_| anyhow::anyhow!("timed out waiting for resize"))?
+        .expect("receive resized terminal size");
+    assert_eq!(
+        resized,
+        TerminalSize {
+            rows: 40,
+            cols: 120
+        }
+    );
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn pipe_terminate_aborts_detached_readers() -> anyhow::Result<()> {
     if !setsid_available() {

codex-rs/utils/pty/src/win/psuedocon.rs

@@ -118,6 +118,10 @@ fn windows_build_number() -> Option<u32> {
 
 pub struct PsuedoCon {
     con: HPCON,
+    // CreatePseudoConsole borrows these pipe handles for the lifetime of the
+    // pseudoconsole, so we must keep owning them until ClosePseudoConsole.
+    _input: FileDescriptor,
+    _output: FileDescriptor,
 }
 
 unsafe impl Send for PsuedoCon {}
@@ -149,7 +153,11 @@ impl PsuedoCon {
             result == S_OK,
             "failed to create psuedo console: HRESULT {result}"
         );
-        Ok(Self { con })
+        Ok(Self {
+            con,
+            _input: input,
+            _output: output,
+        })
     }
 
     pub fn resize(&self, size: COORD) -> Result<(), Error> {