feat: US-026 - Fix streaming stdin delivery from PTY to sandbox process

Author: NathanFlurry

native/v8-runtime/src/stream.rs

@@ -3,18 +3,22 @@
 /// Dispatch a stream event into V8 by calling the registered callback function.
 ///
 /// Stream events are sent by the host when async operations (child processes,
-/// HTTP servers) produce data. The event_type determines which V8 dispatch
-/// function is called:
-/// - "child_stdout", "child_stderr", "child_exit" → _childProcessDispatch
-/// - "http_request" → _httpServerDispatch
+/// HTTP servers, stdin) produce data. The event_type determines which V8
+/// dispatch function is called:
+/// - "child_stdout", "child_stderr", "child_exit", "childProcess" → _childProcessDispatch
+/// - "http_request", "httpServerRequest", "httpServerUpgrade", etc. → _httpServerDispatch
+/// - "stdin" → _stdinDispatch
+/// - "netSocket" → _netSocketDispatch
 pub fn dispatch_stream_event(scope: &mut v8::HandleScope, event_type: &str, payload: &[u8]) {
     // Look up the dispatch function on the global object
     let context = scope.get_current_context();
     let global = context.global(scope);
 
     let dispatch_name = match event_type {
-        "child_stdout" | "child_stderr" | "child_exit" => "_childProcessDispatch",
-        "http_request" => "_httpServerDispatch",
+        "child_stdout" | "child_stderr" | "child_exit" | "childProcess" => "_childProcessDispatch",
+        "http_request" | "httpServerRequest" | "httpServerUpgrade" | "upgradeSocketData" | "upgradeSocketEnd" => "_httpServerDispatch",
+        "stdin" => "_stdinDispatch",
+        "netSocket" => "_netSocketDispatch",
         _ => return, // Unknown event type — ignore
     };
 

packages/core/src/kernel/kernel.ts

@@ -316,11 +316,28 @@ class KernelImpl implements Kernel {
 			}
 		})();
 
+		// Start stdin pump: master write → PTY input buffer → slave read → writeStdin
+		// Bridges shell.write() data to the runtime driver's streaming stdin path
+		const stdinPumpPromise = (async () => {
+			try {
+				while (!pump.exited) {
+					const data = await this.ptyManager.read(slaveDescId, 4096);
+					if (!data || data.length === 0) break;
+					proc.writeStdin(data);
+				}
+			} catch {
+				// PTY closed — expected when shell exits
+			}
+			// Signal stdin EOF to the runtime driver
+			try { proc.closeStdin(); } catch { /* already closed */ }
+		})();
+
 		// wait() resolves after both shell exit AND pump drain
 		const waitPromise = proc.wait().then(async (exitCode) => {
 			pump.exited = true;
-			// Wait for pump to finish delivering remaining data
+			// Wait for pumps to finish delivering remaining data
 			await pumpPromise;
+			await stdinPumpPromise;
 			// Clean up controller PID's FD table (incl. PTY master)
 			this.cleanupProcessFDs(controllerPid);
 			return exitCode;

packages/nodejs/src/bridge-contract.ts

@@ -78,6 +78,7 @@ export const HOST_BRIDGE_GLOBAL_KEYS = {
 	resolveModuleSync: "_resolveModuleSync",
 	loadFileSync: "_loadFileSync",
 	ptySetRawMode: "_ptySetRawMode",
+	stdinRead: "_stdinRead",
 	processConfig: "_processConfig",
 	osConfig: "_osConfig",
 	log: "_log",

packages/nodejs/src/bridge-handlers.ts

@@ -1802,6 +1802,10 @@ export function resolveHttpServerResponse(serverId: number, responseJson: string
 export interface PtyBridgeDeps {
 	onPtySetRawMode?: (mode: boolean) => void;
 	stdinIsTTY?: boolean;
+	/** Set by _stdinRead handler — call to deliver data to the pending read */
+	onStdinData?: (data: string) => void;
+	/** Set by _stdinRead handler — call to signal stdin EOF */
+	onStdinEnd?: () => void;
 }
 
 /** Build PTY bridge handlers. */
@@ -1815,6 +1819,41 @@ export function buildPtyBridgeHandlers(deps: PtyBridgeDeps): BridgeHandlers {
 		};
 	}
 
+	// Async bridge handler for streaming stdin reads
+	if (deps.stdinIsTTY) {
+		const stdinQueue: (string | null)[] = [];
+		let stdinReadResolve: ((data: string | null) => void) | null = null;
+
+		handlers[K.stdinRead] = (): Promise<string | null> => {
+			if (stdinQueue.length > 0) {
+				return Promise.resolve(stdinQueue.shift()!);
+			}
+			return new Promise<string | null>((resolve) => {
+				stdinReadResolve = resolve;
+			});
+		};
+
+		deps.onStdinData = (data: string) => {
+			if (stdinReadResolve) {
+				const resolve = stdinReadResolve;
+				stdinReadResolve = null;
+				resolve(data);
+			} else {
+				stdinQueue.push(data);
+			}
+		};
+
+		deps.onStdinEnd = () => {
+			if (stdinReadResolve) {
+				const resolve = stdinReadResolve;
+				stdinReadResolve = null;
+				resolve(null);
+			} else {
+				stdinQueue.push(null);
+			}
+		};
+	}
+
 	return handlers;
 }
 

packages/nodejs/src/bridge/process.ts

@@ -12,6 +12,7 @@ import { URL as WhatwgURL, URLSearchParams as WhatwgURLSearchParams } from "what
 // Use buffer package for spec-compliant Buffer implementation
 import { Buffer as BufferPolyfill } from "buffer";
 import type {
+	BridgeApplyRef,
 	CryptoRandomFillBridgeRef,
 	CryptoRandomUuidBridgeRef,
 	FsFacadeBridge,
@@ -56,6 +57,8 @@ declare const _log: ProcessLogBridgeRef;
 declare const _error: ProcessErrorBridgeRef;
 // Timer reference for actual delays using host's event loop
 declare const _scheduleTimer: ScheduleTimerBridgeRef | undefined;
+// Stdin streaming read — async bridge handler returning next chunk (null = EOF)
+declare const _stdinRead: BridgeApplyRef<[], string | null> | undefined;
 declare const _cryptoRandomFill: CryptoRandomFillBridgeRef | undefined;
 declare const _cryptoRandomUUID: CryptoRandomUuidBridgeRef | undefined;
 // Filesystem bridge for chdir validation
@@ -361,6 +364,9 @@ const _stderr: StdioWriteStream = {
   rows: 24,
 };
 
+// Flag to prevent duplicate stdin read loops
+let _stdinKeepaliveActive = false;
+
 // Stdin stream with data support
 // These are exposed as globals so they can be set after bridge initialization
 type StdinListener = (data?: unknown) => void;
@@ -418,6 +424,47 @@ function _emitStdinData(): void {
   }
 }
 
+/**
+ * Global dispatch handler for streaming stdin events from the host.
+ * Called by the V8 sidecar when it receives a "stdin" stream event.
+ * Pushes data into the stdin stream in real-time for PTY-backed processes.
+ */
+const stdinDispatch = (
+  _eventType: string,
+  payload: string | null,
+): void => {
+  if (payload === null || payload === undefined) {
+    // stdin end signal
+    if (!getStdinEnded()) {
+      setStdinEnded(true);
+      const endListeners = [...(_stdinListeners["end"] || []), ...(_stdinOnceListeners["end"] || [])];
+      _stdinOnceListeners["end"] = [];
+      for (const listener of endListeners) {
+        listener();
+      }
+      const closeListeners = [...(_stdinListeners["close"] || []), ...(_stdinOnceListeners["close"] || [])];
+      _stdinOnceListeners["close"] = [];
+      for (const listener of closeListeners) {
+        listener();
+      }
+    }
+    return;
+  }
+
+  // Streaming data chunk — emit 'data' event if listeners are registered
+  const dataListeners = [...(_stdinListeners["data"] || []), ...(_stdinOnceListeners["data"] || [])];
+  _stdinOnceListeners["data"] = [];
+  if (dataListeners.length > 0) {
+    for (const listener of dataListeners) {
+      listener(payload);
+    }
+  } else {
+    // Buffer if no listeners yet — append to _stdinData for later read()
+    setStdinDataValue(getStdinData() + payload);
+  }
+};
+exposeCustomGlobal("_stdinDispatch", stdinDispatch);
+
 // Stdin stream shape
 interface StdinStream {
   readable: boolean;
@@ -500,6 +547,26 @@ const _stdin: StdinStream = {
     this.paused = false;
     setStdinFlowMode(true);
     _emitStdinData();
+    // Start streaming stdin read loop via _stdinRead bridge handler
+    if (_getStdinIsTTY() && !_stdinKeepaliveActive && typeof _stdinRead !== "undefined") {
+      _stdinKeepaliveActive = true;
+      (async function readLoop() {
+        try {
+          while (true) {
+            const chunk = await _stdinRead!.apply(undefined, [], { result: { promise: true } });
+            if (chunk === null || chunk === undefined) {
+              // EOF — dispatch end signal
+              stdinDispatch("stdin", null);
+              break;
+            }
+            stdinDispatch("stdin", chunk);
+          }
+        } catch {
+          // Bridge error — session closing
+        }
+        _stdinKeepaliveActive = false;
+      })();
+    }
     return this;
   },
 

packages/nodejs/src/execution-driver.ts

@@ -56,6 +56,7 @@ import {
 	buildUpgradeSocketBridgeHandlers,
 	buildModuleResolutionBridgeHandlers,
 	buildPtyBridgeHandlers,
+	type PtyBridgeDeps,
 	createProcessConfigForExecution,
 	resolveHttpServerResponse,
 } from "./bridge-handlers.js";
@@ -297,6 +298,10 @@ export class NodeExecutionDriver implements RuntimeDriver {
 	private flattenedBindings: FlattenedBinding[] | null = null;
 	// Unwrapped filesystem for path translation (toHostPath/toSandboxPath)
 	private rawFilesystem: VirtualFileSystem | undefined;
+	/** Callback invoked when V8 session is ready — used for streaming stdin */
+	onStreamReady?: (sendStreamEvent: (eventType: string, payload: Uint8Array) => void) => void;
+	/** Callback invoked when PTY stdin bridge handler is ready — delivers data to pending _stdinRead */
+	onStdinReady?: (deliver: (data: string) => void, end: () => void) => void;
 
 	constructor(options: NodeExecutionDriverOptions) {
 		this.memoryLimit = options.memoryLimit ?? 128;
@@ -446,6 +451,14 @@ export class NodeExecutionDriver implements RuntimeDriver {
 				}
 			};
 
+			// Notify kernel-runtime that streaming is ready (for PTY stdin)
+			this.onStreamReady?.(sendStreamEvent);
+
+			// Build PTY bridge handlers ONCE — shared between dispatch and main handlers
+			const ptyDeps: PtyBridgeDeps = { onPtySetRawMode: s.onPtySetRawMode, stdinIsTTY: s.processConfig.stdinIsTTY };
+			const ptyHandlers = buildPtyBridgeHandlers(ptyDeps);
+			if (ptyDeps.onStdinData) this.onStdinReady?.(ptyDeps.onStdinData, ptyDeps.onStdinEnd!);
+
 			const netSocketResult = buildNetworkSocketBridgeHandlers({
 				dispatch: (socketId, event, data) => {
 					const payload = JSON.stringify({ socketId, event, data });
@@ -488,10 +501,7 @@ export class NodeExecutionDriver implements RuntimeDriver {
 							return typeof fs.toSandboxPath === "function" ? fs.toSandboxPath(p) : p;
 						},
 					}),
-					...buildPtyBridgeHandlers({
-						onPtySetRawMode: s.onPtySetRawMode,
-						stdinIsTTY: s.processConfig.stdinIsTTY,
-					}),
+					...ptyHandlers,
 					// Custom bindings dispatched through _loadPolyfill
 					...(this.flattenedBindings ? Object.fromEntries(
 						this.flattenedBindings.map(b => [b.key, b.handler])
@@ -543,10 +553,7 @@ export class NodeExecutionDriver implements RuntimeDriver {
 						return typeof rfs?.toSandboxPath === "function" ? rfs.toSandboxPath(p) : p;
 					},
 				}),
-				...buildPtyBridgeHandlers({
-					onPtySetRawMode: s.onPtySetRawMode,
-					stdinIsTTY: s.processConfig.stdinIsTTY,
-				}),
+				...ptyHandlers,
 			};
 
 			// Merge custom bindings into bridge handlers

packages/nodejs/src/kernel-runtime.ts

@@ -353,28 +353,54 @@ class NodeRuntimeDriver implements RuntimeDriver {
       };
     });
 
-    // Stdin buffering — writeStdin collects data, closeStdin resolves the promise
+    // Stdin plumbing — two modes:
+    // 1. Batched (non-PTY): collect chunks, closeStdin concatenates and resolves promise
+    // 2. Streaming (PTY): deliver each writeStdin chunk via _stdinRead bridge handler
+    const isPty = ctx.stdinIsTTY ?? false;
     const stdinChunks: Uint8Array[] = [];
     let stdinResolve: ((data: string | undefined) => void) | null = null;
+    // Callbacks set by _stdinRead bridge handler via onStdinReady
+    let stdinDeliverFn: ((data: string) => void) | null = null;
+    let stdinEndFn: (() => void) | null = null;
     const stdinPromise = new Promise<string | undefined>((resolve) => {
       stdinResolve = resolve;
-      // Auto-resolve on next microtask if nobody calls writeStdin
-      queueMicrotask(() => {
-        if (stdinChunks.length === 0 && stdinResolve) {
-          stdinResolve = null;
-          resolve(undefined);
-        }
-      });
+      if (isPty) {
+        // PTY mode: resolve immediately with no initial stdin data
+        stdinResolve = null;
+        resolve(undefined);
+      } else {
+        // Non-PTY: auto-resolve on next microtask if nobody calls writeStdin
+        queueMicrotask(() => {
+          if (stdinChunks.length === 0 && stdinResolve) {
+            stdinResolve = null;
+            resolve(undefined);
+          }
+        });
+      }
     });
 
     const proc: DriverProcess = {
       onStdout: null,
       onStderr: null,
       onExit: null,
       writeStdin: (data: Uint8Array) => {
-        stdinChunks.push(data);
+        if (isPty && stdinDeliverFn) {
+          // Streaming mode: deliver data to sandbox via _stdinRead bridge handler
+          const text = new TextDecoder().decode(data);
+          stdinDeliverFn(text);
+        } else if (isPty) {
+          // Bridge handler not ready yet — buffer for flush when handler connects
+          stdinChunks.push(data);
+        } else {
+          // Non-PTY batched mode
+          stdinChunks.push(data);
+        }
       },
       closeStdin: () => {
+        if (isPty && stdinEndFn) {
+          stdinEndFn();
+          return;
+        }
         if (stdinResolve) {
           if (stdinChunks.length === 0) {
             // No data written — pass undefined (no stdin), not empty string
@@ -400,8 +426,20 @@ class NodeRuntimeDriver implements RuntimeDriver {
       wait: () => exitPromise,
     };
 
+    // Callback to wire up streaming stdin once _stdinRead bridge handler is ready
+    const setStdinBridge = (deliver: (data: string) => void, end: () => void) => {
+      stdinDeliverFn = deliver;
+      stdinEndFn = end;
+      // Flush any data that arrived before the bridge handler was ready
+      for (const chunk of stdinChunks) {
+        const text = new TextDecoder().decode(chunk);
+        deliver(text);
+      }
+      stdinChunks.length = 0;
+    };
+
     // Launch async — spawn() returns synchronously per RuntimeDriver contract
-    this._executeAsync(command, args, ctx, proc, resolveExit, stdinPromise);
+    this._executeAsync(command, args, ctx, proc, resolveExit, stdinPromise, isPty ? setStdinBridge : undefined);
 
     return proc;
   }
@@ -425,6 +463,7 @@ class NodeRuntimeDriver implements RuntimeDriver {
     proc: DriverProcess,
     resolveExit: (code: number) => void,
     stdinPromise: Promise<string | undefined>,
+    setStdinBridge?: (deliver: (data: string) => void, end: () => void) => void,
   ): Promise<void> {
     const kernel = this._kernel!;
 
@@ -483,6 +522,12 @@ class NodeRuntimeDriver implements RuntimeDriver {
         bindings: this._bindings,
         onPtySetRawMode,
       });
+
+      // Wire streaming stdin for PTY processes via _stdinRead bridge handler
+      if (setStdinBridge) {
+        executionDriver.onStdinReady = setStdinBridge;
+      }
+
       this._activeDrivers.set(ctx.pid, executionDriver);
 
       // Detect ESM files and use V8 native module system