Pipe into untyped destinations by unwrapping boxed native streams

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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

@@ -1,6 +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.

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

@@ -3371,6 +3371,58 @@ function pipeIntoStdio(
   };
 }
 
+/** `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: "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,
@@ -3622,6 +3674,7 @@ export function lowerStreamMethodCall(
     }
     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;

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

@@ -1,7 +1,7 @@
 // Streams that arrive untyped meet NodeJS stream interface slots: an
 // untyped JavaScript function's stream returned as NodeJS.ReadableStream,
-// a checked cast from 'unknown', pipeline() over untyped source and
-// destination stages, and 'for await' over a 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";
@@ -37,6 +37,13 @@ async function main(): Promise<void> {
   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()));