Repository: denoland/deno
Pull request: fix(ext/node): forward stdout from worker_threads to parent

Closes #31889

## Summary
- Adds `stdout` and `stderr` as `Readable` streams on the `Worker` class, and `stdin` (null by default), matching the Node.js API surface
- Forwards `process.stdout.write()` and `process.stderr.write()` from worker threads to the parent via internal messages, so `worker.stdout`/`worker.stderr` emit data and support `.pipe()`
- Closes stdio streams with `push(null)` when the worker exits
- Fixes `execArgv` validation to skip non-flag arguments (e.g. `"node"` and `"development"` in `--conditions node --conditions development`)
- Enables `test-worker-no-stdin-stdout-interaction.js` node_compat test

## Test plan
- [x] `cargo test --test node_compat parallel -- --filter test-worker-no-stdin-stdout-interaction.js`
- [x] `cargo test --test node_compat parallel -- --filter test-worker-execargv` (no regression)
- [x] Vitest 4 threads pool works end-to-end with [reproduction repo](https://github.com/ignatij/deno-vitest-4-compatibility-issue)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Changed files:
filename: ext/node/polyfills/worker_threads.ts
patch: @@ -38,6 +38,7 @@ import {
   validateObject,
 } from "ext:deno_node/internal/validators.mjs";
 import { EventEmitter } from "node:events";
+import { Readable } from "node:stream";
 import {
   BroadcastChannel as WebBroadcastChannel,
   refBroadcastChannel,
@@ -97,17 +98,23 @@ function isWorkerOnlineMsg(data: unknown): data is WorkerOnlineMsg {
     (data as { "type": unknown })["type"] === "WORKER_ONLINE";
 }
 
-interface WorkerStderrMsg {
-  type: "WORKER_STDERR";
+interface WorkerStdioMsg {
+  type: "WORKER_STDERR" | "WORKER_STDOUT";
   data: string;
 }
 
-function isWorkerStderrMsg(data: unknown): data is WorkerStderrMsg {
+function isWorkerStderrMsg(data: unknown): data is WorkerStdioMsg {
   return typeof data === "object" && data !== null &&
     ObjectHasOwn(data, "type") &&
     (data as { "type": unknown })["type"] === "WORKER_STDERR";
 }
 
+function isWorkerStdoutMsg(data: unknown): data is WorkerStdioMsg {
+  return typeof data === "object" && data !== null &&
+    ObjectHasOwn(data, "type") &&
+    (data as { "type": unknown })["type"] === "WORKER_STDOUT";
+}
+
 // Flags that are valid Node.js environment flags but not allowed in workers
 // because they affect per-process state.
 const workerDisallowedFlags = new SafeSet([
@@ -176,8 +183,12 @@ class NodeWorker extends EventEmitter {
     codeRangeSizeMb: -1,
     stackSizeMb: 4,
   };
+  // https://nodejs.org/api/worker_threads.html#workerstdin
+  stdin = null;
+  // https://nodejs.org/api/worker_threads.html#workerstdout
+  stdout: Readable = new Readable({ read() {} });
   // https://nodejs.org/api/worker_threads.html#workerstderr
-  stderr: EventEmitter = new EventEmitter();
+  stderr: Readable = new Readable({ read() {} });
 
   constructor(specifier: URL | string, options?: WorkerOptions) {
     super();
@@ -188,6 +199,11 @@ class NodeWorker extends EventEmitter {
         const invalidFlags = [];
         for (let i = 0; i < options.execArgv.length; i++) {
           const flag = options.execArgv[i];
+          // Items that don't start with '-' are arguments to the
+          // preceding flag (e.g. "--conditions node"), not flags.
+          if (!StringPrototypeStartsWith(flag, "-")) {
+            continue;
+          }
           if (!process.allowedNodeEnvironmentFlags.has(flag)) {
             invalidFlags[invalidFlags.length] = flag;
             continue;
@@ -392,6 +408,15 @@ class NodeWorker extends EventEmitter {
     this.emit("error", err);
   }
 
+  #closeStdio() {
+    if (!this.stdout.readableEnded) {
+      FunctionPrototypeCall(Readable.prototype.push, this.stdout, null);
+    }
+    if (!this.stderr.readableEnded) {
+      FunctionPrototypeCall(Readable.prototype.push, this.stderr, null);
+    }
+  }
+
   #pollControl = async () => {
     while (this.#status === "RUNNING") {
       this.#controlPromise = op_host_recv_ctrl(this.#id);
@@ -408,6 +433,7 @@ class NodeWorker extends EventEmitter {
       switch (type) {
         case 1: { // TerminalError
           this.#status = "CLOSED";
+          this.#closeStdio();
           if (this.listenerCount("error") > 0) {
             const err = new Error(data.errorMessage ?? data.message);
             if (data.name) {
@@ -431,6 +457,7 @@ class NodeWorker extends EventEmitter {
         case 3: { // Close
           debugWT(`Host got "close" message from worker: ${this.#name}`);
           this.#status = "CLOSED";
+          this.#closeStdio();
           if (!this.#exited) {
             this.#exited = true;
             this.emit("exit", data ?? 0);
@@ -471,8 +498,18 @@ class NodeWorker extends EventEmitter {
       ) {
         this.#workerOnline = true;
         this.emit("online");
+      } else if (isWorkerStdoutMsg(message)) {
+        FunctionPrototypeCall(
+          Readable.prototype.push,
+          this.stdout,
+          message.data,
+        );
       } else if (isWorkerStderrMsg(message)) {
-        this.stderr.emit("data", message.data);
+        FunctionPrototypeCall(
+          Readable.prototype.push,
+          this.stderr,
+          message.data,
+        );
       } else {
         this.emit("message", message);
       }
@@ -517,6 +554,7 @@ class NodeWorker extends EventEmitter {
 
     this.#status = "TERMINATED";
     op_host_terminate_worker(this.#id);
+    this.#closeStdio();
 
     if (!this.#exited) {
       this.#exited = true;
@@ -714,6 +752,26 @@ internals.__initWorkerThreads = (
           }
         }
 
+        // Forward stdout writes to the parent so worker.stdout
+        // is readable from the host side.
+        const origStdoutWrite = FunctionPrototypeBind(
+          process.stdout.write,
+          process.stdout,
+        );
+        process.stdout.write = function (chunk, encoding, callback) {
+          parentPort.postMessage({
+            type: "WORKER_STDOUT",
+            data: typeof chunk === "string" ? chunk : String(chunk),
+          });
+          return FunctionPrototypeCall(
+            origStdoutWrite,
+            process.stdout,
+            chunk,
+            encoding,
+            callback,
+          );
+        };
+
         // Forward stderr writes to the parent so worker.stderr
         // is readable from the host side.
         const origStderrWrite = FunctionPrototypeBind(

filename: tests/node_compat/config.jsonc
patch: @@ -1578,6 +1578,7 @@
     "parallel/test-worker-mjs-workerdata.js": {},
     // TODO(bartlomieju): disabled during work on `node:inspector`, this test didn't actualy run before
     // "parallel/test-worker-name.js": {},
+    "parallel/test-worker-no-stdin-stdout-interaction.js": {},
     "parallel/test-worker-on-process-exit.js": {},
     "parallel/test-worker-onmessage.js": {},
     "parallel/test-worker-process-argv.js": {},