Add Windows PTY support for Python worker

Run built-in Python through ConPTY on Windows so it uses the same PTY-backed stdin and sideband accounting path as Unix. Keep the worker protocol unchanged while adding Windows console setup, input draining, and zod/Python parity coverage.

Verification: cargo check; cargo build; python tests/run_integration_tests.py --binary target/debug/mcp-repl.exe; cargo clippy --all-targets --all-features -- -D warnings; cargo test --quiet; cargo +nightly fmt. The required python3 integration command could not run because python3 is not installed on this Windows host.

Author: t-kalinowski

Cargo.toml

@@ -41,7 +41,7 @@
   landlock    = "0.4.4"
   seccompiler = "0.5.0"
 
-[target.'cfg(unix)'.dependencies]
+[target.'cfg(any(unix, windows))'.dependencies]
   portable-pty = "0.9.0"
 
 [target.'cfg(target_os = "macos")'.dependencies]
@@ -56,6 +56,7 @@
     "Win32_Security_Authorization",
     "Win32_Storage_FileSystem",
     "Win32_System_Console",
+    "Win32_System_Environment",
     "Win32_System_IO",
     "Win32_System_JobObjects",
     "Win32_System_Pipes",

docs/plans/active/worker-server-protocol-zod.md

@@ -647,7 +647,7 @@ ordered on the worker-to-server sideband stream. The server must not
 assume that writing the `interrupt` message means the worker has already
 processed it; later `readline_input`, `readline_discard`,
 `readline_start`, and `session_end` events determine recovery.
-Built-in Unix Python currently has a private `python_interrupt` /
+Built-in PTY-backed Python currently has a private `python_interrupt` /
 `python_interrupt_ack` cleanup handshake so it can drain PTY input before
 SIGINT; that acknowledgement is transitional and not part of the generic
 worker protocol.

src/backend.rs

@@ -39,7 +39,9 @@ impl WorkerLaunch {
 
     pub fn stdin_transport(&self) -> WorkerStdinTransport {
         match self {
-            Self::Builtin(Backend::Python) if cfg!(target_family = "unix") => {
+            Self::Builtin(Backend::Python)
+                if cfg!(any(target_family = "unix", target_os = "windows")) =>
+            {
                 WorkerStdinTransport::Pty
             }
             Self::Builtin(_) => WorkerStdinTransport::Pipe,
@@ -198,12 +200,12 @@ mod tests {
             WorkerLaunch::Builtin(Backend::R).stdin_transport(),
             WorkerStdinTransport::Pipe
         );
-        #[cfg(not(target_family = "unix"))]
+        #[cfg(not(any(target_family = "unix", target_os = "windows")))]
         assert_eq!(
             WorkerLaunch::Builtin(Backend::Python).stdin_transport(),
             WorkerStdinTransport::Pipe
         );
-        #[cfg(target_family = "unix")]
+        #[cfg(any(target_family = "unix", target_os = "windows"))]
         assert_eq!(
             WorkerLaunch::Builtin(Backend::Python).stdin_transport(),
             WorkerStdinTransport::Pty

src/ipc.rs

@@ -279,7 +279,9 @@ impl OutputCriticalIpcWriter {
             .writer
             .lock()
             .map_err(|_| io::Error::other("ipc writer mutex poisoned"))?;
-        write_ipc_message(&mut **writer, &message)
+        let result = write_ipc_message(&mut **writer, &message);
+        drop(writer);
+        result
     }
 }
 
@@ -631,7 +633,10 @@ impl ServerIpcConnection {
         Ok(())
     }
 
-    #[cfg_attr(target_family = "unix", allow(dead_code))]
+    #[cfg_attr(
+        any(target_family = "unix", target_family = "windows"),
+        allow(dead_code)
+    )]
     pub fn begin_request(&self) {
         let mut guard = self.inbox.lock().unwrap();
         reset_after_completed_request(&mut guard);
@@ -1009,8 +1014,9 @@ impl WorkerIpcConnection {
 fn write_ipc_message<T: Serialize>(writer: &mut dyn Write, message: &T) -> io::Result<()> {
     let payload = serde_json::to_string(message).map_err(io::Error::other)?;
     writer.write_all(payload.as_bytes())?;
-    writer.write_all(b"\n")?;
-    writer.flush()
+    // IPC transports are unbuffered OS pipes. On Windows named pipes, flushing
+    // can wait for peer drainage, so a complete JSONL write is the sync point.
+    writer.write_all(b"\n")
 }
 
 #[derive(Debug)]
@@ -1113,7 +1119,7 @@ impl IpcServer {
         self,
         handle: IpcHandle,
         handlers: IpcHandlers,
-        child: &mut std::process::Child,
+        child_exited: impl FnMut() -> io::Result<bool>,
         max_wait: Duration,
     ) -> io::Result<()> {
         let Some(server_pipe_to_worker) = self.server_pipe_to_worker else {
@@ -1127,9 +1133,10 @@ impl IpcServer {
             ));
         };
         let start = Instant::now();
-        connect_named_pipe_with_process_retry(&server_pipe_to_worker, child, max_wait)?;
+        let child_exited = std::cell::RefCell::new(child_exited);
+        connect_named_pipe_with_process_retry(&server_pipe_to_worker, &child_exited, max_wait)?;
         let remaining = max_wait.saturating_sub(start.elapsed());
-        connect_named_pipe_with_process_retry(&server_pipe_from_worker, child, remaining)?;
+        connect_named_pipe_with_process_retry(&server_pipe_from_worker, &child_exited, remaining)?;
         let conn = ServerIpcConnection::new(
             IpcTransport {
                 reader: Box::new(server_pipe_from_worker),
@@ -1459,12 +1466,12 @@ fn join_connector_with_grace(connector: thread::JoinHandle<()>, max_wait: Durati
 #[cfg(target_family = "windows")]
 fn connect_named_pipe_with_process_retry(
     server_pipe: &File,
-    child: &mut std::process::Child,
+    child_exited: &std::cell::RefCell<impl FnMut() -> io::Result<bool>>,
     max_wait: Duration,
 ) -> io::Result<()> {
     connect_named_pipe_with_process_retry_impl(
         |timeout| connect_named_pipe(server_pipe, timeout),
-        || child.try_wait().map(|status| status.is_some()),
+        || child_exited.borrow_mut()(),
         max_wait,
     )
 }