Compile NodeJS.ReadableStream and NodeJS.WritableStream as general streams when a program mixes stream producers (#1518)

docs/content/limitations/builtin-apis/streams-and-readline.mdx

@@ -1,4 +1,7 @@
 - **Streams and readline:** Stream chunks are Buffers or strings.
+  `NodeJS.ReadableStream` and `NodeJS.WritableStream`, and `stream.Readable` and `stream.Writable` under `@types/node`, accept every stream: when a program passes file, zlib, or `node:stream` streams through them, child-process pipes and `process.stdout` and `process.stderr` join those streams as `Readable` and `Writable` instances.
+  `readable.pipe(process.stdout)` and `readable.pipe(process.stderr)` compile and, as in Node, do not end the process stream.
+  A `pipeline` stage that arrives as an untyped value must be a native stream at run time; other values throw a not-supported error.
   `Readable.from` accepts arrays, Sets, Map and Set iterators, generators, async generators, and class iterables and async iterables whose values are strings or Buffers, and in JavaScript any iterable or async iterable of them; a stream that mixes strings and Buffers is unsupported.
   In JavaScript, a sync iterator value with a `then` method that is not a promise, and an object whose `[Symbol.asyncIterator]` or `[Symbol.iterator]` method is a generator method, are unsupported.
   Object-mode streams of other values (including `Readable.from` over a `Map`, whose values are entry arrays) are unsupported.
@@ -11,6 +14,7 @@
   `stream.compose` accepts two or more streams, or one `Duplex`; functions, iterables, and web streams as stages are unsupported, and the `ERR_INVALID_ARG_VALUE` message for a stream that is no longer readable or writable names the stream's class instead of printing the stream.
   `pipeline` from `node:stream/promises` accepts a trailing `{ signal, end }` object, and `finished` accepts `readable`, `writable`, `error`, `cleanup`, and `signal` options, with boolean literals for the flags.
   Stream constructors accept a `signal` option in an options object literal, and `stream.addAbortSignal` accepts any stream.
+  A `transform` or `flush` option callback whose completion callback types `data` as `any`, as @types/node's `TransformCallback` does, may pass a Buffer, a string, or nothing; any other value throws a TypeError where the callback is called, instead of Node's `ERR_INVALID_ARG_TYPE` stream error.
   `node:readline` and `node:readline/promises` (also reached as `readline.promises`) interfaces, from `createInterface(options)` or `createInterface(input, output, completer, terminal)`, read `process.stdin` or a readable stream: `question`, `prompt`, `write` (with a key object literal of `name`, `ctrl`, `meta`, and `shift`), `pause`, `resume`, `'line'`/`'close'`/`'pause'`/`'resume'`/`'error'`/`'history'`/`'SIGINT'`/`'SIGTSTP'`/`'SIGCONT'` listeners, and `for await` compile.
   The `history`, `historySize`, `removeHistoryDuplicates`, `completer`, `tabSize`, and `escapeCodeTimeout` options are validated as in Node.
   With `terminal: true`, or with no `terminal` option and `process.stdout` as the output when it is a TTY, the interface runs Node's terminal mode: the editing keys, the kill ring, undo and redo, history recording and recall, Tab completion through the completer, Ctrl+C, Ctrl+D, and Ctrl+Z, `TERM=dumb`, and the same escape sequences written to the output.

docs/content/limitations/child-process-workers.mdx

@@ -12,3 +12,7 @@ Message listeners may return `void` or `Promise<void>`.
 Messages use newline-delimited JSON, so their static types must be JSON-stringifiable or `unknown` backed by a checked JSON tree.
 `send()` returns the channel's backpressure state and invokes its callback later with `null` or an `Error`.
 `disconnect()` changes `connected` synchronously, before the later disconnect events.
+
+A `'data'` listener on a child's `stdout` or `stderr` whose chunk is `unknown` or `any` receives strings when a `setEncoding()` statement on the same stream precedes it in its block or an enclosing block, and Buffers when the function never calls `setEncoding()` on that stream.
+Any other placement of `setEncoding()` is refused at compile time, and a `setEncoding()` call that runs elsewhere afterwards makes the listener throw a TypeError instead of receiving strings.
+An inline `'exit'` or `'close'` listener may declare `(code: number, signal: string)`, which TypeScript accepts through the catch-all `on` overload; it receives the `null` values Node passes, and a typed use of a `null` value throws a TypeError.

docs/content/limitations/console-and-process-streams.mdx

@@ -9,4 +9,5 @@ JavaScript can store `process.stdin`, `stdout`, and `stderr`, replace output `wr
 Input supports data/end/error listeners, pause/resume, buffered reads, and TTY raw mode.
 Output writes are synchronous, with successful callbacks on the next-tick queue; queued backpressure is not modeled.
 Stored streams do not support piping, encoding changes, arbitrary properties, or other events.
-TypeScript output parameters support `write(string)`.
+A TypeScript `NodeJS.WritableStream` parameter that only receives `process.stdout` or `process.stderr` supports `write(string)`.
+When other streams flow into the same parameter, the process streams join them as `Writable` instances with the full writable surface.

packages/compiler/ambient/scriptc-node-fallback.d.ts

@@ -80,21 +80,54 @@ declare namespace NodeJS {
     path?: string | undefined;
     syscall?: string | undefined;
   }
-  /* The piped child-output stream (child.stdout / child.stderr under
-   * stdio ["ignore", "pipe", "pipe"]): 'data' fires one Buffer chunk per
-   * read (≤64KB, consumer-driven — no listener, no read), 'end' fires
-   * once at EOF and always BEFORE the child's 'exit'. A flowing stream
-   * keeps the loop alive. */
+  /* The readable-stream interface every Node stream implements. The
+   * piped child-output stream (child.stdout / child.stderr under stdio
+   * ["ignore", "pipe", "pipe"]) is declared with it: 'data' fires one
+   * Buffer chunk per read (≤64KB, consumer-driven — no listener, no
+   * read), 'end' fires once at EOF and always BEFORE the child's 'exit'. A
+   * flowing stream keeps the loop alive. A program that also passes
+   * other streams (fs, zlib, node:stream) where this interface is
+   * expected compiles it as the node:stream Readable, with child output
+   * bridged into one. */
   interface ReadableStream {
     on(event: "data", listener: (chunk: Buffer) => void): void;
     on(event: "data", listener: (chunk: string) => void): void;
     on(event: "end", listener: () => void): void;
+    on(event: string, listener: (...args: any[]) => void): void;
     once(event: "data", listener: (chunk: Buffer) => void): void;
     once(event: "data", listener: (chunk: string) => void): void;
     once(event: "end", listener: () => void): void;
+    once(event: string, listener: (...args: any[]) => void): void;
     /* setEncoding('utf8') makes subsequent data chunks strings, including
      * multibyte sequences split across native pipe reads. */
     setEncoding(encoding: string): this;
+    readonly readable: boolean;
+    pause(): this;
+    resume(): this;
+    isPaused(): boolean;
+    pipe<T extends WritableStream>(destination: T, options?: { end?: boolean }): T;
+    unpipe(destination?: WritableStream): this;
+    [Symbol.asyncIterator](): AsyncIterableIterator<string | Buffer>;
+  }
+  /* The writable-stream interface every Node stream implements:
+   * process.stdout/process.stderr, fs.WriteStream and the node:stream
+   * Writable family. Written only to the process streams, it compiles as
+   * their descriptor; with other streams flowing into it, as the
+   * node:stream Writable (the process streams convert into one). */
+  interface WritableStream {
+    readonly writable: boolean;
+    write(data: string | Uint8Array, callback?: (error?: Error | null) => void): boolean;
+    write(
+      data: string | Uint8Array,
+      encoding: BufferEncoding,
+      callback?: (error?: Error | null) => void,
+    ): boolean;
+    end(callback?: () => void): void;
+    end(data: string | Uint8Array, callback?: () => void): void;
+    end(data: string | Uint8Array, encoding: BufferEncoding, callback?: () => void): void;
+    /* Node's EventEmitter signature: listeners take untyped arguments. */
+    on(event: string, listener: (...args: any[]) => void): void;
+    once(event: string, listener: (...args: any[]) => void): void;
   }
 }
 
@@ -3066,8 +3099,10 @@ declare module "child_process" {
     destroy(): void;
     on(event: "drain" | "finish", listener: () => void): void;
     on(event: "error", listener: (err: Error) => void): void;
+    on(event: string, listener: (...args: any[]) => void): void;
     once(event: "drain" | "finish", listener: () => void): void;
     once(event: "error", listener: (err: Error) => void): void;
+    once(event: string, listener: (...args: any[]) => void): void;
   }
 
   export interface ChildProcess extends Disposable {
@@ -6414,7 +6449,7 @@ declare module "node:cluster" {
 /* node:tty — the fd probe (the same isatty(3) behind process.*.isTTY)
  * and the supported structural view of a process output stream. */
 declare module "tty" {
-  export interface WriteStream {
+  export interface WriteStream extends NodeJS.WritableStream {
     write(data: string | Uint8Array, callback?: (error?: Error | null) => void): boolean;
     write(
       data: string | Uint8Array,
@@ -6664,8 +6699,8 @@ declare module "stream" {
     resume(): this;
     isPaused(): boolean;
     setEncoding(encoding: string): this;
-    pipe<T extends Writable>(destination: T, options?: { end?: boolean }): T;
-    unpipe(destination?: Writable): this;
+    pipe<T extends NodeJS.WritableStream>(destination: T, options?: { end?: boolean }): T;
+    unpipe(destination?: NodeJS.WritableStream): this;
     destroy(error?: Error): this;
     readonly readable: boolean;
     readonly readableEnded: boolean;

packages/compiler/src/backend/llvm/emitter.ts

@@ -182,6 +182,7 @@ import {
   genResultThunkFor,
   childExitThunkFor,
   childExitSignalThunkFor,
+  childDataDynThunkFor,
   childDataThunkFor,
   execFileThunkFor,
   ipcMessageThunkFor,
@@ -5576,6 +5577,10 @@ export class LlEmitter {
     return childDataThunkFor(this, param);
   }
 
+  childDataDynThunkFor(chunk: "bytes" | "str"): string {
+    return childDataDynThunkFor(this, chunk);
+  }
+
   emitterFixedAdapter(cbT: IrType & { kind: "func" }): { fn: string; shim: string } {
     return emitterFixedAdapter(this, cbT);
   }

packages/compiler/src/backend/llvm/expr-callbacks.ts

@@ -771,6 +771,31 @@ export function childDataThunkFor(host: LlvmEmitterContext, param: IrType): stri
   return sym;
 }
 
+/** The child-stream 'data' thunk for an unknown/any chunk parameter: the
+ * runtime's Buffer (or, after setEncoding, string) chunk boxes into a
+ * checked-dynamic value the listener owns. */
+export function childDataDynThunkFor(host: LlvmEmitterContext, chunk: "bytes" | "str"): string {
+  const key = `cdd:${chunk}`;
+  let sym = host.resolveThunks.get(key);
+  if (sym) return sym;
+  sym = `sc_cdd_${host.resolveThunks.size}`;
+  host.resolveThunks.set(key, sym);
+  const box = chunk === "bytes" ? "scr_dyn_new_buffer" : "scr_dyn_new_str";
+  host.declare(`declare ptr @${box}(ptr)`);
+  host.resolveThunkDefs.push(
+    `define internal void @${sym}(ptr %cb, ptr %chunk) ${FN_ATTRS} { ; child 'data' → dyn (${chunk})`,
+    `entry:`,
+    `  %d = call ptr @${box}(ptr %chunk)`,
+    `  %fnp = getelementptr inbounds %ScrClosure, ptr %cb, i64 0, i32 1`,
+    `  %fn = load ptr, ptr %fnp`,
+    `  call void %fn(ptr %cb, ptr %d)`,
+    `  ret void`,
+    `}`,
+    ``,
+  );
+  return sym;
+}
+
 export function emitterFixedAdapter(
   host: LlvmEmitterContext,
   cbT: IrType & { kind: "func" },

packages/compiler/src/backend/llvm/expr-stream-callbacks.ts

@@ -17,7 +17,7 @@ export function streamDataAdapter(
   const p = cbT.params[0];
   if (
     cbT.params.length > 1 ||
-    (p && p.kind !== "bytes" && p.kind !== "string" && p.kind !== "dyn")
+    (p && p.kind !== "bytes" && p.kind !== "string" && p.kind !== "dyn" && p.kind !== "union")
   ) {
     throw new InternalCompilerError(
       "llvm emitter bug: stream data listener param shape (frontend must fence)",
@@ -73,6 +73,46 @@ export function streamDataAdapter(
     );
     finish("%r0");
     d.push(`}`, ``);
+  } else if (p.kind === "union") {
+    // A `Buffer | string` chunk: wrap whichever slot the stream filled at
+    // its arm of the (boxed) union.
+    const def = host.unionsById.get(p.unionId);
+    const bytesTag = def ? def.arms.findIndex((a) => a.kind === "bytes" && a.elem === "u8") : -1;
+    const strTag = def ? def.arms.findIndex((a) => a.kind === "string") : -1;
+    if (
+      !def ||
+      host.nullableUnions.has(p.unionId) ||
+      bytesTag < 0 ||
+      strTag < 0 ||
+      def.arms.length !== 2
+    )
+      throw new InternalCompilerError(
+        "llvm emitter bug: stream data listener union is not Buffer | string (frontend must fence)",
+      );
+    host.declare(`declare ptr @scr_union_new_ref(i32, ptr, ptr, ptr, ptr)`);
+    host.declare(`declare ptr @scr_bytes_retain_v(ptr)`);
+    host.declare(`declare void @scr_bytes_release_v(ptr)`);
+    host.declare(`declare ptr @scr_str_retain_v(ptr)`);
+    host.declare(`declare void @scr_str_release_v(ptr)`);
+    d.push(
+      `  %uslot = alloca ptr`,
+      `  %isb = icmp ne ptr %a0, null`,
+      `  br i1 %isb, label %buf, label %str`,
+      `buf:`,
+      `  %rb = call ptr @scr_bytes_retain_v(ptr %a0)`,
+      `  %ub = call ptr @scr_union_new_ref(i32 ${bytesTag}, ptr %rb, ptr @scr_bytes_retain_v, ptr @scr_bytes_release_v, ptr null)`,
+      `  store ptr %ub, ptr %uslot`,
+      `  br label %go`,
+      `str:`,
+      `  %rs = call ptr @scr_str_retain_v(ptr %a1)`,
+      `  %us = call ptr @scr_union_new_ref(i32 ${strTag}, ptr %rs, ptr @scr_str_retain_v, ptr @scr_str_release_v, ptr null)`,
+      `  store ptr %us, ptr %uslot`,
+      `  br label %go`,
+      `go:`,
+      `  %uv = load ptr, ptr %uslot`,
+    );
+    finish("%uv");
+    d.push(`}`, ``);
   } else {
     // dyn: box by runtime tag — the JS lane's adapter parameter.
     host.declare(`declare ptr @scr_dyn_new_buffer(ptr)`);

packages/compiler/src/backend/llvm/lib-process.ts

@@ -305,6 +305,28 @@ export function emitChildProcessLibCall(host: LlvmEmitterContext, e: LibCallExpr
     B.line(`${raw} = call ptr @${get}(ptr ${recv.name}) ; +1 or NULL`);
     return host.wrapNullable(raw, raw, def!.arms[streamTag]!, streamTag, e.type, nullTag);
   }
+  if (e.fn === "child.stdioStream") {
+    // The general stream representation's stdio read: the child's bridged
+    // runtime stream (+1) or NULL for an unpiped slot.
+    if (e.type.kind !== "union")
+      throw new InternalCompilerError(`llvm emitter bug: ${e.fn} result is not a union`);
+    const def = host.unionsById.get(e.type.unionId);
+    const streamTag = def ? def.arms.findIndex((a) => a.kind === "object") : -1;
+    const nullTag = def ? def.arms.findIndex((a) => a.kind === "nullT") : -1;
+    const fd = e.args[1];
+    if (streamTag < 0 || nullTag < 0 || fd?.kind !== "numLit")
+      throw new InternalCompilerError(`llvm emitter bug: ${e.fn} lacks its arms or slot`);
+    const recv = host.emitExpr(e.args[0]!);
+    const key = fd.value === 0 ? "stdin" : fd.value === 1 ? "stdout" : "stderr";
+    host.declare(`declare ptr @scr_stream_child_stdio(ptr, ptr, ${host.sizeType})`);
+    const raw = B.tmp();
+    B.line(
+      `${raw} = call ptr @scr_stream_child_stdio(ptr ${recv.name}, ptr ${host.cstr(key)}, ${host.sizeType} ${key.length}) ; +1 or NULL`,
+    );
+    const out = host.wrapNullable(raw, raw, def!.arms[streamTag]!, streamTag, e.type, nullTag);
+    host.emitPendingCheck();
+    return out;
+  }
   if (e.fn === "writer.writeString" || e.fn === "writer.writeBytes") {
     const writer = host.emitExpr(e.args[0]!);
     const data = host.emitExpr(e.args[1]!);

packages/compiler/src/backend/llvm/lib-shared.ts

@@ -727,6 +727,8 @@ export const LIB_FN_SYMS: Record<string, string> = {
   "child.unref": "scr_child_unref",
   "child.ref": "scr_child_ref",
   "procStream.write": "scr_proc_stream_write",
+  "stream.stdioWritable": "scr_stream_stdio_writable",
+  "stream.unboxDyn": "scr_stream_unbox_dyn",
   "process.kill": "scr_process_kill_named",
   "process.killNum": "scr_process_kill",
   // fs.promises (scr_bytes_io.c / scr_lib.c): the promise forms settle
@@ -1120,6 +1122,7 @@ export const LIB_FN_SYMS: Record<string, string> = {
   "readable.opFind": "scr_stream_op_find",
   "readable.opReduce": "scr_stream_op_reduce",
   "readable.pipe": "scr_stream_pipe",
+  "readable.pipeGeneral": "scr_stream_pipe_general",
   "writable.cork": "scr_stream_cork",
   "writable.uncork": "scr_stream_uncork",
   "stream.destroyErr": "scr_stream_destroy",
@@ -1606,6 +1609,7 @@ export const USES_TIMERS_LIB_FNS = new Set<string>([
   "readable.opReduce",
   "readable.resume",
   "readable.pipe",
+  "readable.pipeGeneral",
   "readable.unpipe",
   "writable.write",
   "writable.writeStr",

packages/compiler/src/backend/llvm/lib-stream.ts

@@ -56,6 +56,8 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV
       host.declare(`declare void @scr_child_stream_thunk0(ptr, ptr)`);
     } else if (param.kind === "union") {
       adapter = host.childDataThunkFor(param);
+    } else if (param.kind === "dyn") {
+      adapter = host.childDataDynThunkFor("bytes");
     } else {
       adapter = "scr_child_stream_thunk_bytes";
       host.declare(`declare void @scr_child_stream_thunk_bytes(ptr, ptr)`);
@@ -78,7 +80,8 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV
   if (e.fn === "stream.onDataStr") {
     host.usesTimers = true;
     const cbT = e.args[1]!.type;
-    if (cbT.kind !== "func" || cbT.params[0]?.kind !== "string") {
+    const strParam = cbT.kind === "func" ? cbT.params[0]?.kind : undefined;
+    if (strParam !== "string" && strParam !== "dyn") {
       throw new InternalCompilerError(
         "llvm emitter bug: stream.onDataStr callback not a string func",
       );
@@ -87,10 +90,12 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV
     const cb = host.emitExpr(e.args[1]!);
     const once = host.emitExpr(e.args[2]!);
     host.toHeap(cb);
-    host.declare(`declare void @scr_child_stream_thunk_str(ptr, ptr)`);
+    let adapter = "scr_child_stream_thunk_str";
+    if (strParam === "dyn") adapter = host.childDataDynThunkFor("str");
+    else host.declare(`declare void @scr_child_stream_thunk_str(ptr, ptr)`);
     host.declare(`declare void @scr_child_stream_on_data_str(ptr, ptr, ptr, i1 zeroext)`);
     B.line(
-      `call void @scr_child_stream_on_data_str(ptr ${s0.name}, ptr ${cb.name}, ptr @scr_child_stream_thunk_str, i1 ${once.name})`,
+      `call void @scr_child_stream_on_data_str(ptr ${s0.name}, ptr ${cb.name}, ptr @${adapter}, i1 ${once.name})`,
     );
     return { name: "", type: e.type };
   }

packages/compiler/src/frontend/lowering/builtins/child-process-methods.ts

@@ -1,5 +1,6 @@
 import { boolLit, strLit } from "../../../ir/build.js";
 import * as ts from "../../ts7/adapter.js";
+import { requestGeneralStreamSurface } from "../general-streams.js";
 import { type Lowerer } from "../lowerer.js";
 import { isJsSourceFile, locOf } from "../../program.js";
 import { isChildSurfaceMember } from "../surfaces.js";
@@ -131,8 +132,16 @@ export function lowerChildMethodCall(
     ) {
       for (const param of listenerNode.parameters.nodes) lowerer.checkedCallbackParams.add(param);
     }
-    const cb = lowerer.lowerExpr(call.arguments.nodes[1]!);
+    const narrowed =
+      event === "exit" || event === "close"
+        ? narrowedExitListenerParams(lowerer, listenerNode)
+        : [];
+    const cb =
+      narrowed.length > 0
+        ? lowerWithCheckedParams(lowerer, listenerNode, narrowed)
+        : lowerer.lowerExpr(call.arguments.nodes[1]!);
     if (
+      narrowed.length > 0 ||
       (isJsSourceFile(listenerNode.getSourceFile()) && cb.type.kind === "dyn") ||
       (cb.type.kind === "func" &&
         (cb.type.rest ||
@@ -301,10 +310,12 @@ export function lowerChildMethodCall(
 
 /** Method calls on piped child-output stream receivers (child.stdout /
  * child.stderr — the childStream kind): on/once("data" | "end").
- * 'data' listeners take zero parameters, `(chunk: Buffer)`, or a
- * `Buffer | string`-union chunk (the ngrok appendOutput shape — the
- * runtime only ever fires Buffers; the compiler-emitted adapter wraps
- * the chunk at the union's Buffer arm); 'end' listeners take none.
+ * 'data' listeners take zero parameters, `(chunk: Buffer)`, `(chunk:
+ * string)`, an unknown/any chunk (boxed as a checked-dynamic value; see
+ * childStreamEncodedAt), or a `Buffer | string`-union chunk (the ngrok
+ * appendOutput shape — the runtime only ever fires Buffers; the
+ * compiler-emitted adapter wraps the chunk at the union's Buffer arm);
+ * 'end' listeners take none.
  * Statement position only (Node returns the stream for chaining; here
  * the result is void). Chained receivers (`child.stdout?.on(...)`)
  * ride the optional-chain re-dispatch (chainBlocked). Everything else
@@ -335,6 +346,7 @@ export function lowerChildStreamMethodCall(
     const evT = lowerer.typeOf(call.arguments.nodes[0]!);
     const event = evT.isStringLiteralType() ? evT.value : null;
     if (event !== "data" && event !== "end") {
+      requestGeneralStreamSurface(lowerer, access.expression);
       lowerer.noLowering(
         `stream.${name}(${event === null ? "non-literal event" : `"${event}"`}, ...)`,
         call.arguments.nodes[0]!,
@@ -342,6 +354,7 @@ export function lowerChildStreamMethodCall(
       );
     }
     if (!ts.isExpressionStatement(call.parent)) {
+      requestGeneralStreamSurface(lowerer, access.expression);
       lowerer.unsupported(
         "SC1090",
         call,
@@ -376,23 +389,32 @@ export function lowerChildStreamMethodCall(
     if (
       param !== undefined &&
       param.kind !== "string" &&
+      param.kind !== "dyn" &&
       !(param.kind === "bytes" && param.elem === "u8") &&
       !unionOk(param)
     ) {
       lowerer.unsupported(
         "SC1090",
         call.arguments.nodes[1]!,
-        `data listeners whose parameter is not 'Buffer', 'string', or a Buffer-armed union (got '${lowerer.fmt(param)}')`,
+        `data listeners whose parameter is not 'Buffer', 'string', 'unknown', or a Buffer-armed union (got '${lowerer.fmt(param)}')`,
       );
     }
+    // An unknown/any chunk is a Buffer, or a string once setEncoding ran.
+    // The registration takes whichever the enclosing code establishes; a
+    // later setEncoding that contradicts it throws the runtime's
+    // listener-type TypeError instead of delivering the other kind.
+    const strings =
+      param?.kind === "string" ||
+      (param?.kind === "dyn" && childStreamEncodedAt(lowerer, access.expression, call));
     return {
       kind: "libCall",
-      fn: param?.kind === "string" ? "stream.onDataStr" : "stream.onData",
+      fn: strings ? "stream.onDataStr" : "stream.onData",
       args: [receiver, cb, once],
       type: VOID,
       loc,
     };
   }
+  requestGeneralStreamSurface(lowerer, access.expression);
   lowerer.noLowering(
     `ReadableStream.${name}`,
     call,
@@ -418,6 +440,7 @@ export function lowerChildWriterMethodCall(
 
   if (name === "write") {
     if (call.arguments.nodes.length !== 1 || call.arguments.nodes.some(ts.isSpreadElement)) {
+      requestGeneralStreamSurface(lowerer, access.expression);
       lowerer.noLowering(
         `child stdin write with ${call.arguments.nodes.length} arguments`,
         call,
@@ -461,6 +484,7 @@ export function lowerChildWriterMethodCall(
     const eventType = lowerer.typeOf(call.arguments.nodes[0]!);
     const event = eventType.isStringLiteralType() ? eventType.value : null;
     if (event !== "drain" && event !== "finish" && event !== "error") {
+      requestGeneralStreamSurface(lowerer, access.expression);
       lowerer.noLowering(
         `child stdin ${name}(${event === null ? "non-literal event" : `"${event}"`}, ...)`,
         call.arguments.nodes[0]!,
@@ -507,6 +531,7 @@ export function lowerChildWriterMethodCall(
     return { kind: "libCall", fn, args: [receiver, cb, once], type: VOID, loc };
   }
 
+  requestGeneralStreamSurface(lowerer, access.expression);
   lowerer.noLowering(
     `Writable.${name}`,
     call,
@@ -514,3 +539,255 @@ export function lowerChildWriterMethodCall(
     lowerer.checker.getSymbolAtLocation(access.name),
   );
 }
+
+/** A receiver spelled as a plain access path (`child.stdout`, `out`): its
+ * root symbol and member names, or null for other forms. A `const` alias
+ * of an access path (`const out = child.stdout`) resolves to that path,
+ * and so does a method call declared to return `this`
+ * (`child.stdout.setEncoding("utf8")`), so every spelling of one stream
+ * gets the same key. */
+function accessPathOf(
+  lowerer: Lowerer,
+  expr: ts.Expression,
+  depth = 0,
+): { root: ts.Symbol; path: string } | null {
+  if (depth > 16) return null;
+  if (ts.isParenthesizedExpression(expr) || ts.isNonNullExpression(expr)) {
+    return accessPathOf(lowerer, expr.expression, depth + 1);
+  }
+  if (ts.isIdentifier(expr)) {
+    const root = lowerer.checker.getSymbolAtLocation(expr);
+    if (!root) return null;
+    const init = constInitializerOf(root);
+    const aliased = init === null ? null : accessPathOf(lowerer, init, depth + 1);
+    return aliased ?? { root, path: "" };
+  }
+  if (ts.isPropertyAccessExpression(expr) && !expr.questionDotToken) {
+    const base = accessPathOf(lowerer, expr.expression, depth + 1);
+    return base === null ? null : { root: base.root, path: `${base.path}.${expr.name.text}` };
+  }
+  if (returnsReceiver(lowerer, expr)) {
+    return accessPathOf(
+      lowerer,
+      (expr.expression as ts.PropertyAccessExpression).expression,
+      depth + 1,
+    );
+  }
+  return null;
+}
+
+/** A method call (`recv.m(...)`, not optional) whose declaration returns
+ * `this`, so its value is the receiver itself. */
+function returnsReceiver(lowerer: Lowerer, expr: ts.Expression): expr is ts.CallExpression {
+  if (
+    !ts.isCallExpression(expr) ||
+    expr.questionDotToken ||
+    !ts.isPropertyAccessExpression(expr.expression) ||
+    expr.expression.questionDotToken
+  )
+    return false;
+  const declaration = lowerer.checker.getResolvedSignature(expr)?.declaration;
+  return declaration?.type?.kind === ts.SyntaxKind.ThisType;
+}
+
+/** A call of a child_process factory (spawn, fork, exec, execFile), each
+ * of which returns a new ChildProcess. */
+function createsChildProcess(lowerer: Lowerer, call: ts.CallExpression): boolean {
+  const declaration = lowerer.checker.getResolvedSignature(call)?.declaration;
+  if (
+    declaration === undefined ||
+    !ts.isFunctionDeclaration(declaration) ||
+    declaration.name === undefined ||
+    !["spawn", "fork", "exec", "execFile"].includes(declaration.name.text)
+  )
+    return false;
+  for (let node: ts.Node | undefined = declaration.parent; node; node = node.parent) {
+    if (ts.isModuleDeclaration(node) && ts.isStringLiteral(node.name)) {
+      return node.name.text === "child_process" || node.name.text === "node:child_process";
+    }
+  }
+  return false;
+}
+
+/** The statement that always evaluates `call` when it runs: the call is
+ * the statement's expression, a variable's initializer, or the receiver of
+ * a method chain that is one of those. Null for calls under a condition,
+ * an operator, an argument list or any other construct. */
+function unconditionalStatementOf(call: ts.CallExpression): ts.Statement | null {
+  let node: ts.Node = call;
+  for (;;) {
+    const parent: ts.Node | undefined = node.parent;
+    if (parent === undefined) return null;
+    if (
+      ts.isParenthesizedExpression(parent) ||
+      ts.isNonNullExpression(parent) ||
+      (ts.isPropertyAccessExpression(parent) &&
+        parent.expression === node &&
+        !parent.questionDotToken) ||
+      (ts.isCallExpression(parent) && parent.expression === node && !parent.questionDotToken)
+    ) {
+      node = parent;
+      continue;
+    }
+    if (ts.isExpressionStatement(parent)) return parent;
+    if (ts.isVariableDeclaration(parent) && parent.initializer === node) {
+      const list = parent.parent;
+      const statement = list?.parent;
+      return list !== undefined &&
+        ts.isVariableDeclarationList(list) &&
+        statement !== undefined &&
+        ts.isVariableStatement(statement)
+        ? statement
+        : null;
+    }
+    return null;
+  }
+}
+
+/** The initializer of a symbol declared once as a `const` variable, else null. */
+function constInitializerOf(symbol: ts.Symbol): ts.Expression | null {
+  const decls = symbol.declarations;
+  if (decls.length !== 1) return null;
+  const decl = decls[0]!;
+  if (
+    !ts.isVariableDeclaration(decl) ||
+    !ts.isIdentifier(decl.name) ||
+    decl.initializer === undefined ||
+    (ts.getCombinedNodeFlags(decl) & ts.NodeFlags.Const) === 0
+  )
+    return null;
+  return decl.initializer;
+}
+
+/** Whether two resolved access paths name different objects: the same root
+ * with different members, or two `const` roots each initialized by `new`
+ * or a child_process factory (such as two `spawn()` results). Any other
+ * call may return an object another binding already holds. */
+function distinctPaths(
+  lowerer: Lowerer,
+  a: { root: ts.Symbol; path: string },
+  b: { root: ts.Symbol; path: string },
+): boolean {
+  if (a.root === b.root) return a.path !== b.path;
+  const fresh = (root: ts.Symbol): boolean => {
+    let init = constInitializerOf(root);
+    while (init !== null && ts.isParenthesizedExpression(init)) init = init.expression;
+    if (init === null) return false;
+    return (
+      ts.isNewExpression(init) || (ts.isCallExpression(init) && createsChildProcess(lowerer, init))
+    );
+  };
+  return fresh(a.root) && fresh(b.root);
+}
+
+/** Whether an unknown-typed 'data' listener registered by `call` receives
+ * strings: a `setEncoding` statement on the same receiver path precedes
+ * the registration in its own block or an enclosing block of the same
+ * function. With no setEncoding on that path in the function, chunks are
+ * Buffers. Any other placement leaves the encoding state unclear, which
+ * is reported. */
+function childStreamEncodedAt(
+  lowerer: Lowerer,
+  receiver: ts.Expression,
+  call: ts.CallExpression,
+): boolean {
+  const key = accessPathOf(lowerer, receiver);
+  const statement: ts.Node = call.parent!;
+  let scope: ts.Node | undefined = statement;
+  while (scope && !ts.isFunctionLike(scope) && !ts.isSourceFile(scope)) scope = scope.parent;
+  const unclear = (): never =>
+    lowerer.unsupported(
+      "SC1090",
+      call.arguments.nodes[1]!,
+      "untyped child-stream 'data' listeners whose setEncoding state is unclear at their registration sites (declare the chunk as 'Buffer' or 'string')",
+    );
+  if (key === null || scope === undefined) return unclear();
+  const preceding = new Set<ts.Node>();
+  for (let node: ts.Node = statement; node !== scope;) {
+    const parent: ts.Node | undefined = node.parent;
+    if (parent === undefined) break;
+    if (ts.isBlock(parent) || ts.isSourceFile(parent)) {
+      for (const sibling of parent.statements.nodes) {
+        if (sibling === node) break;
+        preceding.add(sibling);
+      }
+    }
+    node = parent;
+  }
+  const receiverType = lowerer.checker.getNonNullableType(lowerer.typeOf(receiver));
+  let before = false;
+  let elsewhere = false;
+  const visit = (node: ts.Node): void => {
+    if (node !== scope && ts.isFunctionLike(node)) return;
+    if (
+      ts.isCallExpression(node) &&
+      ts.isPropertyAccessExpression(node.expression) &&
+      node.expression.name.text === "setEncoding"
+    ) {
+      const target = node.expression.expression;
+      const path = accessPathOf(lowerer, target);
+      if (path !== null && path.root === key.root && path.path === key.path) {
+        const statement = unconditionalStatementOf(node);
+        if (statement !== null && preceding.has(statement)) before = true;
+        else elsewhere = true;
+      } else if (
+        (path === null || !distinctPaths(lowerer, path, key)) &&
+        lowerer.checker.isTypeAssignableTo(receiverType, lowerer.typeOf(target))
+      ) {
+        // Another spelling that can hold this stream (a parameter, a `let`
+        // copy, a call result): its setEncoding may be this stream's.
+        elsewhere = true;
+      }
+    }
+    ts.forEachChild(node, visit);
+  };
+  visit(scope);
+  if (elsewhere) return unclear();
+  return before;
+}
+
+/** An inline 'exit'/'close' listener may declare `(code: number, signal:
+ * string)`: TypeScript accepts it through ChildProcess's catch-all `on`
+ * overload, while Node passes `number | null` and `Signals | null`. Those
+ * parameters bind as checked-dynamic values (the socket 'data' listener
+ * precedent), so the body observes the null Node passes; a typed use of a
+ * null value throws the checked conversion's TypeError. Returns the
+ * narrowed parameter symbols, empty when the declared types already
+ * admit null. */
+function narrowedExitListenerParams(lowerer: Lowerer, listener: ts.Expression): ts.Symbol[] {
+  if (!ts.isArrowFunction(listener) && !ts.isFunctionExpression(listener)) return [];
+  if (isJsSourceFile(listener.getSourceFile())) return [];
+  const symbols: ts.Symbol[] = [];
+  const wanted: readonly string[] = ["f64", "string"];
+  listener.parameters.nodes.slice(0, 2).forEach((param, i) => {
+    if (!ts.isIdentifier(param.name) || param.dotDotDotToken || param.initializer) return;
+    if (param.questionToken) return;
+    const symbol = lowerer.checker.getSymbolAtLocation(param.name);
+    if (!symbol || lowerer.runtimeOptionalBindingTypes.has(symbol)) return;
+    if (lowerer.paramShape(param).type.kind !== wanted[i]) return;
+    symbols.push(symbol);
+  });
+  return symbols;
+}
+
+/** Lowers an inline listener with the given parameters bound as
+ * checked-dynamic values typed `unknown` for the body. */
+function lowerWithCheckedParams(
+  lowerer: Lowerer,
+  listener: ts.Expression,
+  symbols: readonly ts.Symbol[],
+): IrExpr {
+  const previousImplicit = lowerer.implicitParamTypes;
+  const implicit = new Map(previousImplicit ?? []);
+  for (const symbol of symbols) {
+    implicit.set(symbol, lowerer.checker.getUnknownType());
+    lowerer.runtimeOptionalBindingTypes.set(symbol, DYN);
+  }
+  lowerer.implicitParamTypes = implicit;
+  try {
+    return lowerer.lowerExpr(listener);
+  } finally {
+    for (const symbol of symbols) lowerer.runtimeOptionalBindingTypes.delete(symbol);
+    lowerer.implicitParamTypes = previousImplicit;
+  }
+}

packages/compiler/src/frontend/lowering/builtins/process.ts

@@ -3,6 +3,7 @@ import { envSnapshotHelper, lowerProcessLoadEnvFile } from "./environment.js";
 import { dynUndefinedExpr, boolLit, numLit, strLit, varRef } from "../../../ir/build.js";
 import { timerStyleCallback } from "../lower-timers.js";
 import * as ts from "../../ts7/adapter.js";
+import { requestGeneralStreamSurface } from "../general-streams.js";
 import { type Lowerer, own } from "../lowerer.js";
 import { isJsSourceFile, locOf } from "../../program.js";
 import { isSafeToDiscard } from "../expressions/evaluation-safety.js";
@@ -225,6 +226,7 @@ export function lowerProcStreamMethodCall(
     const receiver = lowerer.lowerExpr(access.expression);
     const data = lowerer.lowerExpr(call.arguments.nodes[0]!);
     if (data.type.kind !== "string") {
+      requestGeneralStreamSurface(lowerer, access.expression);
       lowerer.noLowering(
         `write of '${lowerer.fmt(data.type)}' data on a stream value`,
         call.arguments.nodes[0]!,
@@ -233,6 +235,7 @@ export function lowerProcStreamMethodCall(
     }
     return { kind: "libCall", fn: "procStream.write", args: [receiver, data], type: BOOL, loc };
   }
+  requestGeneralStreamSurface(lowerer, access.expression);
   lowerer.noLowering(
     `WritableStream.${name}`,
     call,

packages/compiler/src/frontend/lowering/builtins/properties.ts

@@ -1,5 +1,5 @@
 import { isBuiltinClassInstance } from "./receiver-types.js";
-import { strLit } from "../../../ir/build.js";
+import { numLit, strLit } from "../../../ir/build.js";
 import * as ts from "../../ts7/adapter.js";
 import { type Lowerer } from "../lowerer.js";
 import { locOf } from "../../program.js";
@@ -192,6 +192,25 @@ export function lowerBuiltinExtraProperty(
   // +1-or-NULL runtime pair.
   if (kind === "child" && (name === "stdin" || name === "stdout" || name === "stderr")) {
     const receiver = lowerer.lowerExpr(expr.expression);
+    // Under the general stream representation (general-streams.ts) the
+    // declared stream type maps to a runtime stream class: the read
+    // answers the child's bridged stream (one per stdio slot, so repeated
+    // reads keep identity), fed by the same pipe.
+    const general = lowerer.generalNodeStreams ? generalChildStdioType(lowerer, expr, name) : null;
+    if (general) {
+      const type: IrType = {
+        kind: "union",
+        unionId: lowerer.unions.intern([general, { kind: "nullT" }]),
+      };
+      const read: IrExpr = {
+        kind: "libCall",
+        fn: "child.stdioStream",
+        args: [receiver, numLit(name === "stdin" ? 0 : name === "stdout" ? 1 : 2, loc)],
+        type,
+        loc,
+      };
+      return lowerer.maybeNarrow(read, expr);
+    }
     const streamType = name === "stdin" ? CHILDWRITER_T : CHILDSTREAM_T;
     const type: IrType = {
       kind: "union",
@@ -259,3 +278,21 @@ export function lowerBuiltinExtraProperty(
   }
   return null;
 }
+
+/** The runtime stream class a child stdio read produces under the general
+ * stream representation, or null when the declared type keeps the stdio
+ * handle (the fallback declarations' ChildStdin). */
+function generalChildStdioType(
+  lowerer: Lowerer,
+  expr: ts.PropertyAccessExpression,
+  name: "stdin" | "stdout" | "stderr",
+): IrType | null {
+  const className = name === "stdin" ? "%Writable" : "%Readable";
+  const mapped = lowerer.mapTypeOf(lowerer.typeOf(expr));
+  if (!mapped) return null;
+  const arms =
+    mapped.kind === "union" ? (lowerer.unions.get(mapped.unionId)?.arms ?? []) : [mapped];
+  return arms.some((arm) => arm.kind === "object" && arm.className === className)
+    ? { kind: "object", className }
+    : null;
+}

packages/compiler/src/frontend/lowering/expressions/narrowing.ts

@@ -26,6 +26,7 @@ import { jsBindingHasOpenWrites } from "../lower-stmts.js";
 import { narrowClassUnion, narrowStoredClassValue, structuralNarrowing } from "../class-unions.js";
 import { checkedClassAssertion, checkedClassUnionAssertion } from "../class-assertions.js";
 import { isOptionalProcessStreamProperty } from "../builtins/process.js";
+import { requestGeneralStreamSurface } from "../general-streams.js";
 import { literalValues } from "../../literal-values.js";
 import { literalUnionArm } from "../../union-discriminants.js";
 import { checkedArmExtraction } from "../coercions/unions.js";
@@ -1450,6 +1451,9 @@ export function lowerAsExpression(
     if (DYN_HANDLE_KINDS.has(target.kind)) {
       return { kind: "dynCheck", value: inner, type: target, loc: locOf(expr) };
     }
+    // `value as NodeJS.ReadableStream`: the general stream classes take
+    // the checked cast (general-streams.ts).
+    requestGeneralStreamSurface(lowerer, expr);
     lowerer.unsupported(
       "SC1090",
       expr,

packages/compiler/src/frontend/lowering/general-streams.ts

@@ -0,0 +1,114 @@
+/* The general stream representation (TypeMapperCtx.generalNodeStreams).
+ *
+ * NodeJS.ReadableStream and NodeJS.WritableStream (and @types/node's
+ * stream.Readable / stream.Writable, which also type child stdio) name the
+ * interface every Node stream implements. By default they map to the
+ * child/process stdio handles: the specialized representation of the
+ * values that commonly fill them (child.stdout, process.stdout). A program
+ * that moves another stream — fs.createReadStream(), a zlib Transform, a
+ * Readable subclass — into one of those slots needs the general
+ * representation instead: the runtime stream classes, with the stdio
+ * handles converted into them where they are read or flow.
+ *
+ * The decision is per program. Reachable emit lowers without it; a flow
+ * that needs it marks the lowerer (needsGeneralNodeStreams) and lowerToIr
+ * lowers the whole program again with it. Programs without such a flow
+ * never take the second pass, so their IR is unchanged. */
+import type { IrExpr, IrType } from "../../ir/ir.js";
+import { everyTypeChild } from "../../ir/traverse.js";
+import * as ts from "../ts7/adapter.js";
+import { isGeneralNodeStreamInterface } from "../type-mapper.js";
+import type { Lowerer } from "./lowerer.js";
+import { streamSidesOf } from "./lower-stream.js";
+
+const STDIO_STREAM_KINDS: ReadonlySet<IrType["kind"]> = new Set([
+  "childStream",
+  "childWriter",
+  "procStream",
+]);
+
+/** Whether `type` (through union arms, record fields and the structural
+ * type children) contains a part satisfying `test`. */
+function containsPart(lowerer: Lowerer, type: IrType, test: (part: IrType) => boolean): boolean {
+  const seen = new Set<string>();
+  const walk = (part: IrType): boolean => {
+    if (test(part)) return true;
+    if (part.kind === "union") {
+      if (seen.has(`u:${part.unionId}`)) return false;
+      seen.add(`u:${part.unionId}`);
+      return (lowerer.unions.get(part.unionId)?.arms ?? []).some(walk);
+    }
+    if (part.kind === "record") {
+      if (seen.has(`r:${part.shapeId}`)) return false;
+      seen.add(`r:${part.shapeId}`);
+      return (lowerer.shapes.get(part.shapeId)?.fields ?? []).some((f) => walk(f.type));
+    }
+    let found = false;
+    everyTypeChild(part, (child) => {
+      found = walk(child);
+      return !found;
+    });
+    return found;
+  };
+  return walk(type);
+}
+
+function isGeneralStream(lowerer: Lowerer, part: IrType): boolean {
+  return (
+    part.kind === "object" && streamSidesOf(lowerer, lowerer.classes.get(part.className)) !== null
+  );
+}
+
+/** A mismatched flow (requireExactShape's input) that the general stream
+ * representation resolves: a general stream into a stdio-represented
+ * slot, a child pipe handle into an 'unknown' slot, or an 'unknown' value
+ * into a child-pipe slot (the general classes take a checked cast). */
+export function needsGeneralNodeStreams(
+  lowerer: Lowerer,
+  actual: IrType,
+  expected: IrType,
+): boolean {
+  const stdio = (part: IrType): boolean => STDIO_STREAM_KINDS.has(part.kind);
+  const pipeHandle = (part: IrType): boolean =>
+    part.kind === "childStream" || part.kind === "childWriter";
+  if (actual.kind === "dyn") return containsPart(lowerer, expected, pipeHandle);
+  if (containsPart(lowerer, expected, stdio))
+    return containsPart(lowerer, actual, (p) => isGeneralStream(lowerer, p));
+  return expected.kind === "dyn" && containsPart(lowerer, actual, pipeHandle);
+}
+
+/** Under the general representation, process.stdout / process.stderr (the
+ * procStream fd scalar) flowing into a runtime Writable slot: an
+ * fd-backed Writable whose writes take the same synchronous path as
+ * process.stdout.write. Null when the flow is not that one. */
+export function stdioIntoGeneralStream(
+  lowerer: Lowerer,
+  expr: IrExpr,
+  expected: IrType,
+): IrExpr | null {
+  if (!lowerer.generalNodeStreams || expr.type.kind !== "procStream") return null;
+  const writable: IrType = { kind: "object", className: "%Writable" };
+  const accepts =
+    (expected.kind === "object" && expected.className === "%Writable") ||
+    (expected.kind === "union" && lowerer.armTag(expected.unionId, writable) >= 0);
+  if (!accepts) return null;
+  return {
+    kind: "libCall",
+    fn: "stream.stdioWritable",
+    args: [expr],
+    type: writable,
+    loc: expr.loc,
+  };
+}
+
+/** A stdio-handle receiver whose member has no lowering on the handle
+ * (`out.on("error", ...)`, `child.stdout.pipe(...)`, `output.end()`): when
+ * its declared type is one the general representation maps to a runtime
+ * stream class, the program lowers again with it, where the full stream
+ * surface applies. The caller still reports its fence for this pass. */
+export function requestGeneralStreamSurface(lowerer: Lowerer, receiver: ts.Expression): void {
+  if (lowerer.generalNodeStreams) return;
+  const type = lowerer.checker.getNonNullableType(lowerer.typeOf(receiver));
+  if (isGeneralNodeStreamInterface(type, lowerer.typeCtx))
+    lowerer.requiresGeneralNodeStreams = true;
+}

packages/compiler/src/frontend/lowering/lower-event-emitter.ts

@@ -1245,7 +1245,19 @@ export function lowerEmitterMethodCall(
         );
       }
       const p = cbT.params[0];
-      if (p !== undefined && !(p.kind === "bytes" && p.elem === "u8") && p.kind !== "string") {
+      // A `Buffer | string` chunk (NodeJS.ReadableStream's listener
+      // shape) takes whichever side the stream delivers.
+      const chunkUnion =
+        p?.kind === "union" &&
+        (lowerer.unions.get(p.unionId)?.arms ?? []).every(
+          (arm) => (arm.kind === "bytes" && arm.elem === "u8") || arm.kind === "string",
+        );
+      if (
+        p !== undefined &&
+        !(p.kind === "bytes" && p.elem === "u8") &&
+        p.kind !== "string" &&
+        !chunkUnion
+      ) {
         lowerer.noLowering(
           `a 'data' listener whose chunk parameter is '${lowerer.fmt(p)}'`,
           args[1]!,

packages/compiler/src/frontend/lowering/lower-modules.ts

@@ -872,6 +872,14 @@ export function moduleArtifacts(
             valueInstanceOfRoots.add(root.def.name);
           }
           break;
+        case "libCall":
+          // The general stream representation's child stdio read bridges
+          // through the runtime install that boxes both stream classes.
+          if (node.fn === "child.stdioStream") {
+            visitClass("%Readable");
+            visitClass("%Writable");
+          }
+          break;
         case "dynFrom":
           if (boxedChild(node.value.type)) {
             // A generic child can expose native stdio without importing stream.

packages/compiler/src/frontend/lowering/lower-stdlib.ts

@@ -4,6 +4,7 @@ import { InternalCompilerError } from "../../errors.js";
 import * as ts from "../ts7/adapter.js";
 import { lowerToNumberArgument } from "./lower-exprs.js";
 import type { Lowerer } from "./lowerer.js";
+import { isDeclaredInAmbientNamespace } from "../type-mapper.js";
 import {
   arrayOf,
   BOOL,
@@ -1023,7 +1024,15 @@ export function fenceStaticAbortControllerMemberRead(
 function isStdlibFetchInterface(lowerer: Lowerer, node: ts.Node, owner: string): ts.Symbol | null {
   const type = lowerer.checker.getBaseTypeOfLiteralType(lowerer.typeOf(node));
   for (const symbol of [type.getSymbol(), type.getAliasSymbol()]) {
-    if (symbol?.name === owner && lowerer.isStdlibSymbol(symbol)) return symbol;
+    // NodeJS.ReadableStream shares the WHATWG interface's name.
+    if (
+      symbol?.name === owner &&
+      lowerer.isStdlibSymbol(symbol) &&
+      !symbol.declarations.some(
+        (d) => ts.isInterfaceDeclaration(d) && isDeclaredInAmbientNamespace(d, "NodeJS"),
+      )
+    )
+      return symbol;
   }
   return null;
 }

packages/compiler/src/frontend/lowering/lower-stmts.ts

@@ -1,5 +1,6 @@
 import { declaredMemberOf } from "./expressions/property-access.js";
 import { hostsAll } from "./expressions/hosted-narrowing.js";
+import { requestGeneralStreamSurface } from "./general-streams.js";
 import {
   matchAllRowsReadByElements,
   storedMatchAllRowsReadByElements,
@@ -8971,6 +8972,8 @@ export function lowerForOf(lowerer: Lowerer, stmt: ts.ForOfStatement): IrStmt {
     }
     const builtin = lowerForAwaitBuiltin(lowerer, stmt);
     if (builtin) return builtin;
+    // A NodeJS.ReadableStream iterates as the general Readable.
+    requestGeneralStreamSurface(lowerer, stmt.expression);
     lowerer.unsupported(
       "SC1070",
       stmt,

packages/compiler/src/frontend/lowering/lower-stream.ts

@@ -49,6 +49,7 @@ import { sideEffectFreeOptionValue } from "./surfaces.js";
 import { lowerDynObjectLiteral } from "./expressions/object-literals.js";
 import { dynUndefinedExpr, varRef } from "../../ir/build.js";
 import { genResultRecord } from "../type-mapper.js";
+import { requestGeneralStreamSurface } from "./general-streams.js";
 import { extractIteratorValue, genStep, genValue, stepDrivable } from "./lower-generators.js";
 import {
   BOOL,
@@ -586,6 +587,33 @@ function dynOptionsValue(lowerer: Lowerer, arg: ts.Expression): IrExpr | null {
   return v.type.kind === "dyn" ? v : null;
 }
 
+/** @types/node declares the transform/flush completion callback as
+ * `(error?: Error | null, data?: any) => void`. Outside objectMode Node
+ * accepts only a Buffer, a string or nothing as `data`, so the callback
+ * parameter is typed with that union while the option callback lowers: a
+ * checked-dynamic value passed as `data` converts at the call. Returns the
+ * pinned parameter symbol (the caller unpins it), or null when the
+ * declared type needs no pin. */
+function pinAnyDataCompletion(
+  lowerer: Lowerer,
+  which: string,
+  fnNode: ts.MethodDeclaration | ts.FunctionExpression | ts.ArrowFunction,
+): ts.Symbol | null {
+  const index = which === "transform" ? 2 : which === "flush" ? 0 : -1;
+  const param = index < 0 ? undefined : fnNode.parameters.nodes[index];
+  if (param === undefined || !ts.isIdentifier(param.name) || param.dotDotDotToken) return null;
+  const symbol = lowerer.checker.getSymbolAtLocation(param.name);
+  if (!symbol || lowerer.stageSourceTypes.has(symbol)) return null;
+  const declared = lowerer.paramShape(param).type;
+  if (declared.kind !== "func" || declared.rest || declared.params[1]?.kind !== "dyn") return null;
+  const data = unionOf(lowerer, [BYTES, STRING, { kind: "undefinedT" }]);
+  lowerer.stageSourceTypes.set(
+    symbol,
+    funcOf([declared.params[0]!, data, ...declared.params.slice(2)], declared.ret),
+  );
+  return symbol;
+}
+
 /** Parses the options object literal of a stream constructor. `which`
  * names the callbacks this class accepts, in canonical flag order. */
 function parseStreamOptions(
@@ -671,22 +699,27 @@ function parseStreamOptions(
         continue;
       }
       const full = tuples[name]!.map((t) => (t === "fn" ? null : t));
-      // The completion-callback tail: typed by the checker contextually;
-      // accept the user's declared func type as the tuple's entry so the
-      // prefix check compares like with like.
-      const { funcType } = lowerer.lambdaSignature(fnNode);
-      const tuple: IrType[] = full.map((t, i) => {
-        if (t !== null) return t;
-        const declared = funcType.params[i];
-        if (declared !== undefined && declared.kind !== "func" && declared.kind !== "dyn") {
-          lowerer.noLowering(
-            `a '${name}' callback whose parameter ${i} is '${lowerer.fmt(declared)}' where Node passes the completion callback`,
-            fnNode!,
-          );
-        }
-        return declared ?? funcOf([], VOID);
-      });
-      present.set(name, lowerStreamCallback(lowerer, name, fnNode, thisType, tuple));
+      const pinned = pinAnyDataCompletion(lowerer, name, fnNode);
+      try {
+        // The completion-callback tail: typed by the checker contextually;
+        // accept the user's declared func type as the tuple's entry so the
+        // prefix check compares like with like.
+        const { funcType } = lowerer.lambdaSignature(fnNode);
+        const tuple: IrType[] = full.map((t, i) => {
+          if (t !== null) return t;
+          const declared = funcType.params[i];
+          if (declared !== undefined && declared.kind !== "func" && declared.kind !== "dyn") {
+            lowerer.noLowering(
+              `a '${name}' callback whose parameter ${i} is '${lowerer.fmt(declared)}' where Node passes the completion callback`,
+              fnNode!,
+            );
+          }
+          return declared ?? funcOf([], VOID);
+        });
+        present.set(name, lowerStreamCallback(lowerer, name, fnNode, thisType, tuple));
+      } finally {
+        if (pinned) lowerer.stageSourceTypes.delete(pinned);
+      }
       continue;
     }
     // `{ signal }` shorthand: the binding is the value.
@@ -2044,6 +2077,24 @@ function lowerStreamArg(
   const v = isHashValueType(lowered.type)
     ? hashStreamOf(lowered, what === "pipeline" || what === "compose")
     : lowered;
+  // A checked-dynamic pipeline stage (an untyped source or destination):
+  // a boxed native stream unwraps at run time; other values (iterables,
+  // functions) throw the not-supported Error there.
+  if (v.type.kind === "dyn" && what === "pipeline") {
+    return {
+      kind: "libCall",
+      fn: "stream.unboxDyn",
+      args: [
+        v,
+        strLit(
+          "pipeline stages other than native streams (iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet",
+          locOf(node),
+        ),
+      ],
+      type: { kind: "object", className: "%Duplex" },
+      loc: locOf(node),
+    };
+  }
   const info = v.type.kind === "object" ? lowerer.classes.get(v.type.className) : undefined;
   if (!streamSidesOf(lowerer, info)) {
     // A provably-non-stream value in a JS source (the invalid-input
@@ -3296,6 +3347,124 @@ function pipeEndOption(lowerer: Lowerer, options: ts.Expression | undefined, loc
   return end;
 }
 
+/** `readable.pipe(process.stdout)`: the process stream joins the pipe as
+ * an fd-backed Writable (stream.stdioWritable), and — like Node, which
+ * never ends process.stdout or process.stderr from a pipe — the source's
+ * end does not end it. pipe() answers the process stream itself. */
+function pipeIntoStdio(
+  lowerer: Lowerer,
+  call: ts.CallExpression,
+  access: ts.PropertyAccessExpression,
+  target: IrExpr,
+): IrExpr {
+  const loc = locOf(call);
+  // JS order: the receiver, then the destination, then the options.
+  const receiverValue = lowerer.lowerExpr(access.expression);
+  const source = lowerer.declareHiddenLocal("%pipeSource", receiverValue.type);
+  const stdio = lowerer.declareHiddenLocal("%pipeStdio", target.type);
+  const options = call.arguments.nodes[1];
+  const stmts: IrStmt[] = [
+    { kind: "varDecl", localId: source.id, init: receiverValue, loc },
+    { kind: "varDecl", localId: stdio.id, init: target, loc },
+  ];
+  // The options literal still evaluates; its end flag has no effect here.
+  if (options !== undefined) {
+    stmts.push({ kind: "exprStmt", expr: pipeEndOption(lowerer, options, loc), loc });
+  }
+  const receiver: IrExpr = { kind: "varRef", localId: source.id, type: receiverValue.type, loc };
+  const writable: IrType = { kind: "object", className: "%Writable" };
+  stmts.push({
+    kind: "exprStmt",
+    expr: {
+      kind: "libCall",
+      fn: "readable.pipe",
+      args: [
+        receiver,
+        {
+          kind: "libCall",
+          fn: "stream.stdioWritable",
+          args: [{ kind: "varRef", localId: stdio.id, type: target.type, loc }],
+          type: writable,
+          loc,
+        },
+        boolLit(false, loc),
+      ],
+      type: writable,
+      loc,
+    },
+    loc,
+  });
+  return {
+    kind: "seqExpr",
+    stmts,
+    result: { kind: "varRef", localId: stdio.id, type: target.type, loc },
+    type: target.type,
+    loc,
+  };
+}
+
+/** The pipe() libCall for a destination of type `dst`. Under the general
+ * stream representation a Writable slot can hold the converted
+ * process.stdout / process.stderr, which a pipe never ends (as in Node), so
+ * the pipe takes readable.pipeGeneral, which checks for them. */
+function generalPipe(lowerer: Lowerer, dst: IrType): "readable.pipe" | "readable.pipeGeneral" {
+  return lowerer.generalNodeStreams && dst.kind === "object" && dst.className === "%Writable"
+    ? "readable.pipeGeneral"
+    : "readable.pipe";
+}
+
+/** `readable.pipe(untyped)`: a checked-dynamic destination unwraps a boxed
+ * native stream at run time (stream.unboxDyn; other values throw the
+ * not-supported Error), and pipe() answers the destination value itself. */
+function pipeIntoDyn(
+  lowerer: Lowerer,
+  call: ts.CallExpression,
+  access: ts.PropertyAccessExpression,
+  target: IrExpr,
+): IrExpr {
+  const loc = locOf(call);
+  const receiverValue = lowerer.lowerExpr(access.expression);
+  const source = lowerer.declareHiddenLocal("%pipeSource", receiverValue.type);
+  const destination = lowerer.declareHiddenLocal("%pipeDestination", target.type);
+  const stmts: IrStmt[] = [
+    { kind: "varDecl", localId: source.id, init: receiverValue, loc },
+    { kind: "varDecl", localId: destination.id, init: target, loc },
+  ];
+  const end = pipeEndOption(lowerer, call.arguments.nodes[1], loc);
+  const duplex: IrType = { kind: "object", className: "%Duplex" };
+  stmts.push({
+    kind: "exprStmt",
+    expr: {
+      kind: "libCall",
+      fn: lowerer.generalNodeStreams ? "readable.pipeGeneral" : "readable.pipe",
+      args: [
+        { kind: "varRef", localId: source.id, type: receiverValue.type, loc },
+        {
+          kind: "libCall",
+          fn: "stream.unboxDyn",
+          args: [
+            { kind: "varRef", localId: destination.id, type: target.type, loc },
+            strLit("pipe destinations other than native streams are not supported yet", loc),
+          ],
+          type: duplex,
+          loc,
+        },
+        end,
+      ],
+      type: duplex,
+      loc,
+    },
+    loc,
+  });
+  return {
+    kind: "seqExpr",
+    stmts,
+    result: { kind: "varRef", localId: destination.id, type: target.type, loc },
+    type: target.type,
+    loc,
+  };
+}
+
 export function lowerStreamMethodCall(
   lowerer: Lowerer,
   call: ts.CallExpression,
@@ -3545,22 +3714,26 @@ export function lowerStreamMethodCall(
         loc,
       };
     }
-    const dst = lowerer.lowerExpr(args[0]!);
+    const target = lowerer.lowerExpr(args[0]!);
+    if (target.type.kind === "procStream") return pipeIntoStdio(lowerer, call, access, target);
+    if (target.type.kind === "dyn") return pipeIntoDyn(lowerer, call, access, target);
+    const dst = target;
     const dstInfo =
       dst.type.kind === "object" ? lowerer.classes.get(dst.type.className) : undefined;
     const dstSides = streamSidesOf(lowerer, dstInfo);
     if (dstSides !== "w" && dstSides !== "rw") {
+      requestGeneralStreamSurface(lowerer, args[0]!);
       lowerer.noLowering(
         `pipe into a '${lowerer.fmt(dst.type)}'`,
         args[0]!,
-        "the destination must be a stream Writable/Duplex (process streams and sockets are not pipe destinations yet)",
+        "the destination must be a stream Writable/Duplex, process.stdout or process.stderr (sockets are not pipe destinations yet)",
       );
     }
     const end = pipeEndOption(lowerer, args[1], loc);
     const receiver = lowerer.lowerExpr(access.expression);
     return {
       kind: "libCall",
-      fn: "readable.pipe",
+      fn: generalPipe(lowerer, dst.type),
       args: [receiver, dst, end],
       type: dst.type,
       loc,

packages/compiler/src/frontend/lowering/lowerer.ts

@@ -1,3 +1,4 @@
+import { needsGeneralNodeStreams, stdioIntoGeneralStream } from "./general-streams.js";
 import { extendAssertedRecordMembers } from "./assertion-members.js";
 import { jsDynamicDeclarations } from "./js-dynamic-objects.js";
 import {
@@ -883,6 +884,10 @@ export interface LowererMode {
   /** Statically resolved child_process.fork program roots, in stable target-id order. */
   forkTargets?: readonly ts.SourceFile[];
   frontendServices?: FrontendServices | undefined;
+  /** The program's general stream representation (TypeMapperCtx.
+   * generalNodeStreams). lowerToIr turns it on only after a pass without
+   * it saw a stream flow that needs it (requiresGeneralNodeStreams). */
+  generalNodeStreams?: boolean;
   /** LoadResult.inlineRequireCycleCalls. */
   inlineRequireCycleCalls?: ReadonlySet<ts.Node> | undefined;
   /** LowerOptions.libraryCallbacks (see there). */
@@ -1039,10 +1044,40 @@ export function lowerToIr(
   for (const d of extraModuleCycleDiags) reachableEmit.pushDiag(d);
   for (const d of ffiValidation.diagnostics) reachableEmit.pushDiag(d);
   const emitted = reachableEmit.emitReachable(options.libRoots);
-  const { reachable } = emitted;
+  let { reachable } = emitted;
   let result = emitted.result;
   let resultLowerer = reachableEmit;
   timing("reachable-emit", { reachable: reachable.size });
+  // The general stream representation (TypeMapperCtx.generalNodeStreams)
+  // is a per-program decision: only a program whose reachable code moved
+  // a general stream into a stdio-represented stream slot lowers again
+  // with it. Every other program keeps the single pass above unchanged.
+  const generalNodeStreams = reachableEmit.requiresGeneralNodeStreams;
+  if (generalNodeStreams) {
+    const general = new Lowerer(program, entry, moduleOrder, {
+      frontendServices: options.frontendServices,
+      projectCheckJs: options.projectCheckJs ?? false,
+      targetPlatform,
+      startupCrash,
+      inlineRequireCycleCalls: options.inlineRequireCycleCalls,
+      ffiImports,
+      libraryCallbacks,
+      forkTargets,
+      workerTargets,
+      externalTypes,
+      externalTypeSpecifiersByFile,
+      tierMode,
+      ffiBindingSymbols: ffiValidation.symbolsByName,
+      generalNodeStreams: true,
+    });
+    for (const d of extraModuleCycleDiags) general.pushDiag(d);
+    for (const d of ffiValidation.diagnostics) general.pushDiag(d);
+    const generalEmitted = general.emitReachable(options.libRoots);
+    reachable = generalEmitted.reachable;
+    result = generalEmitted.result;
+    resultLowerer = general;
+    timing("general-streams-relower", { reachable: reachable.size });
+  }
   // A generic class-rest support decision can depend on record metadata
   // whose historical owner is discovered only while retained bodies lower.
   // If the settled answer is a fence, rerun the ordinary reachable emit so
@@ -1050,10 +1085,11 @@ export function lowerToIr(
   // declarators stay unvisited, bindings block, cascades and stats match the
   // historical compiler. This is a rare compatibility fallback; programs
   // without such a settled fence retain checker-backed IR exactly once.
-  if (reachableEmit.requiresHistoricalOrderRelower) {
+  if (resultLowerer.requiresHistoricalOrderRelower) {
     const emit = new Lowerer(program, entry, moduleOrder, {
       frontendServices: options.frontendServices,
       projectCheckJs: options.projectCheckJs ?? false,
+      ...(generalNodeStreams ? { generalNodeStreams: true } : {}),
       reachable,
       targetPlatform,
       startupCrash,
@@ -1144,24 +1180,34 @@ export function lowerToIr(
     timing("wasm-discovery", { modules: wasmModules.length });
   }
   if (options.coverage !== true) return result;
-  const remainder = new Lowerer(program, entry, moduleOrder, {
-    frontendServices: options.frontendServices,
-    projectCheckJs: options.projectCheckJs ?? false,
-    reachable,
-    remainder: true,
-    alreadyFlushed: resultLowerer.flushedSymbols,
-    inlineRequireCycleCalls: options.inlineRequireCycleCalls,
-    targetPlatform,
-    ffiImports,
-    libraryCallbacks,
-    forkTargets,
-    workerTargets,
-    ffiBindingSymbols: ffiValidation.symbolsByName,
-    externalTypes,
-    externalTypeSpecifiersByFile,
-    tierMode,
-  });
-  const rem = remainder.run();
+  const remainderPass = (general: boolean): Lowerer =>
+    new Lowerer(program, entry, moduleOrder, {
+      frontendServices: options.frontendServices,
+      projectCheckJs: options.projectCheckJs ?? false,
+      ...(general ? { generalNodeStreams: true } : {}),
+      reachable,
+      remainder: true,
+      alreadyFlushed: resultLowerer.flushedSymbols,
+      inlineRequireCycleCalls: options.inlineRequireCycleCalls,
+      targetPlatform,
+      ffiImports,
+      libraryCallbacks,
+      forkTargets,
+      workerTargets,
+      ffiBindingSymbols: ffiValidation.symbolsByName,
+      externalTypes,
+      externalTypeSpecifiersByFile,
+      tierMode,
+    });
+  let remainder = remainderPass(generalNodeStreams);
+  let rem = remainder.run();
+  // Unreached code reports what it would compile to once reached: a
+  // stream flow there that needs the general representation lowers the
+  // remainder again with it, as reachable emit would.
+  if (!generalNodeStreams && remainder.requiresGeneralNodeStreams) {
+    remainder = remainderPass(true);
+    rem = remainder.run();
+  }
   timing("coverage-remainder");
   // Tiers (tier-analysis.ts): reached statements in dynamic-tier functions,
   // and with tierReport the report-only demotion analysis.
@@ -2048,7 +2094,9 @@ export class Lowerer {
   /** A pipeline() function stage's `source` parameter: the IR type of what
    * the pipeline passes it (the previous stage as an async iterable), set
    * while the stage lowers. Parameter typing and `for await` heads read
-   * it before the checker's `AsyncIterable<T>`. */
+   * it before the checker's `AsyncIterable<T>`. A stream option callback's
+   * completion parameter declared with an any-typed `data` is pinned here
+   * too while the callback lowers (pinAnyDataCompletion). */
   readonly stageSourceTypes = new Map<ts.Symbol, IrType>();
   /** Set when a match result lowers to execGroups/matchGroups, the inputs
    * of regex span folding. */
@@ -2526,6 +2574,12 @@ export class Lowerer {
   /** Settled generic class-rest metadata requires the historical emit
    * fallback so its fence can poison the original statement atomically. */
   requiresHistoricalOrderRelower = false;
+  /** A general stream value (a runtime stream class instance) reached a
+   * slot the child/process stdio handles represent — NodeJS.ReadableStream,
+   * NodeJS.WritableStream, or @types/node's stream.Readable/Writable — or
+   * a stdio handle reached an 'unknown' slot. lowerToIr then lowers the
+   * program again under the general stream representation. */
+  requiresGeneralNodeStreams = false;
   /** Helpers that snapshot shape declaration order into their bodies.
    * Reachability lowers inits before the declarations they discover, so
    * these rebuild after the worklist restores historical shape metadata. */
@@ -2984,6 +3038,7 @@ export class Lowerer {
    * refuse and fence exactly as before. Stage S6 demotes the units of
    * committed refusals in "auto" and keeps them refusals in "static-only". */
   readonly tierMode: TierMode;
+  readonly generalNodeStreams: boolean;
   /** Functions finishModule placed in the dynamic tier (ir/tiers.ts). */
   dynamicFunctions: DynamicFunction[] = [];
   /** Their IR, for coverage's dynamic statement count. */
@@ -3112,6 +3167,7 @@ export class Lowerer {
     this.startupCrash = mode.startupCrash ?? null;
     this.targetPlatform = mode.targetPlatform ?? process.platform;
     this.tierMode = mode.tierMode ?? "auto";
+    this.generalNodeStreams = mode.generalNodeStreams ?? false;
     this.workerTargets = mode.workerTargets ?? [];
     this.workerTargets.forEach((sf, id) => {
       this.workerTargetIdByPath.set(tsgoPath(resolve(sf.fileName)), id);
@@ -3208,6 +3264,7 @@ export class Lowerer {
       isExternalTypeFile: (sf) =>
         this.externalTypeSpecifiersByFile.has(tsgoPath(resolve(sf.fileName))),
       typeMemo: this.typeMemo,
+      ...(this.generalNodeStreams ? { generalNodeStreams: true } : {}),
       canMemoizeType: () =>
         this.typeParamBindings === null &&
         this.typeParamTsBindings === null &&
@@ -8492,6 +8549,8 @@ export class Lowerer {
    * validator ICEs on them as the usual backstop. */
   requireExactShape(node: ts.Node, actual: IrType, expected: IrType): void {
     if (typeEquals(actual, expected)) return;
+    if (!this.generalNodeStreams && needsGeneralNodeStreams(this, actual, expected))
+      this.requiresGeneralNodeStreams = true;
     // dyn mismatches first. A dyn ('unknown') value flowing into a typed
     // slot needs a CHECKED cast — the hint points at `as <type>`; tsc
     // usually rejects this before we do, so the fence is mostly defensive.
@@ -8709,6 +8768,9 @@ export class Lowerer {
     }
     if (expr.type.kind === "void" && expected.kind !== "void" && this.neverValued.has(expr))
       return this.divergentValue(expr, expected);
+    // process.stdout/stderr into a general Writable slot (general-streams.ts).
+    const stdioStream = stdioIntoGeneralStream(this, expr, expected);
+    if (stdioStream) return this.coerceToExpected(stdioStream, expected);
     // Iterator-typed slots: native iterators and differently typed
     // generators step through an adapter generator.
     if (expected.kind === "generator" && !typeEquals(expr.type, expected)) {

packages/compiler/src/frontend/type-mapper.ts

@@ -980,7 +980,7 @@ function isDeclaredInAmbientModule(d: ts.Declaration, name: string): boolean {
 /** True when the declaration's nearest enclosing namespace is `name`
  * (`declare global { namespace NodeJS { ... } }` — the NodeJS-global
  * interfaces @types/node declares outside any ambient module). */
-function isDeclaredInAmbientNamespace(d: ts.Declaration, name: string): boolean {
+export function isDeclaredInAmbientNamespace(d: ts.Declaration, name: string): boolean {
   let node: ts.Node | undefined = d.parent;
   while (node) {
     if (ts.isModuleDeclaration(node) && ts.isIdentifier(node.name)) {
@@ -991,6 +991,31 @@ function isDeclaredInAmbientNamespace(d: ts.Declaration, name: string): boolean
   return false;
 }
 
+/** Whether `type` is one of the stream spellings the general stream
+ * representation (TypeMapperCtx.generalNodeStreams) maps differently:
+ * NodeJS.ReadableStream / NodeJS.WritableStream, @types/node's
+ * stream.Readable / stream.Writable, or the fallback declarations'
+ * ChildStdin. */
+export function isGeneralNodeStreamInterface(
+  type: ts.Type,
+  ctx: Pick<TypeMapperCtx, "isStdlibFile">,
+): boolean {
+  const psym = type.getSymbol() ?? type.getAliasSymbol();
+  if (psym === undefined) return false;
+  return psym.declarations.some((d) => {
+    if (!(ts.isInterfaceDeclaration(d) || ts.isClassDeclaration(d))) return false;
+    if (!ctx.isStdlibFile(d.getSourceFile())) return false;
+    if (psym.name === "ReadableStream" || psym.name === "WritableStream")
+      return ts.isInterfaceDeclaration(d) && isDeclaredInAmbientNamespace(d, "NodeJS");
+    if (psym.name === "ChildStdin") return isDeclaredInAmbientModule(d, "child_process");
+    return (
+      (psym.name === "Readable" || psym.name === "Writable") &&
+      isNodeTypesPath(d.getSourceFile().fileName) &&
+      isDeclaredInAmbientModule(d, "stream")
+    );
+  });
+}
+
 /** A type declared in the shipped WebAssembly namespace (scriptc.d.ts):
  * its modules, instances, memories, tables, globals, error classes and the
  * Exports/Imports records. */
@@ -1146,6 +1171,13 @@ export interface TypeMapperCtx {
    * type descriptions for project-owned values, but their imported runtime
    * bindings are fenced separately by the Lowerer. */
   isExternalTypeFile: (sf: ts.SourceFile) => boolean;
+  /** The program's general stream representation (lowerToIr decides it
+   * per program): NodeJS.ReadableStream / NodeJS.WritableStream and
+   * @types/node's stream.Readable / stream.Writable map to the runtime
+   * stream classes instead of the child/process stdio handles, so values
+   * from every stream producer share their slots. Off (the default), the
+   * stdio handles keep those spellings. */
+  generalNodeStreams?: boolean;
   /** Successful context-free mappings, keyed by the
    * TypeScript server's stable type id. A generated facade can reference the
    * same large record/function type thousands of times; remapping its full
@@ -3179,6 +3211,20 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
   // global-namespace interface real code writes in its own signatures —
   // prefixStream's `output` param). The representation is the raw fd
   // scalar; the lowered surface is write(string).
+  // Under the program's general stream representation (generalNodeStreams)
+  // NodeJS.WritableStream is the runtime Writable instead: the process
+  // streams convert into it where they flow (process.stdioWritable).
+  const nodeWritableInterface =
+    psym?.name === "WritableStream" &&
+    psym.declarations.some(
+      (d) =>
+        ts.isInterfaceDeclaration(d) &&
+        ctx.isStdlibFile(d.getSourceFile()) &&
+        isDeclaredInAmbientNamespace(d, "NodeJS"),
+    );
+  if (nodeWritableInterface && ctx.generalNodeStreams === true) {
+    return { kind: "object", className: "%Writable" };
+  }
   if (
     (psym?.name === "WriteStream" &&
       psym.declarations.some(
@@ -3188,13 +3234,7 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
           (isDeclaredInAmbientModule(d, "tty") ||
             (isDeclaredInAmbientModule(d, "process") && isDeclaredInAmbientNamespace(d, "NodeJS"))),
       )) ||
-    (psym?.name === "WritableStream" &&
-      psym.declarations.some(
-        (d) =>
-          ts.isInterfaceDeclaration(d) &&
-          ctx.isStdlibFile(d.getSourceFile()) &&
-          isDeclaredInAmbientNamespace(d, "NodeJS"),
-      ))
+    nodeWritableInterface
   ) {
     return PROCSTREAM_T;
   }
@@ -3204,6 +3244,7 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
   // of the broad @types class fence where they are constructed.
   if (
     (psym?.name === "ChildStdin" &&
+      ctx.generalNodeStreams !== true &&
       psym.declarations.some(
         (d) =>
           ts.isInterfaceDeclaration(d) &&
@@ -3211,6 +3252,7 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
           isDeclaredInAmbientModule(d, "child_process"),
       )) ||
     (psym?.name === "Writable" &&
+      ctx.generalNodeStreams !== true &&
       psym.declarations.some(
         (d) =>
           (ts.isInterfaceDeclaration(d) || ts.isClassDeclaration(d)) &&
@@ -3221,6 +3263,20 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
   ) {
     return { kind: "childWriter" };
   }
+  // Under the general stream representation the fallback declarations'
+  // ChildStdin is the runtime Writable too (child.stdioStream bridges it).
+  if (
+    ctx.generalNodeStreams === true &&
+    psym?.name === "ChildStdin" &&
+    psym.declarations.some(
+      (d) =>
+        ts.isInterfaceDeclaration(d) &&
+        ctx.isStdlibFile(d.getSourceFile()) &&
+        isDeclaredInAmbientModule(d, "child_process"),
+    )
+  ) {
+    return { kind: "object", className: "%Writable" };
+  }
   // The static node:stream classes (the shipped fallback declarations —
   // NOT @types/node, whose stream.Readable also types child stdio; under
   // @types/node the childStream mapping below keeps priority and the
@@ -3228,6 +3284,9 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
   // branch: the fallback's Readable must map to the runtime class, and
   // under the fallback child.stdout is NodeJS.ReadableStream (its own
   // branch below), so the two never collide.
+  // @types/node's Duplex/Transform/PassThrough classes have no child-stdio
+  // reading, so they map to the runtime classes too; its Readable and
+  // Writable join them under the general representation.
   for (const [irName, rec] of RUNTIME_STREAM_CLASSES) {
     if (
       psym?.name === rec.lib &&
@@ -3235,38 +3294,39 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
         (d) =>
           ts.isClassDeclaration(d) &&
           ctx.isStdlibFile(d.getSourceFile()) &&
-          !isNodeTypesPath(d.getSourceFile().fileName) &&
+          (!isNodeTypesPath(d.getSourceFile().fileName) ||
+            ctx.generalNodeStreams === true ||
+            (rec.lib !== "Readable" && rec.lib !== "Writable")) &&
           isDeclaredInAmbientModule(d, "stream"),
       )
     ) {
       return { kind: "object", className: irName };
     }
   }
-  // node:zlib's stream classes (the shipped fallback declarations): their
-  // instances are runtime Transform streams.
+  // node:zlib's stream classes (the shipped fallback declarations and
+  // @types/node's): their instances are runtime Transform streams.
   if (
     psym !== undefined &&
     ZLIB_STREAM_MODES.has(psym.name) &&
     psym.declarations.some(
       (d) =>
-        ts.isClassDeclaration(d) &&
+        (ts.isClassDeclaration(d) ||
+          (ts.isInterfaceDeclaration(d) && isNodeTypesPath(d.getSourceFile().fileName))) &&
         ctx.isStdlibFile(d.getSourceFile()) &&
-        !isNodeTypesPath(d.getSourceFile().fileName) &&
         isDeclaredInAmbientModule(d, "zlib"),
     )
   ) {
     return { kind: "object", className: "%Transform" };
   }
-  // fs.ReadStream / fs.WriteStream (the shipped fallback declarations):
-  // their instances are runtime Readable / Writable streams
-  // (scr_fs_stream.c).
+  // fs.ReadStream / fs.WriteStream (the shipped fallback declarations and
+  // @types/node's): their instances are runtime Readable / Writable
+  // streams (scr_fs_stream.c).
   if (
     (psym?.name === "ReadStream" || psym?.name === "WriteStream") &&
     psym.declarations.some(
       (d) =>
         ts.isClassDeclaration(d) &&
         ctx.isStdlibFile(d.getSourceFile()) &&
-        !isNodeTypesPath(d.getSourceFile().fileName) &&
         isDeclaredInAmbientModule(d, "fs"),
     )
   ) {
@@ -3288,6 +3348,9 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
           isDeclaredInAmbientNamespace(d, "NodeJS"),
       ))
   ) {
+    // The general representation: the runtime Readable, which child
+    // stdio reads produce through the child stream bridge.
+    if (ctx.generalNodeStreams === true) return { kind: "object", className: "%Readable" };
     return { kind: "childStream" };
   }
   // dgram.Socket: the same ambiguous "Socket" name disambiguated by its
@@ -4334,10 +4397,18 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
       const mapped = mapType(checker.getTypeOfSymbol(p), ctx);
       if (!mapped) return false;
       const arms = mapped.kind === "union" ? ctx.unions.get(mapped.unionId)?.arms : [mapped];
+      // Under the general stream representation the stdio members are the
+      // runtime stream classes.
+      const isExpected = (arm: IrType): boolean =>
+        arm.kind === expected ||
+        (ctx.generalNodeStreams === true &&
+          arm.kind === "object" &&
+          ((expected === "childStream" && arm.className === "%Readable") ||
+            (expected === "childWriter" && arm.className === "%Writable")));
       return (
         arms !== undefined &&
-        arms.some((arm) => arm.kind === expected) &&
-        arms.every((arm) => arm.kind === expected || arm.kind === unit)
+        arms.some(isExpected) &&
+        arms.every((arm) => isExpected(arm) || arm.kind === unit)
       );
     };
     const childShaped =

packages/compiler/src/ir/builtin-effects.ts

@@ -75,6 +75,8 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "process.stdoutWriteBytes",
   "process.stderrWriteBytes",
   "procStream.write",
+  "child.stdioStream",
+  "stream.unboxDyn",
   "weakMap.new",
   "weakSet.new",
   "dyn.fromEntries",
@@ -338,6 +340,7 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "stream.onDyn",
   "readable.resume",
   "readable.pipe",
+  "readable.pipeGeneral",
   "readable.unpipe",
   "writable.write",
   "writable.writeStr",

packages/compiler/src/ir/builtin-signatures.ts

@@ -2615,8 +2615,13 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
    * queue, end/destroy settle it, writable is a pure state read, and
    * listener callbacks move into the writer registry. */
   "child.stdin": { argTypes: [CHILD_T], result: VOID },
+  /** Under the general stream representation (general-streams.ts), a
+   * stdio read (slot 0, 1 or 2) answers the child's bridged runtime stream
+   * — `Writable | null` for stdin, `Readable | null` otherwise; the bridge
+   * throws when the program lacks the stream unit's boxing helpers. */
+  "child.stdioStream": { argTypes: [CHILD_T, F64], result: VOID },
   // Listener registrations: callback func shapes are program-dependent
-  // (zero-param / Buffer / Buffer-armed union) — the slots pin arity,
+  // (zero-param / Buffer / string / unknown / Buffer-armed union) — the slots pin arity,
   // the stream receiver, and the once-flag.
   "stream.onData": { argTypes: [CHILDSTREAM_T, null, BOOL], result: VOID },
   "stream.onDataStr": { argTypes: [CHILDSTREAM_T, null, BOOL], result: VOID },
@@ -2636,6 +2641,21 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
    * stderrWrite paths so buffering and ordering stay identical. Data
    * borrowed; the bool is Node's always-true backpressure signal. */
   "procStream.write": { argTypes: [PROCSTREAM_T, STRING], result: BOOL },
+  /** process.stdout/stderr flowing into a runtime Writable slot under the
+   * general stream representation: an fd-backed Writable (+1) whose
+   * writes take process.stdout.write's synchronous path. */
+  /** A pipeline() stage that is a checked-dynamic value: a boxed native
+   * stream unwraps by identity (+1); anything else throws the given
+   * not-supported message. Typed as a Duplex so every stage position
+   * accepts it; the runtime pipeline checks the stream's real sides. */
+  "stream.unboxDyn": {
+    argTypes: [DYN, STRING],
+    result: { kind: "object", className: "%Duplex" },
+  },
+  "stream.stdioWritable": {
+    argTypes: [PROCSTREAM_T],
+    result: { kind: "object", className: "%Writable" },
+  },
   "child.killed": { argTypes: [CHILD_T], result: BOOL },
   "child.kill": { argTypes: [CHILD_T, STRING], result: BOOL },
   "child.killNum": { argTypes: [CHILD_T, F64], result: BOOL },
@@ -3673,7 +3693,9 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
    * returns the below-hwm answer. readable.read: (recv, size — -1 for
    * absent) → Buffer|null union. pause/resume return recv +1 (`this`);
    * isPaused answers the flag. readable.pipe: (recv, dst, end) → dst +1,
-   * fires 'pipe' on dst; readable.unpipe (recv[, dst]) → recv +1.
+   * fires 'pipe' on dst; readable.pipeGeneral: the same, with `end` off
+   * when dst is the converted process.stdout / process.stderr;
+   * readable.unpipe (recv[, dst]) → recv +1.
    * writable.write/writeStr: (recv, chunk[, cb]) → below-hwm bool (cb
    * MOVES when present, called after the user write completes).
    * writable.end: (recv, flags[, chunk][, cb]) → recv +1. cork/uncork are
@@ -3871,6 +3893,7 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
   "readable.resume": { argTypes: [null], result: VOID },
   "readable.isPaused": { argTypes: [null], result: BOOL },
   "readable.pipe": { argTypes: [null, null, BOOL], result: VOID },
+  "readable.pipeGeneral": { argTypes: [null, null, BOOL], result: VOID },
   "readable.unpipe": { argTypes: [null], result: VOID },
   "readable.flowing": { argTypes: [null], result: VOID },
   "writable.write": { argTypes: [null, BYTES_U8], result: BOOL },

packages/compiler/src/ir/runtime-features.ts

@@ -149,6 +149,7 @@ const ABORT_SIGNAL_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
 
 /** Entry points that link the stream unit (with STREAM_LIB_PREFIXES). */
 const STREAM_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
+  "child.stdioStream",
   "console.bind",
   "crypto.hashStream",
   "crypto.hashStreamPiped",
@@ -167,6 +168,8 @@ const STREAM_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "stream.destroyErr",
   "stream.errored",
   "stream.prop",
+  "stream.stdioWritable",
+  "stream.unboxDyn",
   "stream.workerStdio",
 ]);
 

packages/compiler/src/ir/validate.ts

@@ -5228,6 +5228,20 @@ function validateFunction(
       }
       return true;
     }
+    if (e.fn === "child.stdioStream") {
+      const def = e.type.kind === "union" ? unions.get(e.type.unionId) : undefined;
+      const slot = e.args[1];
+      const className =
+        slot?.kind === "numLit" ? (slot.value === 0 ? "%Writable" : "%Readable") : null;
+      const ok =
+        def &&
+        className !== null &&
+        def.arms.length === 2 &&
+        def.arms.some((a) => a.kind === "object" && a.className === className) &&
+        def.arms.some((a) => a.kind === "nullT");
+      if (!ok) err(`libCall ${e.fn} must return the stream class of its stdio slot or null`, e.loc);
+      return true;
+    }
     if (e.fn === "child.stdin" || e.fn === "child.stdout" || e.fn === "child.stderr") {
       const def = e.type.kind === "union" ? unions.get(e.type.unionId) : undefined;
       const handleKind = e.fn === "child.stdin" ? "childWriter" : "childStream";
@@ -5719,9 +5733,9 @@ function validateFunction(
         }
         return true;
       }
-      if (e.fn === "readable.pipe") {
+      if (e.fn === "readable.pipe" || e.fn === "readable.pipeGeneral") {
         if (!isStreamObject(e.args[1]?.type) || !typeEquals(e.type, e.args[1]!.type)) {
-          err(`libCall readable.pipe must return its destination's type`, e.loc);
+          err(`libCall ${e.fn} must return its destination's type`, e.loc);
         }
         return true;
       }

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6463-node-readable-stream-producers.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6463-node-readable-stream-producers.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6464-node-writable-stream-producers.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6464-node-writable-stream-producers.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6465-node-stream-js-boundary/main.ts.json

@@ -0,0 +1,7 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6465-node-stream-js-boundary/consume.js",
+  "<repo>/tests/corpus/6465-node-stream-js-boundary/main.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6466-node-stdio-stream-surface.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6466-node-stdio-stream-surface.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6467-node-stream-untyped-sources/main.ts.json

@@ -0,0 +1,7 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6467-node-stream-untyped-sources/sources.js",
+  "<repo>/tests/corpus/6467-node-stream-untyped-sources/main.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6468-node-stdio-writable-slot.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6468-node-stdio-writable-slot.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6469-node-stdio-writable-js-pipe/main.ts.json

@@ -0,0 +1,7 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6469-node-stdio-writable-js-pipe/pipe.js",
+  "<repo>/tests/corpus/6469-node-stdio-writable-js-pipe/main.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6470-stream-transform-any-data-callback.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6470-stream-transform-any-data-callback.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6471-stream-data-buffer-or-string.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6471-stream-data-buffer-or-string.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6472-child-data-unknown-chunk.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6472-child-data-unknown-chunk.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/fixtures/node-types/child-stdin-fences.ts.json

@@ -1,6 +0,0 @@
-{
- "order": [
-  "<repo>/tests/fixtures/node-types/child-stdin-fences.ts"
- ],
- "diags": []
-}

packages/compiler/test/ts7/baselines/order-parity/tests/fixtures/node-types/child-stdin-streams.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/fixtures/node-types/child-stdin-streams.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/fixtures/node-types/node-stream-interfaces.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/fixtures/node-types/node-stream-interfaces.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/fixtures/node-types/stream-construction.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/fixtures/node-types/stream-construction.ts"
+ ],
+ "diags": []
+}

packages/runtime/src/scr_stream.c

@@ -6451,8 +6451,10 @@ static ScrDyn *scr_stream_view_call(ScrClosure *cb, ScrDyn *const *args, size_t
         end = option->kind == SCR_DYN_UNDEF || scr_dyn_truthy(option);
         scr_dyn_release(option);
       }
+      /* scr_stream_pipe_general: a JavaScript caller can pipe into the
+       * converted process.stdout / process.stderr, which a pipe never ends. */
       ScrStream *returned = !strcmp(method, "pipe")
-          ? scr_stream_pipe(stream, destination, end) : scr_stream_unpipe(stream, destination);
+          ? scr_stream_pipe_general(stream, destination, end) : scr_stream_unpipe(stream, destination);
       scr_stream_release(returned);
       if (!strcmp(method, "pipe") && !scr_exc_pending()) result = scr_dyn_retain(first);
     }
@@ -6619,3 +6621,104 @@ ScrStream *scr_crypto_hash_stream_piped(ScrCryptoHash *hash) {
   if (s) scr_crypto_hash_stream_hold(hash);
   return s;
 }
+
+/* ── the compiler's general stream representation ─────────────────────
+ * NodeJS.ReadableStream / NodeJS.WritableStream slots that also hold other
+ * streams compile as this unit's classes; the stdio handles convert into
+ * them where they are read or flow. Both entry points sit last in the unit
+ * and reference only global symbols and no new constant data, so programs
+ * that never convert link the same bytes as before. */
+
+/* destroy() on a bridged child stdin closes the pipe, as Node's does. */
+static void scr_stream_child_stdin_destroy(ScrClosure *cb, ScrStream *s, ScrError *err) {
+  ScrChildWriter *writer = scr_box_get_ref(cb->caps[0]);
+  scr_child_writer_destroy(writer);
+  scr_child_writer_release(writer);
+  scr_stream_destroy_done(s, err ? scr_error_retain(err) : NULL);
+}
+
+/* A child stdio read (the child.stdioStream libCall): the bridged stream
+ * the checked-dynamic ChildProcess view answers for the same key ("stdin",
+ * "stdout" or "stderr", passed by the compiled program), so both views
+ * share one stream per slot. The stdin stream (the only five-byte key)
+ * gains the destroy hook that closes the pipe. +1, or NULL for an unpiped
+ * slot or a throw. */
+ScrStream *scr_stream_child_stdio(void *child, const char *key, size_t key_len) {
+  ScrDyn *handle = scr_dyn_new_handle(child, SCR_DYNH_CHILD);
+  ScrDyn *box = scr_dyn_handle_ops_of(handle)->get(child, key, key_len);
+  scr_dyn_release(handle);
+  if (box == NULL) return NULL;
+  ScrStream *stream = box->kind == SCR_DYN_TYPED_REF ? scr_dyn_typed_ref_unbox(box) : NULL;
+  scr_dyn_release(box);
+  if (stream != NULL && key_len == 5 && stream->st->destroy_cb == NULL) {
+    ScrChildWriter *writer = scr_child_stdin(child);
+    if (writer != NULL) {
+      ScrClosure *destroy = scr_closure_new((void *)scr_stream_child_stdin_destroy, 1);
+      destroy->caps[0] = scr_box_new_obj(scr_child_writer_retain_v, scr_child_writer_release_v, NULL);
+      scr_box_set_ref(destroy->caps[0], writer);
+      stream->st->destroy_cb = destroy;
+      stream->st->destroy_inv = scr_stream_child_stdin_destroy;
+    }
+  }
+  return stream;
+}
+
+/* process.stdout / process.stderr flowing into a Writable slot (the
+ * stream.stdioWritable libCall): a Writable whose _write hands each chunk
+ * to the descriptor through process.stdout.write's synchronous path and
+ * completes at once, so ordering against console output and direct writes
+ * is unchanged. The closure keeps the fd as given (1 or 2). Each thread
+ * keeps one per descriptor, so every conversion of process.stdout answers
+ * the same stream, as Node's process.stdout is one object: listeners,
+ * writable state and identity are shared. The cache is released at exit. */
+static SCR_TL ScrStream *scr_stream_stdio_writables[2];
+static void scr_stream_stdio_write(ScrClosure *cb, ScrStream *s, ScrBytes *chunk) {
+  int fd = (int)scr_box_get_f64(cb->caps[0]);
+  (void)scr_stdio_write(fd == 2 ? 2 : 1, chunk->data, chunk->len * scr_bytes_elem_size(chunk->elem));
+  scr_stream_write_done(s, NULL);
+}
+static void scr_stream_stdio_writables_free(void) {
+  for (int i = 0; i < 2; i++) {
+    ScrStream *s = scr_stream_stdio_writables[i];
+    scr_stream_stdio_writables[i] = NULL;
+    if (s) scr_stream_release(s);
+  }
+}
+ScrStream *scr_stream_stdio_writable(double fd) {
+  int i = fd == 2 ? 1 : 0;
+  if (scr_stream_stdio_writables[i] == NULL) {
+    if (scr_stream_stdio_writables[1 - i] == NULL) scr_atexit(scr_stream_stdio_writables_free);
+    ScrClosure *write = scr_closure_new((void *)scr_stream_stdio_write, 1);
+    write->caps[0] = scr_box_new(SCR_BOX_F64);
+    scr_box_set_f64(write->caps[0], i ? 2 : 1);
+    scr_stream_stdio_writables[i] = scr_stream_new_writable(-1, true, true, write,
+        scr_stream_stdio_write, NULL, NULL, NULL, NULL);
+  }
+  return scr_stream_retain(scr_stream_stdio_writables[i]);
+}
+
+/* readable.pipe() into a Writable slot under the general representation
+ * (the readable.pipeGeneral libCall), and pipe() called from JavaScript on
+ * a native stream: Node never ends process.stdout or process.stderr from a
+ * pipe, so their converted Writable keeps `end` off. */
+ScrStream *scr_stream_pipe_general(ScrStream *src, ScrStream *dst, bool end) {
+  if (dst == scr_stream_stdio_writables[0] || dst == scr_stream_stdio_writables[1]) end = false;
+  return scr_stream_pipe(src, dst, end);
+}
+
+/* A pipeline() stage that arrives as a checked-dynamic value (the
+ * stream.unboxDyn libCall): a boxed native stream unwraps by identity
+ * (+1); any other value throws the compiled program's not-supported
+ * message (Node also takes iterables and functions there). */
+ScrStream *scr_stream_unbox_dyn(const ScrDyn *value, const ScrStr *message) {
+  if (value->kind == SCR_DYN_TYPED_REF && value->v.typed_ref.type_key_len > 7 &&
+      !memcmp(value->v.typed_ref.type_key, "object:", 7)) {
+    ScrStream *stream = scr_dyn_typed_ref_unbox(value);
+    const ScrVt *vt = stream->vt;
+    if ((vt->pre >= scr_readable_vt.pre && vt->pre <= scr_readable_vt.post) ||
+        (vt->pre >= scr_writable_vt.pre && vt->pre <= scr_writable_vt.post)) return stream;
+    value->v.typed_ref.release(stream);
+  }
+  scr_throw_error_msg_code(SCR_ERR_ERROR, message->data, message->len, "SC2020");
+  return NULL;
+}

packages/runtime/src/scr_stream.h

@@ -125,6 +125,17 @@ void scr_stream_init_passthrough(ScrStream *s, double rhwm, double whwm, bool au
 ScrStream *scr_stream_new_readable_dyn(struct ScrDyn *opts);
 void scr_stream_worker_stdio(void);
 ScrStream *scr_stream_new_writable_dyn(struct ScrDyn *opts);
+/* The general stream representation's conversions (+1): a child's bridged
+ * stdio stream by key (NULL when unpiped), and process.stdout/stderr as a
+ * Writable (fd 1 or 2). */
+ScrStream *scr_stream_child_stdio(void *child, const char *key, size_t key_len);
+ScrStream *scr_stream_stdio_writable(double fd);
+/* readable.pipe() under the general representation: a converted
+ * process.stdout / process.stderr destination is never ended (dst +1). */
+ScrStream *scr_stream_pipe_general(ScrStream *src, ScrStream *dst, bool end);
+/* A boxed native stream out of a checked-dynamic pipeline stage (+1), or
+ * NULL with `message` thrown as the not-supported Error. */
+ScrStream *scr_stream_unbox_dyn(const struct ScrDyn *value, const ScrStr *message);
 ScrStream *scr_stream_new_duplex_dyn(struct ScrDyn *opts);
 ScrStream *scr_stream_new_transform_dyn(struct ScrDyn *opts);
 ScrStream *scr_stream_new_passthrough_dyn(struct ScrDyn *opts);

tests/corpus/6463-node-readable-stream-producers.ts

@@ -0,0 +1,75 @@
+// NodeJS.ReadableStream is the interface every Node readable implements.
+// One helper typed with it consumes a file stream, a Readable subclass,
+// Readable.from, a gunzip Transform fed through pipe(), and a child's
+// stdout. A program that mixes those producers compiles the interface as
+// the node:stream Readable, with the child's pipe bridged into one.
+import { spawn } from "node:child_process";
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+import { Readable } from "node:stream";
+import * as zlib from "node:zlib";
+
+function drain(label: string, src: NodeJS.ReadableStream): Promise<string> {
+  return new Promise((resolve, reject) => {
+    let text = "";
+    let chunks = 0;
+    src.on("data", (chunk: Buffer | string) => {
+      chunks++;
+      text += typeof chunk === "string" ? `[s]${chunk}` : chunk.toString("utf8");
+    });
+    src.on("error", (err) => reject(err));
+    src.on("end", () => {
+      console.log(label, "end", chunks > 0, src.readable);
+      resolve(text);
+    });
+  });
+}
+
+class Countdown extends Readable {
+  private left = 3;
+  _read(): void {
+    this.push(this.left > 0 ? `t-${this.left--};` : null);
+  }
+}
+
+function openText(file: string): NodeJS.ReadableStream {
+  return fs.createReadStream(file);
+}
+
+function openDecoded(file: string): NodeJS.ReadableStream {
+  const stream = fs.createReadStream(file);
+  stream.setEncoding("utf8");
+  return stream;
+}
+
+async function main(): Promise<void> {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), "scriptc-node-readable-"));
+  const plain = path.join(dir, "plain.txt");
+  const packed = path.join(dir, "packed.gz");
+  fs.writeFileSync(plain, "hello from a file\n");
+  fs.writeFileSync(packed, zlib.gzipSync(Buffer.from("inflated text\n")));
+
+  console.log(JSON.stringify(await drain("file", openText(plain))));
+  console.log(JSON.stringify(await drain("decoded", openDecoded(plain))));
+  console.log(JSON.stringify(await drain("subclass", new Countdown())));
+  console.log(JSON.stringify(await drain("from", Readable.from(["a", "b", "c"]))));
+  const gunzip = fs.createReadStream(packed).pipe(zlib.createGunzip());
+  console.log(JSON.stringify(await drain("gunzip", gunzip)));
+
+  const child = spawn("sh", ["-c", "printf 'child line\\n'"], {
+    stdio: ["ignore", "pipe", "inherit"],
+  });
+  const out = child.stdout;
+  if (out !== null) console.log(JSON.stringify(await drain("child", out)));
+  console.log("same stdout", child.stdout === out);
+
+  try {
+    await drain("missing", openText(path.join(dir, "missing.txt")));
+  } catch (err) {
+    console.log("missing", (err as NodeJS.ErrnoException).code);
+  }
+  fs.rmSync(dir, { recursive: true });
+}
+
+void main();

tests/corpus/6464-node-writable-stream-producers.ts

@@ -0,0 +1,68 @@
+// NodeJS.WritableStream is the interface every Node writable implements:
+// process.stdout/stderr, fs.WriteStream and the node:stream Writable
+// family. Helpers typed with it accept all of them, a copy helper pipes a
+// file stream into one, and readable.pipe(process.stdout) leaves the
+// process stream open afterwards, as in Node.
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+import { Readable, Writable } from "node:stream";
+
+function emit(dst: NodeJS.WritableStream, line: string): boolean {
+  return dst.write(`${line}\n`);
+}
+
+function copy(src: NodeJS.ReadableStream, dst: NodeJS.WritableStream): Promise<void> {
+  return new Promise((resolve, reject) => {
+    src.on("error", reject);
+    dst.on("error", reject);
+    dst.on("finish", () => resolve());
+    src.pipe(dst);
+  });
+}
+
+class Collector extends Writable {
+  readonly parts: string[] = [];
+  _write(chunk: Buffer, _encoding: string, callback: (error?: Error | null) => void): void {
+    this.parts.push(chunk.toString("utf8"));
+    callback();
+  }
+}
+
+function finished(stream: NodeJS.WritableStream): Promise<void> {
+  return new Promise((resolve) => {
+    stream.end(() => resolve());
+  });
+}
+
+async function main(): Promise<void> {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), "scriptc-node-writable-"));
+  const target = path.join(dir, "out.txt");
+  const file = fs.createWriteStream(target);
+  console.log("file write", emit(file, "to a file"));
+  console.log("stdout write", emit(process.stdout, "to stdout"));
+  emit(process.stderr, "to stderr");
+  const collector = new Collector();
+  emit(collector, "to a collector");
+  await finished(file);
+  await finished(collector);
+  console.log(JSON.stringify(fs.readFileSync(target, "utf8")));
+  console.log(JSON.stringify(collector.parts));
+
+  const source = path.join(dir, "source.txt");
+  const copied = path.join(dir, "copied.txt");
+  fs.writeFileSync(source, "copied through pipe\n");
+  await copy(fs.createReadStream(source), fs.createWriteStream(copied));
+  console.log(JSON.stringify(fs.readFileSync(copied, "utf8")));
+
+  await new Promise<void>((resolve) => {
+    const piped = Readable.from(["piped ", "to ", "stdout\n"]);
+    piped.on("end", () => resolve());
+    console.log("pipe answers stdout", piped.pipe(process.stdout) === process.stdout);
+  });
+  // The pipe did not end process.stdout.
+  emit(process.stdout, "still writable");
+  fs.rmSync(dir, { recursive: true });
+}
+
+void main();

tests/corpus/6465-node-stream-js-boundary/consume.js

@@ -0,0 +1,12 @@
+export function streamToText(stream) {
+  return new Promise((resolve, reject) => {
+    const chunks = [];
+    stream.on("data", (chunk) => chunks.push(Buffer.from(chunk)));
+    stream.on("error", reject);
+    stream.on("end", () => resolve(Buffer.concat(chunks).toString("utf8")));
+  });
+}
+
+export function describeStream(stream) {
+  console.log(typeof stream.on, typeof stream.pipe, stream.readable);
+}

tests/corpus/6465-node-stream-js-boundary/main.ts

@@ -0,0 +1,33 @@
+// NodeJS.ReadableStream values crossing into untyped JavaScript and into
+// 'unknown': a file stream and a child's stdout share one helper with a
+// JavaScript stream consumer, which sees ordinary Node streams.
+import { spawn } from "node:child_process";
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+import { describeStream, streamToText } from "./consume.js";
+
+function open(file: string): NodeJS.ReadableStream {
+  return fs.createReadStream(file);
+}
+
+function textOf(stream: NodeJS.ReadableStream): Promise<string> {
+  return streamToText(stream);
+}
+
+async function main(): Promise<void> {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), "scriptc-node-stream-js-"));
+  const file = path.join(dir, "input.txt");
+  fs.writeFileSync(file, "through javascript\n");
+  const held: unknown = open(file);
+  console.log(typeof held);
+  describeStream(held);
+  console.log(JSON.stringify(await textOf(open(file))));
+  const child = spawn("sh", ["-c", "printf 'child output\\n'"], {
+    stdio: ["ignore", "pipe", "inherit"],
+  });
+  if (child.stdout !== null) console.log(JSON.stringify(await textOf(child.stdout)));
+  fs.rmSync(dir, { recursive: true });
+}
+
+void main();

tests/corpus/6466-node-stdio-stream-surface.ts

@@ -0,0 +1,45 @@
+// The stdio handles reach the full stream surface when a program uses it:
+// child.stdout piped to process.stdout, a file stream piped into a
+// child's stdin, and an 'error' listener on a NodeJS.WritableStream
+// parameter that only receives the process streams. Each of these
+// compiles the NodeJS stream interfaces as the node:stream classes.
+import { spawn } from "node:child_process";
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+
+function report(out: NodeJS.WritableStream, text: string): void {
+  out.on("error", (err) => console.log("error", err.message));
+  out.write(`${text}\n`);
+}
+
+function echo(text: string): Promise<number | null> {
+  return new Promise((resolve) => {
+    const child = spawn("sh", ["-c", `echo ${text}`], { stdio: ["ignore", "pipe", "inherit"] });
+    child.stdout?.pipe(process.stdout);
+    child.on("close", (code) => resolve(code));
+  });
+}
+
+function cat(input: string): Promise<number | null> {
+  return new Promise((resolve) => {
+    const child = spawn("cat", [], { stdio: ["pipe", "pipe", "inherit"] });
+    child.stdout?.pipe(process.stdout);
+    child.on("close", (code) => resolve(code));
+    if (child.stdin) fs.createReadStream(input).pipe(child.stdin);
+  });
+}
+
+async function main(): Promise<void> {
+  report(process.stdout, "reported");
+  report(process.stderr, "reported to stderr");
+  console.log("echo exit", await echo("piped from a child"));
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), "scriptc-node-stdio-"));
+  const input = path.join(dir, "input.txt");
+  fs.writeFileSync(input, "file contents through cat\n");
+  console.log("cat exit", await cat(input));
+  report(process.stdout, "stdout stays open");
+  fs.rmSync(dir, { recursive: true });
+}
+
+void main();

tests/corpus/6467-node-stream-untyped-sources/main.ts

@@ -0,0 +1,58 @@
+// Streams that arrive untyped meet NodeJS stream interface slots: an
+// untyped JavaScript function's stream returned as NodeJS.ReadableStream,
+// a checked cast from 'unknown', pipe() and pipeline() into untyped
+// destinations, and 'for await' over a NodeJS.ReadableStream
+// parameter fed by a file stream and a child's stdout.
+import { spawn } from "node:child_process";
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+import { pipeline } from "node:stream/promises";
+import * as zlib from "node:zlib";
+import { collector, joined, openUntyped } from "./sources.js";
+
+function open(file: string): NodeJS.ReadableStream {
+  return openUntyped(file);
+}
+
+function held(file: string): unknown {
+  return fs.createReadStream(file);
+}
+
+async function collect(label: string, src: NodeJS.ReadableStream): Promise<void> {
+  let text = "";
+  for await (const chunk of src) text += chunk.toString();
+  console.log(label, JSON.stringify(text));
+}
+
+async function main(): Promise<void> {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), "scriptc-node-stream-untyped-"));
+  const plain = path.join(dir, "plain.txt");
+  const packed = path.join(dir, "packed.gz");
+  fs.writeFileSync(plain, "plain text\n");
+  fs.writeFileSync(packed, zlib.gzipSync(Buffer.from("packed text\n")));
+
+  await collect("returned", open(plain));
+  await collect("cast", held(plain) as NodeJS.ReadableStream);
+  const child = spawn("sh", ["-c", "printf 'child text\\n'"], { stdio: ["ignore", "pipe", "inherit"] });
+  if (child.stdout) await collect("child", child.stdout);
+
+  await new Promise<void>((resolve) => {
+    const piped = fs.createReadStream(plain);
+    piped.on("end", () => resolve());
+    piped.pipe(collector());
+  });
+  console.log("pipe", JSON.stringify(joined()));
+
+  const sink = collector();
+  await pipeline(openUntyped(packed), zlib.createGunzip(), sink);
+  console.log("pipeline", JSON.stringify(joined()));
+  try {
+    await pipeline(openUntyped(path.join(dir, "missing.gz")), zlib.createGunzip(), collector());
+  } catch (err) {
+    console.log("pipeline error", (err as NodeJS.ErrnoException).code);
+  }
+  fs.rmSync(dir, { recursive: true });
+}
+
+void main();

tests/corpus/6467-node-stream-untyped-sources/sources.js

@@ -0,0 +1,20 @@
+import * as fs from "node:fs";
+import { PassThrough } from "node:stream";
+
+const parts = [];
+
+/** @returns {any} */
+export function openUntyped(file) {
+  return fs.createReadStream(file);
+}
+
+/** @returns {any} */
+export function collector() {
+  const sink = new PassThrough();
+  sink.on("data", (chunk) => parts.push(chunk.toString("utf8")));
+  return sink;
+}
+
+export function joined() {
+  return parts.splice(0).join("");
+}

tests/corpus/6468-node-stdio-writable-slot.ts

@@ -0,0 +1,49 @@
+// process.stdout and process.stderr in NodeJS.WritableStream slots of a
+// program that also passes other streams there: every slot holds the same
+// stream object, a pipe through such a slot leaves the process stream
+// open, and Writable's listener methods return the stream for chaining.
+import { Readable, Writable } from "node:stream";
+
+class Sink extends Writable {
+  _write(_chunk: Buffer, _encoding: string, callback: (error?: Error | null) => void): void {
+    callback();
+  }
+}
+
+function same(a: NodeJS.WritableStream, b: NodeJS.WritableStream): boolean {
+  return a === b;
+}
+
+function pipeInto(dst: NodeJS.WritableStream, parts: string[]): Promise<void> {
+  return new Promise((resolve) => {
+    const src = Readable.from(parts);
+    src.on("end", () => resolve());
+    src.pipe(dst);
+  });
+}
+
+function write(dst: NodeJS.WritableStream, text: string): boolean {
+  return dst.write(text);
+}
+
+async function main(): Promise<void> {
+  console.log("stdout same", same(process.stdout, process.stdout));
+  console.log("stderr same", same(process.stderr, process.stderr));
+  console.log("stdout and stderr", same(process.stdout, process.stderr));
+  console.log("sink and stdout", same(new Sink(), process.stdout));
+
+  await pipeInto(process.stdout, ["piped ", "through ", "a slot\n"]);
+  // The pipe did not end process.stdout: writes through a slot still work.
+  console.log("write after pipe", write(process.stdout, "still open\n"));
+  await pipeInto(process.stderr, ["piped to stderr\n"]);
+  write(process.stderr, "stderr still open\n");
+
+  // Writable's listener methods return the stream.
+  const sink = new Sink();
+  await new Promise<void>((resolve) => {
+    sink.on("finish", () => resolve()).end("done");
+  });
+  console.log("sink finished", sink.writableFinished);
+}
+
+void main();

tests/corpus/6469-node-stdio-writable-js-pipe/main.ts

@@ -0,0 +1,36 @@
+// process.stdout and process.stderr reach an untyped JavaScript helper
+// through NodeJS.WritableStream slots, and the helper pipes into them. As
+// in Node, the pipe leaves the process stream open, so later writes
+// through the same slots still work.
+import { Readable, Writable } from "node:stream";
+import { pipeAll } from "./pipe.js";
+
+class Sink extends Writable {
+  readonly parts: string[] = [];
+  _write(chunk: Buffer, _encoding: string, callback: (error?: Error | null) => void): void {
+    this.parts.push(chunk.toString("utf8"));
+    callback();
+  }
+}
+
+function into(destination: NodeJS.WritableStream, parts: string[]): Promise<void> {
+  return new Promise((resolve) => {
+    pipeAll(Readable.from(parts), destination, () => resolve());
+  });
+}
+
+function write(destination: NodeJS.WritableStream, text: string): boolean {
+  return destination.write(text);
+}
+
+async function main(): Promise<void> {
+  const sink = new Sink();
+  await into(sink, ["to ", "a sink"]);
+  console.log("sink", JSON.stringify(sink.parts), sink.writableEnded);
+  await into(process.stdout, ["piped ", "by javascript\n"]);
+  console.log("write after pipe", write(process.stdout, "still open\n"));
+  await into(process.stderr, ["piped to stderr\n"]);
+  write(process.stderr, "stderr still open\n");
+}
+
+void main();

tests/corpus/6469-node-stdio-writable-js-pipe/pipe.js

@@ -0,0 +1,4 @@
+export function pipeAll(source, destination, done) {
+  source.on("end", () => done());
+  source.pipe(destination);
+}

tests/corpus/6470-stream-transform-any-data-callback.ts

@@ -0,0 +1,60 @@
+// Transform option callbacks whose completion callback declares an
+// any-typed `data` (the @types/node TransformCallback spelling): the data
+// passed back may be a Buffer, a string, or omitted.
+import { PassThrough, Transform } from "node:stream";
+import { createGunzip, gzipSync } from "node:zlib";
+
+type Done = (error?: Error | null, data?: any) => void;
+
+interface Progress {
+  bytes: number;
+  chunks: number;
+}
+
+function counting(progress: Progress): Transform {
+  return new Transform({
+    transform(chunk: Buffer, _encoding: string, callback: Done) {
+      progress.bytes += chunk.length;
+      progress.chunks++;
+      callback(null, chunk);
+    },
+  });
+}
+
+const upper = new Transform({
+  transform(chunk: Buffer, _encoding: string, callback: Done) {
+    const text = chunk.toString();
+    if (text === "skip") {
+      callback();
+      return;
+    }
+    callback(null, text.toUpperCase());
+  },
+  flush(callback: Done) {
+    callback(null, Buffer.from("!"));
+  },
+});
+upper.on("data", (d: Buffer) => console.log("upper", JSON.stringify(d.toString())));
+upper.on("end", () => console.log("upper end"));
+upper.write("hello");
+upper.write("skip");
+upper.end("world");
+
+const failing = new Transform({
+  transform(chunk: Buffer, _encoding: string, callback: Done) {
+    callback(new Error(`refused ${chunk.length} bytes`));
+  },
+});
+failing.on("error", (e: Error) => console.log("error", e.message));
+failing.write("abc");
+
+const progress: Progress = { bytes: 0, chunks: 0 };
+const counter = counting(progress);
+const out = new PassThrough();
+counter.pipe(createGunzip()).pipe(out);
+const parts: string[] = [];
+out.on("data", (d: Buffer) => parts.push(d.toString()));
+out.on("end", () => {
+  console.log("gunzip", JSON.stringify(parts.join("")), progress.bytes > 0, progress.chunks);
+});
+counter.end(gzipSync(Buffer.from("compressed payload")));

tests/corpus/6471-stream-data-buffer-or-string.ts

@@ -0,0 +1,44 @@
+// 'data' listeners on runtime streams may declare the chunk as
+// `Buffer | string` (the @types/node ReadStream listener type): the
+// listener receives a Buffer, or a string once an encoding applies.
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+import { PassThrough, Readable } from "node:stream";
+
+function describe(label: string, chunk: Buffer | string): void {
+  if (typeof chunk === "string") console.log(label, "string", JSON.stringify(chunk));
+  else console.log(label, "buffer", chunk.length, JSON.stringify(chunk.toString()));
+}
+
+const raw = new PassThrough();
+raw.on("data", (chunk: Buffer | string) => describe("raw", chunk));
+raw.write("abc");
+raw.end(Buffer.from("de"));
+
+const text = new PassThrough();
+text.setEncoding("utf8");
+text.on("data", (chunk: Buffer | string) => describe("text", chunk));
+text.end("héllo");
+
+const r = new Readable({ read() {} });
+r.on("data", (chunk: Buffer | string) => describe("readable", chunk));
+r.on("end", () => console.log("readable end"));
+r.push("one");
+r.push(null);
+
+const file = path.join(os.tmpdir(), `scriptc-6471-${process.pid}.txt`);
+fs.writeFileSync(file, "file contents\n");
+const stream = fs.createReadStream(file);
+const release = (): void => console.log("released");
+stream.on("data", (chunk: Buffer | string) => describe("file", chunk));
+stream.on("close", release);
+stream.on("error", release);
+stream.on("end", () => {
+  const encoded = fs.createReadStream(file, { encoding: "utf8" });
+  encoded.on("data", (chunk: Buffer | string) => describe("file utf8", chunk));
+  encoded.on("close", () => {
+    fs.unlinkSync(file);
+    console.log("done");
+  });
+});

tests/corpus/6472-child-data-unknown-chunk.ts

@@ -0,0 +1,87 @@
+// A child stream 'data' listener whose chunk is unknown or any receives a
+// Buffer, or a string when setEncoding ran before the listener registered.
+import { spawn } from "node:child_process";
+
+function report(code: number | null, kinds: string[], stdout: string, stderr: string): void {
+  console.log("code", code, "kinds", [...new Set(kinds)].join(","));
+  console.log("stdout", JSON.stringify(stdout));
+  console.log("stderr", JSON.stringify(stderr));
+}
+
+function runEncoded(script: string): Promise<void> {
+  return new Promise((resolve) => {
+    const child = spawn("/bin/sh", ["-c", script], { stdio: "pipe" });
+    let stdout = "";
+    let stderr = "";
+    const kinds: string[] = [];
+    if (child.stdout) {
+      child.stdout.setEncoding("utf8");
+      child.stdout.on("data", (c: any) => {
+        kinds.push(typeof c);
+        stdout += c;
+      });
+    }
+    const err = child.stderr;
+    if (err !== null) {
+      err.setEncoding("utf8");
+      err.on("data", (c: unknown) => {
+        kinds.push(typeof c);
+        stderr += String(c);
+      });
+    }
+    child.on("close", (code) => {
+      report(code, kinds, stdout, stderr);
+      resolve();
+    });
+  });
+}
+
+function runRaw(script: string): Promise<void> {
+  return new Promise((resolve) => {
+    const child = spawn("/bin/sh", ["-c", script], { stdio: "pipe" });
+    let stdout = "";
+    let stderr = "";
+    const kinds: string[] = [];
+    child.stdout?.on("data", (c: unknown) => {
+      kinds.push(Buffer.isBuffer(c) ? "buffer" : typeof c);
+      stdout += String(c);
+    });
+    if (child.stderr) {
+      child.stderr.on("data", (c: any) => {
+        const chunk = String(c);
+        kinds.push(Buffer.isBuffer(c) ? "buffer" : typeof c);
+        stderr += chunk;
+      });
+    }
+    child.on("close", (code) => {
+      report(code, kinds, stdout, stderr);
+      resolve();
+    });
+  });
+}
+
+// setEncoding through one const alias of the stream and the listener
+// through another: both name child.stdout, so chunks are strings.
+function runAliased(script: string): Promise<void> {
+  return new Promise((resolve) => {
+    const child = spawn("/bin/sh", ["-c", script], { stdio: "pipe" });
+    const encoded = child.stdout!;
+    const listened = child.stdout!;
+    let stdout = "";
+    const kinds: string[] = [];
+    encoded.setEncoding("utf8");
+    listened.on("data", (c: unknown) => {
+      kinds.push(typeof c);
+      stdout += String(c);
+    });
+    child.on("close", (code) => {
+      report(code, kinds, stdout, "");
+      resolve();
+    });
+  });
+}
+
+runEncoded("printf 'h\\303\\251llo'; echo warn >&2")
+  .then(() => runRaw("printf raw; echo oops >&2; exit 2"))
+  .then(() => runAliased("printf alias"))
+  .then(() => console.log("done"));

tests/diagnostics/child-process.ts

@@ -42,3 +42,13 @@ const staticWorker = new URL("./child-process.ts", import.meta.url);
 fork(staticWorker, [], { serialization: "advanced" });
 fork(staticWorker, [], { stdio: ["ignore", "ignore", "ignore", "ignore"] });
 // Keep each fence on its own statement so diagnostics remain site-specific.
+// An unknown 'data' chunk takes the setEncoding state established before
+// the listener registers; a setEncoding placed elsewhere leaves it unclear.
+const pipedOut = piped.stdout!;
+pipedOut.on("data", (chunk: unknown) => console.log(String(chunk)));
+pipedOut.setEncoding("utf8");
+// A setEncoding through a copy that can hold the same stream (a `let`
+// binding here) leaves the state unclear as well.
+let pipedErr = piped.stderr!;
+pipedErr.setEncoding("utf8");
+piped.stderr!.on("data", (chunk: unknown) => console.log(String(chunk)));

tests/fixtures/node-types/child-stdin-fences.ts

@@ -1,6 +0,0 @@
-import { spawn } from "node:child_process";
-
-const child = spawn("node", []);
-child.stdin.write("x", "utf8");
-child.stdin.end("x");
-child.stdin.destroy(new Error("stop"));

tests/fixtures/node-types/child-stdin-streams.ts

@@ -0,0 +1,40 @@
+// child.stdin forms beyond write(string): a write with an encoding, end
+// with a final chunk, and destroy with an error observed by an 'error'
+// listener. They compile through the general stream representation, where
+// child.stdin is the child's bridged Writable.
+import { spawn } from "node:child_process";
+
+function feed(): Promise<void> {
+  return new Promise((resolve) => {
+    const child = spawn("cat", [], { stdio: ["pipe", "pipe", "inherit"] });
+    let out = "";
+    child.stdout.on("data", (chunk: Buffer) => {
+      out += chunk.toString();
+    });
+    child.on("close", (code) => {
+      console.log("fed", JSON.stringify(out), code);
+      resolve();
+    });
+    child.stdin.write("first ", "utf8");
+    child.stdin.end("second\n");
+  });
+}
+
+function abort(): Promise<void> {
+  return new Promise((resolve) => {
+    const child = spawn("cat", [], { stdio: ["pipe", "ignore", "inherit"] });
+    child.stdin.on("error", (err) => console.log("stdin error", err.message));
+    child.on("close", (code) => {
+      console.log("aborted", code);
+      resolve();
+    });
+    child.stdin.destroy(new Error("stop"));
+  });
+}
+
+async function main(): Promise<void> {
+  await feed();
+  await abort();
+}
+
+void main();

tests/fixtures/node-types/node-stream-interfaces.ts

@@ -0,0 +1,61 @@
+// @types/node's stream interfaces fed by every producer: NodeJS.ReadableStream
+// and stream.Readable slots take file streams, a gunzip Transform and a
+// child's stdout; NodeJS.WritableStream and stream.Writable slots take file
+// streams and process.stdout.
+import { spawn } from "node:child_process";
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+import { Readable, Writable } from "node:stream";
+import * as zlib from "node:zlib";
+
+function drain(label: string, src: NodeJS.ReadableStream): Promise<string> {
+  return new Promise((resolve, reject) => {
+    let text = "";
+    src.on("data", (chunk: Buffer | string) => {
+      text += chunk.toString();
+    });
+    src.on("error", reject);
+    src.on("end", () => {
+      console.log(label, "end");
+      resolve(text);
+    });
+  });
+}
+
+function emit(dst: NodeJS.WritableStream, line: string): void {
+  dst.write(`${line}\n`);
+}
+
+function finish(dst: Writable): Promise<void> {
+  return new Promise((resolve) => dst.end(() => resolve()));
+}
+
+async function main(): Promise<void> {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), "scriptc-node-types-streams-"));
+  const plain = path.join(dir, "plain.txt");
+  const packed = path.join(dir, "packed.gz");
+  fs.writeFileSync(plain, "from a file\n");
+  fs.writeFileSync(packed, zlib.gzipSync(Buffer.from("from gunzip\n")));
+  const file: Readable = fs.createReadStream(plain);
+  console.log(JSON.stringify(await drain("file", file)));
+  console.log(JSON.stringify(await drain("gunzip", fs.createReadStream(packed).pipe(zlib.createGunzip()))));
+  const child = spawn("sh", ["-c", "printf 'from a child\\n'"], { stdio: ["ignore", "pipe", "inherit"] });
+  if (child.stdout) console.log(JSON.stringify(await drain("child", child.stdout)));
+
+  const target = path.join(dir, "out.txt");
+  const out: Writable = fs.createWriteStream(target);
+  emit(out, "to a file");
+  emit(process.stdout, "to stdout");
+  await finish(out);
+  console.log(JSON.stringify(fs.readFileSync(target, "utf8")));
+  await new Promise<void>((resolve) => {
+    const piped = fs.createReadStream(plain);
+    piped.on("end", () => resolve());
+    piped.pipe(process.stdout);
+  });
+  emit(process.stdout, "stdout still open");
+  fs.rmSync(dir, { recursive: true });
+}
+
+void main();

tests/fixtures/node-types/stream-construction.ts

@@ -0,0 +1,121 @@
+// Streams constructed under @types/node: option-bag Transform and
+// PassThrough streams with @types/node's TransformCallback, a gunzip stage,
+// fs.ReadStream listeners, and child-process listeners declared with
+// unknown chunks and non-null exit types.
+import { spawn } from "node:child_process";
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+import { PassThrough, Transform } from "node:stream";
+import { createGunzip, gzipSync } from "node:zlib";
+
+interface Progress {
+  bytes: number;
+  chunks: number;
+}
+
+function createCounter(progress: Progress) {
+  return new Transform({
+    transform(chunk, _encoding, callback) {
+      const length = typeof chunk === "string" ? Buffer.byteLength(chunk) : chunk.length;
+      progress.bytes += length;
+      progress.chunks++;
+      callback(null, chunk);
+    },
+    flush(callback) {
+      callback();
+    },
+  });
+}
+
+function gunzipThrough(payload: Buffer): Promise<string> {
+  return new Promise((resolve) => {
+    const progress: Progress = { bytes: 0, chunks: 0 };
+    const counter = createCounter(progress);
+    const gunzip = createGunzip();
+    const out = new PassThrough();
+    const parts: string[] = [];
+    counter.pipe(gunzip).pipe(out);
+    out.on("data", (d: Buffer) => parts.push(d.toString()));
+    out.on("end", () => resolve(`${parts.join("")} ${progress.bytes > 0} ${progress.chunks}`));
+    counter.end(payload);
+  });
+}
+
+function readFileStream(file: string): Promise<string> {
+  return new Promise((resolve) => {
+    const stream = fs.createReadStream(file);
+    let text = "";
+    const release = () => console.log("released");
+    stream.on("close", release);
+    stream.on("error", release);
+    stream.on("data", (chunk) => {
+      text += chunk.toString();
+    });
+    stream.on("end", () => resolve(text));
+  });
+}
+
+function run(script: string): Promise<string> {
+  return new Promise((resolve, reject) => {
+    const child = spawn("/bin/sh", ["-c", script], { stdio: "pipe" });
+    let stdout = "";
+    let stderr = "";
+    if (child.stdout) {
+      child.stdout.setEncoding("utf8");
+      child.stdout.on("data", (c) => {
+        stdout += c;
+      });
+    }
+    if (child.stderr) {
+      child.stderr.on("data", (c) => {
+        const chunk = String(c);
+        stderr += chunk;
+      });
+    }
+    child.on("error", reject);
+    child.on("close", (code: number, signal: string) => {
+      if (code === 0) return resolve(stdout);
+      reject(new Error(`Exited with ${code || signal}: ${stderr.trim()}`));
+    });
+  });
+}
+
+// setEncoding returns its stream: a listener on the returned value, or on
+// the stream after `const out = stream.setEncoding(...)`, receives strings.
+function readEncoded(script: string): Promise<string> {
+  return new Promise((resolve) => {
+    const child = spawn("/bin/sh", ["-c", script], { stdio: "pipe" });
+    let outText = "";
+    let errText = "";
+    const out = child.stdout.setEncoding("utf8");
+    out.on("data", (c: unknown) => {
+      outText += typeof c + ":" + String(c) + ";";
+    });
+    const err = child.stderr.setEncoding("utf8");
+    child.stderr.on("data", (c: unknown) => {
+      errText += typeof c + ":" + String(c) + ";";
+    });
+    void err;
+    child.on("close", () => resolve(outText + " err " + errText));
+  });
+}
+
+async function main(): Promise<void> {
+  console.log("gunzip", await gunzipThrough(gzipSync(Buffer.from("payload"))));
+  const file = path.join(os.tmpdir(), `scriptc-stream-construction-${process.pid}.txt`);
+  fs.writeFileSync(file, "file text\n");
+  console.log("file", JSON.stringify(await readFileStream(file)));
+  fs.unlinkSync(file);
+  console.log("run", JSON.stringify(await run("echo hi; echo warn >&2")));
+  console.log("encoded", JSON.stringify(await readEncoded("printf hi; printf warn >&2")));
+  for (const script of ["echo bad >&2; exit 3", "kill -TERM $$"]) {
+    try {
+      await run(script);
+    } catch (e) {
+      console.log("error", (e as Error).message);
+    }
+  }
+}
+
+void main();

tests/harness/__snapshots__/child-process.ts.txt

@@ -91,4 +91,18 @@ child-process.ts:43:64 - error SC2020: 'fork without a fourth-slot IPC channel'
      |                                                                ^~~~~~~~
   44 | // Keep each fence on its own statement so diagnostics remain site-specific.
 
-  hint: the fourth stdio entry must be "ipc"
+  hint: the fourth stdio entry must be "ipc"
+
+child-process.ts:48:21 - error SC1090: untyped child-stream 'data' listeners whose setEncoding state is unclear at their registration sites (declare the chunk as 'Buffer' or 'string') are not supported yet
+
+  47 | const pipedOut = piped.stdout!;
+  48 | pipedOut.on("data", (chunk: unknown) => console.log(String(chunk)));
+     |                     ^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+  49 | pipedOut.setEncoding("utf8");
+
+child-process.ts:54:26 - error SC1090: untyped child-stream 'data' listeners whose setEncoding state is unclear at their registration sites (declare the chunk as 'Buffer' or 'string') are not supported yet
+
+  53 | pipedErr.setEncoding("utf8");
+  54 | piped.stderr!.on("data", (chunk: unknown) => console.log(String(chunk)));
+     |                          ^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+  55 | 

tests/harness/corpus-costs/6468-node-stdio-writable-slot.ts.json

@@ -0,0 +1 @@
+{ "plain": 3000, "san": 3500 }

tests/harness/corpus-costs/6469-node-stdio-writable-js-pipe/main.ts.json

@@ -0,0 +1 @@
+{ "plain": 3000, "san": 3500 }

tests/harness/project-config.test.ts

@@ -518,6 +518,42 @@ test("node-types: stream subclasses extend the @types/node stream classes", asyn
   expect(comparableStderr(native.stderr)).toBe(node.stderr);
 });
 
+test("node-types: NodeJS stream interfaces and stream.Readable/Writable take every stream producer", async () => {
+  const outDir = outDirFor("node-stream-interfaces");
+  const entry = join(nodeTypesDir, "node-stream-interfaces.ts");
+  const result = await compile(entry, {
+    outPath: join(outDir, "node-stream-interfaces"),
+    outDir,
+    sanitize,
+  });
+  expect(result.ok, !result.ok ? JSON.stringify(result.diagnostics, null, 2) : "").toBe(true);
+  if (!result.ok) return;
+  const [native, node] = await Promise.all([
+    execFileAsync(result.binaryPath),
+    execFileAsync(process.execPath, ["--disable-warning=MODULE_TYPELESS_PACKAGE_JSON", entry]),
+  ]);
+  expect(native.stdout).toBe(node.stdout);
+  expect(comparableStderr(native.stderr)).toBe(node.stderr);
+});
+
+test("node-types: streams construct and child listeners take @types/node's parameter types", async () => {
+  const outDir = outDirFor("stream-construction");
+  const entry = join(nodeTypesDir, "stream-construction.ts");
+  const result = await compile(entry, {
+    outPath: join(outDir, "stream-construction"),
+    outDir,
+    sanitize,
+  });
+  expect(result.ok, !result.ok ? JSON.stringify(result.diagnostics, null, 2) : "").toBe(true);
+  if (!result.ok) return;
+  const [native, node] = await Promise.all([
+    execFileAsync(result.binaryPath),
+    execFileAsync(process.execPath, ["--disable-warning=MODULE_TYPELESS_PACKAGE_JSON", entry]),
+  ]);
+  expect(native.stdout).toBe(node.stdout);
+  expect(comparableStderr(native.stderr)).toBe(node.stderr);
+});
+
 test("node-types: promisified setTimeout and setImmediate lower as timers/promises", async () => {
   const outDir = outDirFor("promisify-timers");
   const entry = join(nodeTypesDir, "promisify-timers.ts");
@@ -590,26 +626,22 @@ test("node-types: promisified dns lookups settle with the lookup objects", async
   expect(comparableStderr(native.stderr)).toBe(node.stderr);
 });
 
-test("node-types: unsupported child stdin forms remain named fences", async () => {
-  const dir = outDirFor("node-child-stdin-fences");
-  const entry = join(nodeTypesDir, "child-stdin-fences.ts");
+test("node-types: child stdin forms beyond write(string) compile through the general stream representation", async () => {
+  const dir = outDirFor("node-child-stdin-streams");
+  const entry = join(nodeTypesDir, "child-stdin-streams.ts");
   const result = await compile(entry, {
-    outPath: join(dir, "child-stdin-fences"),
+    outPath: join(dir, "child-stdin-streams"),
     outDir: dir,
     sanitize,
   });
-  expect(result.ok).toBe(false);
-  if (result.ok) return;
-  expect(result.diagnostics.map((diagnostic) => diagnostic.code)).toEqual([
-    "SC2020",
-    "SC2020",
-    "SC2020",
-  ]);
-  expect(result.diagnostics.map((diagnostic) => diagnostic.message)).toEqual([
-    expect.stringContaining("child stdin write with 2 arguments"),
-    expect.stringContaining("Writable.end"),
-    expect.stringContaining("Writable.destroy"),
+  expect(result.ok, !result.ok ? JSON.stringify(result.diagnostics, null, 2) : "").toBe(true);
+  if (!result.ok) return;
+  const [native, node] = await Promise.all([
+    execFileAsync(result.binaryPath),
+    execFileAsync(process.execPath, ["--disable-warning=MODULE_TYPELESS_PACKAGE_JSON", entry]),
   ]);
+  expect(native.stdout).toBe(node.stdout);
+  expect(comparableStderr(native.stderr)).toBe(node.stderr);
 });
 
 test("node-types: callback exec uses the typed ExecException overloads", async () => {