Pump untyped iterable pipeline sources

A pipeline whose first stage is a checked-dynamic value now checks it at
run time, as Node's stage ladder does: a stream is piped as before, and a
string or a value with a Symbol.asyncIterator or Symbol.iterator method is
pumped into the next stage through an async generator over its iterator.
A typed Set, Map, generator or async generator object as the first stage
takes the same path. This covers the callback and stream/promises forms,
with or without function stages and pipeline options.

A write whose callback fails synchronously no longer sets needDrain, so no
'drain' follows the error and a pumped source stops at the failed write,
as in Node.

RUNTIME_ABI_VERSION goes from 37 to 42 for scr_stream_dyn_iterable_source.

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

@@ -3,7 +3,8 @@
   `readable.pipe(process.stdout)` and `readable.pipe(process.stderr)` compile and, as in Node, do not end the process stream.
   When an untyped value reaches `NodeJS.ReadableStream` or `NodeJS.WritableStream`, those two interfaces hold any value, so JavaScript stream objects (constructor-function streams, plain objects with `on` and `pipe`) pass through TypeScript functions typed with them and come back to JavaScript as the same objects.
   Member calls on such a value use the native stream when it is one and call the object's own members otherwise; `readable.pipe()` into a JavaScript destination follows Node's pipe protocol.
-  A `pipeline` stage and a `for await` source that arrive as untyped values must be native streams, or `Cipheriv` and `Decipheriv` values, at run time; other values throw a not-supported error.
+  The first `pipeline` stage may arrive as an untyped value that is a native stream, a `Cipheriv` or `Decipheriv` value, or an iterable: a string, an array, a generator object, or any value with a `Symbol.asyncIterator` or `Symbol.iterator` method, which the pipeline pumps into the next stage as Node does.
+  Later `pipeline` stages and `for await` sources that arrive as untyped values must be native streams, or `Cipheriv` and `Decipheriv` values, at run time; other values throw a not-supported error.
   Native streams expose no `emit` to JavaScript, so a JavaScript stream whose own `pipe()` emits on its destination (the legacy `Stream` pipe) cannot pipe into one.
   `Readable.from` accepts strings, Buffers, arrays, Sets, Maps, Map and Set iterators, generators, async generators, and class iterables and async iterables, and in JavaScript any iterable or async iterable.
   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.
@@ -14,7 +15,7 @@
   `node:stream/consumers` and readline interfaces read object-mode streams of strings and Buffers; other values throw a not-supported error.
   `pipeline` accepts function stages anywhere among its streams: async generator functions (`async function* (source)`, iterating `source` with `for await`), an async generator source function (`async function* ({ signal })`), an async generator function as the destination, and an async function destination whose value resolves the promise form and reaches the callback's second argument.
   A stage is written inline, or named by a function declaration or a `const` bound to a function expression that nothing reassigns.
-  The source may also be an array, a string, a synchronous generator object, or a synchronous generator function without parameters; their elements reach the next stage on the same ticks as in Node.
+  The source may also be an array, a string, a `Set`, a `Map`, a generator or async generator object, or a synchronous generator function without parameters; their elements reach the next stage on the same ticks as in Node.
   Stages may take the `{ signal }` argument.
   With a function destination, `pipeline()` returns an object-mode PassThrough that receives an async generator destination's values, or an async function destination's value unless it is `null` or `undefined`.
   `@types/node` types that result as `NodeJS.WritableStream`, so `for await` over it needs the `PassThrough` type (`pipeline(...) as PassThrough`, or a JSDoc `@type`).

packages/cli/test/native-link-info.test.ts

@@ -72,7 +72,7 @@ describe.runIf(supported)("macOS arm64 native link info", () => {
         minimum_os: "14.0",
       },
       program: { object, entry_symbol: "main" },
-      runtime_abi: { version: 37, marker: "scr_runtime_abi_v37" },
+      runtime_abi: { version: 42, marker: "scr_runtime_abi_v42" },
       runtime_pack: {
         kind: "precompiled",
         package: "@scriptc/runtime-darwin-arm64",
@@ -165,6 +165,6 @@ describe.runIf(supported)("macOS arm64 native link info", () => {
       () => null,
       (failure: { stderr?: string }) => failure,
     );
-    expect(error?.stderr).toContain("scr_runtime_abi_v37");
+    expect(error?.stderr).toContain("scr_runtime_abi_v42");
   });
 });

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

@@ -795,6 +795,7 @@ export const LIB_FN_SYMS: Record<string, string> = {
   "procStream.write": "scr_proc_stream_write",
   "stream.stdioWritable": "scr_stream_stdio_writable",
   "stream.unboxDyn": "scr_stream_unbox_dyn",
+  "stream.dynIterableSource": "scr_stream_dyn_iterable_source",
   "stream.handleStream": "scr_stream_handle",
   "stream.pipeDyn": "scr_stream_pipe_dyn",
   "stream.unpipeDyn": "scr_stream_unpipe_dyn",

packages/compiler/src/backend/runtime-abi.ts

@@ -2,8 +2,8 @@
  * of the link contract: a program object keeps this symbol undefined and the
  * matching runtime defines it, so incompatible manual links fail before the
  * program can start. */
-export const RUNTIME_ABI_VERSION = 37 as const;
-export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v37" as const;
+export const RUNTIME_ABI_VERSION = 42 as const;
+export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v42" as const;
 
 /** Object consumption is useful today, but the complete scr_* surface may
  * still change before 1.0. The versioned marker prevents accidental mixing;

packages/compiler/src/backend/runtime-pack-targets.test.ts

@@ -52,7 +52,7 @@ async function fixture(target: NativeTargetSpec, workers = false) {
       object_format: target.objectFormat,
       minimum_os: target.minimumOs,
     },
-    runtime_abi: { version: 37, marker: "scr_runtime_abi_v37" },
+    runtime_abi: { version: 42, marker: "scr_runtime_abi_v42" },
     compiler: { command: "fixture", identity: "fixture", target: target.llvmTriple },
     macros: { executable: [], excluded: ["SCR_LIB"], sanitizer: "external-toolchain-required" },
     flavors: {

packages/compiler/src/backend/runtime-pack.test.ts

@@ -128,7 +128,7 @@ async function fixture(target: NativeTargetSpec = MACOS_ARM64_TARGET) {
       object_format: target.objectFormat,
       minimum_os: target.minimumOs,
     },
-    runtime_abi: { version: 37, marker: "scr_runtime_abi_v37" },
+    runtime_abi: { version: 42, marker: "scr_runtime_abi_v42" },
     compiler: {
       command: "clang",
       identity: "fixture clang",

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

@@ -2837,6 +2837,107 @@ export function streamClassAliasDecl(
 const PIPELINE_STAGE_EXPECTED =
   "of type function or an instance of Blob, ReadableStream, WritableStream, Stream, Iterable, AsyncIterable, or Promise or { readable, writable } pair";
 
+/** A checked-dynamic pipeline stage (an untyped source or destination):
+ * a boxed native stream unwraps at run time; other values (functions,
+ * pairs) throw the not-supported Error there. */
+function unboxDynStage(lowerer: Lowerer, v: IrExpr, node: ts.Expression): IrExpr {
+  noteUntypedStreamFlow(lowerer, null);
+  return {
+    kind: "libCall",
+    fn: "stream.unboxDyn",
+    args: [
+      v,
+      strLit(
+        lowerer.dynNodeStreams
+          ? "pipeline stages other than native streams (JavaScript stream objects, iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet"
+          : "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),
+  };
+}
+
+/** A pipeline whose first stage is checked-dynamic, lowered with that
+ * stage already evaluated: `expr` is the stream it unboxes to, or, when
+ * `iterable` is set, the value to pump as an iterable source. */
+interface DynPipelineSource {
+  expr: IrExpr;
+  iterable: boolean;
+}
+
+/** pipeline() whose first stage is a stream argument (not a function
+ * stage, an iterable of a static type, or a cipher value), lowered here.
+ * When the value is checked-dynamic, Node keeps a stream there and pipes
+ * it, and keeps an iterable (a string, an array, a generator, any value
+ * with an iterator method) and pumps it into the next stage. The value is
+ * evaluated once, and stream.dynIterableSource picks the lowering at run
+ * time. Null when the first stage is not such an argument. */
+function lowerDynSourcePipeline(
+  lowerer: Lowerer,
+  node: ts.Expression,
+  loc: SrcLoc,
+  build: (source: DynPipelineSource) => IrExpr,
+): IrExpr | null {
+  if (pipelineStageFunction(lowerer, node) !== null) return null;
+  if (iterableSourceType(lowerer, node) !== null || isValueStream(lowerer, node)) return null;
+  let value = lowerer.lowerExpr(node);
+  // A Set, a Map or a generator object is iterated through its
+  // checked-dynamic value.
+  if (BOXED_SOURCE_KINDS.has(value.type.kind) && lowerer.dynConvertible(value.type)) {
+    value = { kind: "dynFrom", value, type: DYN, loc: value.loc };
+  }
+  if (value.type.kind !== "dyn") {
+    let stream: IrExpr;
+    try {
+      stream = streamArgOf(lowerer, node, value, "pipeline");
+    } catch (e) {
+      if (e instanceof StreamArgTypeThrow) return e.expr;
+      throw e;
+    }
+    return build({ expr: stream, iterable: false });
+  }
+  const local = lowerer.declareHiddenLocal("%pipelineSource", DYN);
+  const ref = varRef(local.id, DYN, loc);
+  const stream = build({ expr: unboxDynStage(lowerer, ref, node), iterable: false });
+  const iterable = build({ expr: ref, iterable: true });
+  const type = stream.type;
+  return {
+    kind: "seqExpr",
+    stmts: [{ kind: "varDecl", localId: local.id, init: value, loc }],
+    result: {
+      kind: "ternary",
+      cond: { kind: "libCall", fn: "stream.dynIterableSource", args: [ref], type: BOOL, loc },
+      then: lowerer.coerceToExpected(iterable, type),
+      else_: stream,
+      type,
+      loc,
+    },
+    type,
+    loc,
+  };
+}
+
+/** The static types of a pipeline source that lowerDynSourcePipeline boxes
+ * as checked-dynamic values. An array or a string reaches it when the
+ * checker types the argument as any while its value keeps a static type. */
+const BOXED_SOURCE_KINDS: ReadonlySet<IrType["kind"]> = new Set([
+  "set",
+  "map",
+  "generator",
+  "array",
+  "string",
+]);
+
+/** lowerPipelineStages' source arguments for a checked-dynamic first stage. */
+function stagesSource(
+  src: DynPipelineSource | null,
+): [ReadonlyMap<number, IrExpr> | undefined, IrExpr | undefined] {
+  if (src === null) return [undefined, undefined];
+  return src.iterable ? [undefined, src.expr] : [new Map([[0, src.expr]]), undefined];
+}
+
 /** A stream-rooted argument, lowered — or the pointed fence (pipeline's
  * iterable/generator/function stages have no lowering yet). `what` picks
  * the validation ladder: finished() rejects EVERY non-stream through
@@ -2848,7 +2949,16 @@ function lowerStreamArg(
   node: ts.Expression,
   what: "finished" | "pipeline" | "addAbortSignal" | "compose",
 ): IrExpr {
-  const lowered = lowerer.lowerExpr(node);
+  return streamArgOf(lowerer, node, lowerer.lowerExpr(node), what);
+}
+
+/** lowerStreamArg over the argument's lowered value. */
+function streamArgOf(
+  lowerer: Lowerer,
+  node: ts.Expression,
+  lowered: IrExpr,
+  what: "finished" | "pipeline" | "addAbortSignal" | "compose",
+): IrExpr {
   // A Hash, Hmac, Cipheriv or Decipheriv stage is its Transform; pipeline
   // and compose stages keep the value alive until the stream settles.
   const v = isValueStream(lowerer, node)
@@ -2857,24 +2967,7 @@ function lowerStreamArg(
   // 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") {
-    noteUntypedStreamFlow(lowerer, null);
-    return {
-      kind: "libCall",
-      fn: "stream.unboxDyn",
-      args: [
-        v,
-        strLit(
-          lowerer.dynNodeStreams
-            ? "pipeline stages other than native streams (JavaScript stream objects, iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet"
-            : "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),
-    };
-  }
+  if (v.type.kind === "dyn" && what === "pipeline") return unboxDynStage(lowerer, v, 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
@@ -3210,57 +3303,82 @@ export function lowerStreamModuleCall(
           ? pipelineOptions(lowerer, lastArg)
           : null;
       const stageArgs = opts ? args.slice(0, -1) : args;
-      if (hasPipelineFunctionStage(lowerer, stageArgs)) {
-        const staged = lowerPipelineStages(lowerer, stageArgs, loc);
-        const type = staged.destPromise ?? { kind: "promise", inner: VOID };
-        // A promise destination's value settles the result (same type).
-        if (!opts) {
-          return staged.wrap({
-            kind: "libCall",
-            fn: "sp.pipelineStages",
-            args: staged.args,
-            type,
-            loc,
-          });
-        }
-        // the options: end, and the signal (flags bit 6)
-        const flagsArg = staged.args[3]!;
-        if (flagsArg.kind !== "numLit")
-          throw new InternalCompilerError("pipeline flags not a literal");
-        const args = [...staged.args, boolLit(opts.end, loc)];
-        if (opts.signal) {
-          args[3] = numLit(flagsArg.value | 64, loc);
-          args.push(opts.signal);
-        }
-        return staged.wrap({ kind: "libCall", fn: "sp.pipelineStagesOpts", args, type, loc });
-      }
-      let streams: IrExpr[];
-      try {
-        streams = stageArgs.map((a) => lowerStreamArg(lowerer, a, "pipeline"));
-      } catch (e) {
-        if (e instanceof StreamArgTypeThrow) return e.expr;
-        throw e;
-      }
-      if (opts) {
-        const head = [numLit(streams.length, loc), ...streams, boolLit(opts.end, loc)];
-        return {
-          kind: "libCall",
-          fn: opts.signal ? "sp.pipelineOptsSignal" : "sp.pipelineOpts",
-          args: opts.signal ? [...head, opts.signal] : head,
-          type: { kind: "promise", inner: VOID },
-          loc,
-        };
-      }
-      return {
-        kind: "libCall",
-        fn: "sp.pipeline",
-        args: [numLit(streams.length, loc), ...streams],
-        type: { kind: "promise", inner: VOID },
-        loc,
-      };
+      const build = (src: DynPipelineSource | null): IrExpr =>
+        lowerPromisePipeline(lowerer, stageArgs, opts, src, loc);
+      return lowerDynSourcePipeline(lowerer, stageArgs[0]!, loc, build) ?? build(null);
     }
     return null;
   }
+  return lowerStreamModuleCallbackCall(lowerer, call, bi, loc);
+}
+
+/** stream/promises pipeline(...stages[, options]), with the first stage
+ * pre-lowered when it is checked-dynamic (lowerDynSourcePipeline). */
+function lowerPromisePipeline(
+  lowerer: Lowerer,
+  stageArgs: readonly ts.Expression[],
+  opts: ReturnType<typeof pipelineOptions> | null,
+  src: DynPipelineSource | null,
+  loc: SrcLoc,
+): IrExpr {
+  if (src?.iterable === true || hasPipelineFunctionStage(lowerer, stageArgs)) {
+    const staged = lowerPipelineStages(lowerer, stageArgs, loc, ...stagesSource(src));
+    const type = staged.destPromise ?? { kind: "promise", inner: VOID };
+    // A promise destination's value settles the result (same type).
+    if (!opts) {
+      return staged.wrap({
+        kind: "libCall",
+        fn: "sp.pipelineStages",
+        args: staged.args,
+        type,
+        loc,
+      });
+    }
+    // the options: end, and the signal (flags bit 6)
+    const flagsArg = staged.args[3]!;
+    if (flagsArg.kind !== "numLit") throw new InternalCompilerError("pipeline flags not a literal");
+    const args = [...staged.args, boolLit(opts.end, loc)];
+    if (opts.signal) {
+      args[3] = numLit(flagsArg.value | 64, loc);
+      args.push(opts.signal);
+    }
+    return staged.wrap({ kind: "libCall", fn: "sp.pipelineStagesOpts", args, type, loc });
+  }
+  let streams: IrExpr[];
+  try {
+    streams = stageArgs.map((a, i) =>
+      i === 0 && src !== null ? src.expr : lowerStreamArg(lowerer, a, "pipeline"),
+    );
+  } catch (e) {
+    if (e instanceof StreamArgTypeThrow) return e.expr;
+    throw e;
+  }
+  if (opts) {
+    const head = [numLit(streams.length, loc), ...streams, boolLit(opts.end, loc)];
+    return {
+      kind: "libCall",
+      fn: opts.signal ? "sp.pipelineOptsSignal" : "sp.pipelineOpts",
+      args: opts.signal ? [...head, opts.signal] : head,
+      type: { kind: "promise", inner: VOID },
+      loc,
+    };
+  }
+  return {
+    kind: "libCall",
+    fn: "sp.pipeline",
+    args: [numLit(streams.length, loc), ...streams],
+    type: { kind: "promise", inner: VOID },
+    loc,
+  };
+}
+
+/** The node:stream and stream/consumers members of lowerStreamModuleCall. */
+function lowerStreamModuleCallbackCall(
+  lowerer: Lowerer,
+  call: ts.CallExpression,
+  bi: { module: string; member: string },
+  loc: SrcLoc,
+): IrExpr | null {
   if (bi.module === "stream/consumers") {
     const args = call.arguments.nodes;
     // text/json/buffer/arrayBuffer(stream) → a pending promise the
@@ -3411,67 +3529,9 @@ export function lowerStreamModuleCall(
         "the supported form is pipeline(source, ...streams, callback)",
       );
     }
-    if (hasPipelineFunctionStage(lowerer, args.slice(0, -1))) {
-      const staged = lowerPipelineStages(lowerer, args.slice(0, -1), loc);
-      const cbNode = args[args.length - 1]!;
-      const cbParams =
-        ts.isArrowFunction(cbNode) || ts.isFunctionExpression(cbNode)
-          ? cbNode.parameters.nodes.length
-          : 0;
-      if (staged.dest === 2 || (staged.dest === 1 && cbParams > 1)) {
-        // callback(err, value): the destination's value (undefined for
-        // an async generator) reaches the callback through the
-        // checked-dynamic path.
-        const cb = lowerer.lowerExprExpecting(args[args.length - 1]!, DYN);
-        // SCR_PIPE_TYPED_CB: a typed callback whose error parameter cannot
-        // hold undefined takes null for no error; others take undefined.
-        const stagedArgs = [...staged.args];
-        const flagsArg = stagedArgs[3]!;
-        if (flagsArg.kind !== "numLit")
-          throw new InternalCompilerError("pipeline flags not a literal");
-        if (
-          !isJsSourceFile(ts.getSourceFileOfNode(call)) &&
-          !callbackErrorAdmitsUndefined(lowerer, cb)
-        )
-          stagedArgs[3] = numLit(flagsArg.value | 128, loc);
-        return staged.wrap({
-          kind: "libCall",
-          fn: "stream.pipelineStagesDyn",
-          args: [...stagedArgs, cb],
-          type: staged.last,
-          loc,
-        });
-      }
-      const { cb, dyn } = lowerEosCallback(
-        lowerer,
-        "pipeline",
-        args[args.length - 1]!,
-        staged.last,
-      );
-      return staged.wrap({
-        kind: "libCall",
-        fn: dyn ? "stream.pipelineStagesDyn" : "stream.pipelineStages",
-        args: [...staged.args, cb],
-        type: staged.last,
-        loc,
-      });
-    }
-    let streams: IrExpr[];
-    try {
-      streams = args.slice(0, -1).map((a) => lowerStreamArg(lowerer, a, "pipeline"));
-    } catch (e) {
-      if (e instanceof StreamArgTypeThrow) return e.expr;
-      throw e;
-    }
-    const last = streams[streams.length - 1]!;
-    const { cb, dyn } = lowerEosCallback(lowerer, "pipeline", args[args.length - 1]!, last.type);
-    return {
-      kind: "libCall",
-      fn: dyn ? "stream.pipelineDyn" : "stream.pipeline",
-      args: [numLit(streams.length, loc), ...streams, cb],
-      type: last.type,
-      loc,
-    };
+    const build = (src: DynPipelineSource | null): IrExpr =>
+      lowerCallbackPipeline(lowerer, call, args, src, loc);
+    return lowerDynSourcePipeline(lowerer, args[0]!, loc, build) ?? build(null);
   }
 
   if (bi.member === "compose") {
@@ -3484,6 +3544,75 @@ export function lowerStreamModuleCall(
   return null;
 }
 
+/** pipeline(...stages, callback), with the first stage pre-lowered when it
+ * is checked-dynamic (lowerDynSourcePipeline). */
+function lowerCallbackPipeline(
+  lowerer: Lowerer,
+  call: ts.CallExpression,
+  args: readonly ts.Expression[],
+  src: DynPipelineSource | null,
+  loc: SrcLoc,
+): IrExpr {
+  if (src?.iterable === true || hasPipelineFunctionStage(lowerer, args.slice(0, -1))) {
+    const staged = lowerPipelineStages(lowerer, args.slice(0, -1), loc, ...stagesSource(src));
+    const cbNode = args[args.length - 1]!;
+    const cbParams =
+      ts.isArrowFunction(cbNode) || ts.isFunctionExpression(cbNode)
+        ? cbNode.parameters.nodes.length
+        : 0;
+    if (staged.dest === 2 || (staged.dest === 1 && cbParams > 1)) {
+      // callback(err, value): the destination's value (undefined for
+      // an async generator) reaches the callback through the
+      // checked-dynamic path.
+      const cb = lowerer.lowerExprExpecting(args[args.length - 1]!, DYN);
+      // SCR_PIPE_TYPED_CB: a typed callback whose error parameter cannot
+      // hold undefined takes null for no error; others take undefined.
+      const stagedArgs = [...staged.args];
+      const flagsArg = stagedArgs[3]!;
+      if (flagsArg.kind !== "numLit")
+        throw new InternalCompilerError("pipeline flags not a literal");
+      if (
+        !isJsSourceFile(ts.getSourceFileOfNode(call)) &&
+        !callbackErrorAdmitsUndefined(lowerer, cb)
+      )
+        stagedArgs[3] = numLit(flagsArg.value | 128, loc);
+      return staged.wrap({
+        kind: "libCall",
+        fn: "stream.pipelineStagesDyn",
+        args: [...stagedArgs, cb],
+        type: staged.last,
+        loc,
+      });
+    }
+    const { cb, dyn } = lowerEosCallback(lowerer, "pipeline", args[args.length - 1]!, staged.last);
+    return staged.wrap({
+      kind: "libCall",
+      fn: dyn ? "stream.pipelineStagesDyn" : "stream.pipelineStages",
+      args: [...staged.args, cb],
+      type: staged.last,
+      loc,
+    });
+  }
+  let streams: IrExpr[];
+  try {
+    streams = args
+      .slice(0, -1)
+      .map((a, i) => (i === 0 && src !== null ? src.expr : lowerStreamArg(lowerer, a, "pipeline")));
+  } catch (e) {
+    if (e instanceof StreamArgTypeThrow) return e.expr;
+    throw e;
+  }
+  const last = streams[streams.length - 1]!;
+  const { cb, dyn } = lowerEosCallback(lowerer, "pipeline", args[args.length - 1]!, last.type);
+  return {
+    kind: "libCall",
+    fn: dyn ? "stream.pipelineDyn" : "stream.pipeline",
+    args: [numLit(streams.length, loc), ...streams, cb],
+    type: last.type,
+    loc,
+  };
+}
+
 /** A pipeline() function stage: an inline function or arrow function, or
  * an identifier bound to a function declaration or to a `const`
  * initialized with one, which nothing reassigns. */
@@ -3788,6 +3917,237 @@ function lowerWebStreamSource(
   };
 }
 
+/** A checked-dynamic pipeline source that is an iterable at run time
+ * (stream.dynIterableSource), as an async generator closure that iterates
+ * it as Node's pipeline does with `for await`: the value's async iterator,
+ * or its sync iterator stepped like the typed iterable sources. Each value
+ * is yielded; leaving early closes the iterator through its return(). */
+function lowerDynIterableSource(
+  lowerer: Lowerer,
+  value: IrExpr,
+  loc: SrcLoc,
+): { closure: IrExpr; out: SyncGenType } | null {
+  const out: SyncGenType = {
+    kind: "generator",
+    async: true,
+    yieldT: DYN,
+    retT: VOID,
+    nextT: UNDEFINED_T,
+  };
+  const genResultT = genResultRecord(DYN, VOID, lowerer.shapes, lowerer.unions);
+  if (!genResultT) return null;
+  const key = "dynIterableSource";
+  let maker = lowerer.valueHelpers.get(key);
+  if (!maker) {
+    maker = `%stream.dynIterableSource.${lowerer.valueHelpers.size}`;
+    const genName = `${maker}.gen`;
+    const frame = new LiftedFrame(loc);
+    const srcId = frame.local("src", DYN, false);
+    const itId = frame.local("it", DYN, false);
+    const syncId = frame.local("sync", BOOL, false);
+    const nextId = frame.local("next", DYN, false);
+    const stepId = frame.local("step", DYN, true);
+    const activeId = frame.local("active", BOOL, true);
+    const closeId = frame.local("close", DYN, false);
+    const it = (): IrExpr => varRef(itId, DYN, loc);
+    const sync = (): IrExpr => varRef(syncId, BOOL, loc);
+    const step = (): IrExpr => varRef(stepId, DYN, loc);
+    const get = (obj: IrExpr, name: string): IrExpr => ({
+      kind: "dynKeyGet",
+      value: obj,
+      key: strLit(name, loc),
+      type: DYN,
+      loc,
+    });
+    const awaitDyn = (v: IrExpr): IrExpr => ({
+      kind: "libCall",
+      fn: "async.awaitDyn",
+      args: [v],
+      type: DYN,
+      loc,
+    });
+    const call = (callee: IrExpr, calleeName: string): IrExpr => ({
+      kind: "dynCall",
+      callee,
+      receiver: it(),
+      calleeName,
+      args: [],
+      type: DYN,
+      loc,
+    });
+    const setActive = (on: boolean): IrStmt => ({
+      kind: "assign",
+      localId: activeId,
+      value: boolLit(on, loc),
+      loc,
+    });
+    const nextCall = (): IrExpr => call(varRef(nextId, DYN, loc), "iterator method");
+    const body: IrStmt[] = [
+      {
+        kind: "varDecl",
+        localId: itId,
+        init: {
+          kind: "libCall",
+          fn: "dyn.iteratorResult",
+          args: [
+            {
+              kind: "libCall",
+              fn: "dyn.asyncIteratorOpen",
+              args: [varRef(srcId, DYN, loc), strLit("", loc)],
+              type: DYN,
+              loc,
+            },
+          ],
+          type: DYN,
+          loc,
+        },
+        loc,
+      },
+      {
+        kind: "varDecl",
+        localId: syncId,
+        init: { kind: "libCall", fn: "dyn.asyncIteratorWasSync", args: [], type: BOOL, loc },
+        loc,
+      },
+      { kind: "varDecl", localId: nextId, init: get(it(), "next"), loc },
+      { kind: "varDecl", localId: stepId, init: dynUndefinedExpr(loc), loc },
+      { kind: "varDecl", localId: activeId, init: boolLit(false, loc), loc },
+      {
+        kind: "tryCatch",
+        tryBody: [
+          {
+            kind: "while",
+            cond: boolLit(true, loc),
+            body: [
+              setActive(false),
+              {
+                kind: "assign",
+                localId: stepId,
+                value: {
+                  kind: "libCall",
+                  fn: "dyn.iteratorResult",
+                  args: [
+                    {
+                      kind: "ternary",
+                      cond: sync(),
+                      then: nextCall(),
+                      else_: awaitDyn(nextCall()),
+                      type: DYN,
+                      loc,
+                    },
+                  ],
+                  type: DYN,
+                  loc,
+                },
+                loc,
+              },
+              {
+                kind: "if",
+                cond: {
+                  kind: "dynTest",
+                  test: "truthy",
+                  value: get(step(), "done"),
+                  type: BOOL,
+                  loc,
+                },
+                then: [{ kind: "break", loc }],
+                else_: null,
+                loc,
+              },
+              setActive(true),
+              {
+                kind: "exprStmt",
+                expr: { kind: "yieldExpr", value: get(step(), "value"), type: VOID, loc },
+                loc,
+              },
+            ],
+            loc,
+          },
+        ],
+        catchBody: null,
+        catchLocalId: null,
+        finallyBody: [
+          {
+            kind: "if",
+            cond: varRef(activeId, BOOL, loc),
+            then: [
+              { kind: "varDecl", localId: closeId, init: get(it(), "return"), loc },
+              {
+                kind: "if",
+                cond: {
+                  kind: "unary",
+                  op: "!",
+                  operand: {
+                    kind: "dynTest",
+                    test: "nullish",
+                    value: varRef(closeId, DYN, loc),
+                    type: BOOL,
+                    loc,
+                  },
+                  type: BOOL,
+                  loc,
+                },
+                then: [
+                  {
+                    kind: "exprStmt",
+                    expr: awaitDyn(call(varRef(closeId, DYN, loc), "iterator return")),
+                    loc,
+                  },
+                ],
+                else_: null,
+                loc,
+              },
+            ],
+            else_: null,
+            loc,
+          },
+        ],
+        loc,
+      },
+      { kind: "return", value: null, loc },
+    ];
+    const srcLocal = frame.locals.find((l) => l.id === srcId)!;
+    srcLocal.boxed = true;
+    lowerer.valueHelpers.set(key, maker);
+    lowerer.liftedFns.push({
+      name: genName,
+      params: [],
+      returnType: VOID,
+      async: true,
+      captures: [{ localId: srcId, name: "src", type: DYN }],
+      locals: frame.locals,
+      body,
+      generator: { yieldT: DYN, nextT: UNDEFINED_T, resultType: genResultT },
+      loc,
+    });
+    const makerSrc: IrLocal = { id: "src.0", name: "src", type: DYN, mutable: false, boxed: true };
+    lowerer.liftedFns.push({
+      name: maker,
+      params: [{ localId: makerSrc.id, name: makerSrc.name, type: DYN }],
+      returnType: funcOf([], out),
+      locals: [makerSrc],
+      body: [
+        {
+          kind: "return",
+          value: {
+            kind: "closure",
+            fnName: genName,
+            captures: [makerSrc.id],
+            type: funcOf([], out),
+            loc,
+          },
+          loc,
+        },
+      ],
+      loc,
+    });
+  }
+  return {
+    closure: { kind: "call", callee: maker, args: [value], type: funcOf([], out), loc },
+    out,
+  };
+}
+
 function lowerIterableSource(
   lowerer: Lowerer,
   node: ts.Expression,
@@ -3951,6 +4311,7 @@ function lowerPipelineStages(
   stageArgs: readonly ts.Expression[],
   loc: SrcLoc,
   preLowered?: ReadonlyMap<number, IrExpr>,
+  dynSource?: IrExpr,
 ): PipelineStagesLowering {
   if (stageArgs.length > 52) {
     lowerer.noLowering(
@@ -3975,6 +4336,15 @@ function lowerPipelineStages(
   const argOrder: { list: IrExpr[]; index: number }[] = [];
   for (let j = 0; j < stageArgs.length; j++) {
     const arg = stageArgs[j]!;
+    if (j === 0 && dynSource !== undefined) {
+      const source = lowerDynIterableSource(lowerer, dynSource, loc);
+      if (source === null) lowerer.noLowering("pipeline over this source", arg);
+      positions += 1;
+      argOrder.push({ list: stages, index: stages.length });
+      stages.push(source!.closure);
+      srcT = source!.out;
+      continue;
+    }
     // compose's head and tail functions arrive as Duplex.from streams
     const fn = preLowered?.has(j) ? null : pipelineStageFunction(lowerer, arg);
     const syncSourceFn =

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

@@ -91,6 +91,7 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "procStream.write",
   "child.stdioStream",
   "stream.unboxDyn",
+  "stream.dynIterableSource",
   "stream.handleStream",
   "stream.pipeDyn",
   "stream.unpipeDyn",

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

@@ -2786,6 +2786,10 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
     argTypes: [DYN, STRING],
     result: { kind: "object", className: "%Duplex" },
   },
+  /** Whether pipeline() pumps a checked-dynamic first stage as an
+   * iterable (a string, or a value with an iterator method that is not a
+   * stream with pipe() and on()) instead of piping it as a stream. */
+  "stream.dynIterableSource": { argTypes: [DYN], result: BOOL },
   /** The Transform a Cipheriv or Decipheriv reads and writes as a stream
    * (+1, created on first use and owned by the cipher). */
   "stream.handleStream": {

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

@@ -263,6 +263,7 @@ const STREAM_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "stream.prop",
   "stream.stdioWritable",
   "stream.unboxDyn",
+  "stream.dynIterableSource",
   "stream.handleStream",
   "stream.unpipeDyn",
   "stream.workerStdio",

packages/compiler/test/native-codegen-integration.test.ts

@@ -378,7 +378,7 @@ done:
     expect(await symbols(helperObject, ["-u"])).toEqual(await symbols(clangObject, ["-u"]));
     expect(await symbols(helperObject, ["-gU"])).toEqual(await symbols(clangObject, ["-gU"]));
     expect(await symbols(helperObject, ["-gU"])).toEqual(["0000000000000000 T _main"]);
-    expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v37");
+    expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v42");
 
     const clangExe = join(dir, "clang-program");
     const linkDriver = join(dir, "clang-driver.c");
@@ -712,7 +712,7 @@ done:
       .split("\n");
     expect(undefinedSymbols).toContain("_sf_scale");
     expect(undefinedSymbols).toContain("_sf_callback_mix");
-    expect(undefinedSymbols).toContain("_scr_runtime_abi_v37");
+    expect(undefinedSymbols).toContain("_scr_runtime_abi_v42");
   });
 
   test("helper and clang object paths retain the same outbound FFI C ABI", async () => {

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7503-pipeline-untyped-sources.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7503-pipeline-untyped-sources.mjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7504-pipeline-untyped-source-errors.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7504-pipeline-untyped-source-errors.mjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7505-pipeline-untyped-sources.cjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7505-pipeline-untyped-sources.cjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7506-pipeline-any-source.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7506-pipeline-any-source.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7507-stream-write-error-drain.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7507-stream-write-error-drain.mjs"
+ ],
+ "diags": []
+}

packages/runtime-darwin-arm64/scripts/build.mjs

@@ -366,7 +366,7 @@ async function build() {
       package: packageManifest.name,
       version: packageManifest.version,
       target: RUNTIME_PACK_MATRIX.target,
-      runtime_abi: { version: 37, marker: "scr_runtime_abi_v37" },
+      runtime_abi: { version: 42, marker: "scr_runtime_abi_v42" },
       compiler: {
         command: compiler,
         identity: compilerVersion,

packages/runtime-pack-common/scripts/build.mjs

@@ -469,7 +469,7 @@ async function build() {
       package: packageManifest.name,
       version: packageManifest.version,
       target: matrix.target,
-      runtime_abi: { version: 37, marker: "scr_runtime_abi_v37" },
+      runtime_abi: { version: 42, marker: "scr_runtime_abi_v42" },
       compiler: {
         command: compiler,
         identity: compilerVersion,

packages/runtime/src/scr_console.c

@@ -148,7 +148,7 @@ static void scr_enable_utf8_console_output(void) {
 }
 #endif
 
-void scr_runtime_abi_v37(void) {}
+void scr_runtime_abi_v42(void) {}
 
 void scr_init(void) {
 #ifdef SCR_WORKERS

packages/runtime/src/scr_core.h

@@ -81,7 +81,7 @@ void scr_init(void);
 /* Program objects emitted by the bundled LLVM helper reference this symbol.
  * Its versioned spelling makes a mismatched manual runtime link fail before
  * the program can start. */
-void scr_runtime_abi_v37(void);
+void scr_runtime_abi_v42(void);
 void scr_runtime_workers_v8(void);
 
 /* ── the trap funnel (scr_console.c; scr_library.c under -DSCR_LIB) ──────

packages/runtime/src/scr_stream.c

@@ -6866,10 +6866,12 @@ static bool scr_stream_write_chunk_enc(ScrStream *s, ScrBytes *chunk /*borrowed*
     scr_stream_do_write(s, scr_stream_wchunk_retain(st, chunk), cb, enc);
   }
   /* Node computes the answer AFTER the (possibly synchronous) write:
-   * a sync completion has already drained its bytes from length. */
-  bool ret = st->w.length < st->w.hwm && !st->destroyed && !st->errored;
-  if (!ret) st->w.need_drain = true;
-  return ret;
+   * a sync completion has already drained its bytes from length. Only
+   * the length sets needDrain; an errored or destroyed stream answers
+   * false without waiting for 'drain'. */
+  bool below = st->w.length < st->w.hwm;
+  if (!below) st->w.need_drain = true;
+  return below && !st->destroyed && !st->errored;
 }
 
 static bool scr_stream_write_chunk(ScrStream *s, ScrBytes *chunk /*borrowed*/, ScrClosure *cb /*moves*/) {
@@ -8407,3 +8409,31 @@ ScrStream *scr_stream_unbox_dyn(const ScrDyn *value, const ScrStr *message) {
   scr_throw_error_msg_code(SCR_ERR_ERROR, message->data, message->len, "SC2020");
   return NULL;
 }
+
+/* pipeline.js's first stage: a value with pipe() and on() methods is a
+ * readable node stream, which the stage ladder tests before isIterable;
+ * a string, or a value with a Symbol.asyncIterator or Symbol.iterator
+ * method, is an iterable the pipeline pumps into the next stage. */
+bool scr_stream_dyn_iterable_source(const ScrDyn *value) {
+  ScrStream *stream = scr_stream_of_handle(value);
+  if (!stream) stream = scr_stream_boxed(value);
+  if (stream) {
+    scr_stream_release(stream);
+    return false;
+  }
+  if (value->kind == SCR_DYN_STR) return true;
+  if (value->kind == SCR_DYN_OBJ) {
+    ScrDyn *pipe = scr_dyn_obj_read(value, "pipe", 4);
+    if (!pipe) return false;
+    bool streamish = pipe->kind == SCR_DYN_FUNC;
+    scr_dyn_release(pipe);
+    if (streamish) {
+      ScrDyn *on = scr_dyn_obj_read(value, "on", 2);
+      if (!on) return false;
+      streamish = on->kind == SCR_DYN_FUNC;
+      scr_dyn_release(on);
+      if (streamish) return false;
+    }
+  }
+  return scr_dyn_iterable_kind(value) > 0;
+}

packages/runtime/src/scr_stream.h

@@ -145,6 +145,10 @@ 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);
+/* Whether pipeline() keeps a checked-dynamic first stage as an iterable
+ * to pump rather than as a stream (false with an exception pending when
+ * reading the value's members throws). */
+bool scr_stream_dyn_iterable_source(const struct ScrDyn *value);
 /* Handles that are Transforms (ScrDynHandleOps.transform: Cipheriv and
  * Decipheriv). scr_stream_of_handle: the handle's Transform (+1), built on
  * first use, or NULL when the value is not such a handle.

tests/corpus/7503-pipeline-untyped-sources.mjs

@@ -0,0 +1,102 @@
+// pipeline() over untyped iterable sources: arrays, strings, Sets, generator
+// objects, async generator objects, class iterables and object values,
+// into Writables, a Transform chain, a function stage and the promise form.
+// Each source is pumped as Node pumps an iterable, on the same ticks.
+import { pipeline, Writable, Transform } from "node:stream";
+import { pipeline as pipelineP } from "node:stream/promises";
+
+const log = [];
+function show(chunk) {
+  return typeof chunk === "string" ? chunk : Buffer.isBuffer(chunk) ? chunk.toString() : JSON.stringify(chunk);
+}
+function objectSink(label) {
+  return new Writable({
+    objectMode: true,
+    write(chunk, _enc, cb) {
+      log.push(`${label} write ${show(chunk)}`);
+      cb();
+    },
+    final(cb) { log.push(`${label} final`); cb(); },
+  });
+}
+function sink(label, objectMode = false) {
+  if (objectMode) return objectSink(label);
+  return new Writable({
+    write(chunk, _enc, cb) {
+      log.push(`${label} write ${typeof chunk === "string" ? chunk : Buffer.isBuffer(chunk) ? chunk.toString() : JSON.stringify(chunk)}`);
+      cb();
+    },
+    final(cb) { log.push(`${label} final`); cb(); },
+  });
+}
+function* gen() {
+  log.push("gen start");
+  yield "g1";
+  log.push("gen mid");
+  yield "g2";
+  log.push("gen end");
+}
+async function* agen() {
+  yield "a1";
+  await null;
+  yield "a2";
+}
+const sources = {
+  array: () => ["x", "y"],
+  string: () => "hey",
+  set: () => new Set(["s1", "s2"]),
+  gen: () => gen(),
+  agen: () => agen(),
+  iterable: () => ({ *[Symbol.iterator]() { yield "i1"; yield "i2"; } }),
+  objects: () => [{ n: 1 }, { n: 2 }],
+};
+const names = Object.keys(sources);
+async function run(name) {
+  const src = sources[name]();
+  const objectMode = name === "objects";
+  await new Promise((resolve) => {
+    queueMicrotask(() => log.push(`${name} microtask`));
+    pipeline(src, sink(name, objectMode), (err) => {
+      log.push(`${name} done ${err ? err.code ?? err.message : err}`);
+      resolve();
+    });
+    log.push(`${name} after call`);
+  });
+}
+for (const n of names) await run(n);
+
+// a transform chain and a function stage after an untyped source
+const src2 = ["p", "q"];
+await new Promise((resolve) =>
+  pipeline(
+    src2,
+    new Transform({ transform(c, _e, cb) { cb(null, c.toString().toUpperCase()); } }),
+    sink("transform"),
+    (err) => { log.push(`transform done ${err}`); resolve(); },
+  ),
+);
+const src3 = new Set(["f1", "f2"]);
+await new Promise((resolve) =>
+  pipeline(
+    src3,
+    async function* (source) { for await (const c of source) yield c + "!"; },
+    sink("fnstage"),
+    (err) => { log.push(`fnstage done ${err}`); resolve(); },
+  ),
+);
+// promises form
+const src4 = ["m", "n"];
+await pipelineP(src4, sink("promises"));
+log.push("promises resolved");
+// an error thrown by the source
+function* bad() { yield "ok"; throw new Error("source failed"); }
+const src5 = bad();
+await new Promise((resolve) =>
+  pipeline(src5, sink("bad"), (err) => { log.push(`bad done ${err && err.message}`); resolve(); }),
+);
+// a source of non-string values into a byte-mode writable
+const src6 = [1, 2];
+await new Promise((resolve) =>
+  pipeline(src6, sink("numbers"), (err) => { log.push(`numbers done ${err && err.code}`); resolve(); }),
+);
+console.log(log.join("\n"));

tests/corpus/7504-pipeline-untyped-source-errors.mjs

@@ -0,0 +1,90 @@
+// pipeline() over untyped iterable sources: a destination that fails closes
+// the source, backpressure waits for drain, an abort signal stops the pump,
+// and async generator and Map sources reach object-mode destinations.
+import { pipeline, Writable, PassThrough } from "node:stream";
+import { pipeline as pipelineP } from "node:stream/promises";
+
+function* counted(label, n) {
+  try {
+    for (let i = 0; i < n; i++) {
+      console.log(label, "yield", i);
+      yield `${label}${i}`;
+    }
+  } finally {
+    console.log(label, "finally");
+  }
+}
+function pick(v) {
+  return v;
+}
+
+// the destination fails on its second write: the source's finally runs
+await new Promise((resolve) => {
+  let n = 0;
+  const dst = new Writable({
+    write(chunk, _e, cb) {
+      n++;
+      console.log("fail write", String(chunk));
+      cb(n === 2 ? new Error("write failed") : null);
+    },
+  });
+  pipeline(pick(counted("f", 5)), dst, (err) => {
+    console.log("fail done", err && err.message);
+    resolve();
+  });
+});
+
+// backpressure: a one-byte highWaterMark and asynchronous writes
+await new Promise((resolve) => {
+  const dst = new Writable({
+    highWaterMark: 1,
+    write(chunk, _e, cb) {
+      console.log("slow write", String(chunk));
+      setTimeout(cb, 1);
+    },
+  });
+  pipeline(pick(counted("b", 3)), dst, (err) => {
+    console.log("slow done", err);
+    resolve();
+  });
+});
+
+// an abort signal
+const ac = new AbortController();
+try {
+  await pipelineP(
+    pick(counted("s", 10)),
+    new Writable({
+      write(chunk, _e, cb) {
+        console.log("signal write", String(chunk));
+        if (String(chunk) === "s1") ac.abort();
+        cb();
+      },
+    }),
+    { signal: ac.signal },
+  );
+} catch (err) {
+  console.log("signal rejected", err.name);
+}
+
+// an untyped async generator into a PassThrough read with for await
+const out = new PassThrough({ objectMode: true });
+const both = pipelineP(
+  pick((async function* () {
+    yield "u1";
+    yield "u2";
+  })()),
+  out,
+);
+for await (const c of out) console.log("passthrough", c);
+await both;
+console.log("passthrough resolved");
+
+// a Map source of entries into an object-mode sink
+await new Promise((resolve) => {
+  pipeline(
+    pick(new Map([["k", 1]])),
+    new Writable({ objectMode: true, write(c, _e, cb) { console.log("map write", JSON.stringify(c)); cb(); } }),
+    (err) => { console.log("map done", err); resolve(); },
+  );
+});

tests/corpus/7505-pipeline-untyped-sources.cjs

@@ -0,0 +1,40 @@
+// pipeline() over untyped sources in CommonJS: a source that a function
+// answers, which may be an array, a generator, a stream, a string, or a
+// Buffer, whose bytes are numbers that a byte-mode Writable refuses.
+const { pipeline, Readable, Writable } = require("node:stream");
+
+function sink(label) {
+  const parts = [];
+  return new Writable({
+    write(chunk, _enc, cb) {
+      parts.push(String(chunk));
+      cb();
+    },
+    final(cb) {
+      console.log(label, parts.join("|"));
+      cb();
+    },
+  });
+}
+
+function* letters() {
+  yield "a";
+  yield "b";
+}
+
+const sources = [
+  () => ["one", "two"],
+  () => letters(),
+  () => Readable.from(["r1", "r2"]),
+  () => "xyz",
+  () => Buffer.from("buf"),
+];
+
+function run(i) {
+  if (i === sources.length) return;
+  pipeline(sources[i](), sink(`source ${i}`), (err) => {
+    console.log(`source ${i} done`, err === undefined ? "ok" : err.code);
+    run(i + 1);
+  });
+}
+run(0);

tests/corpus/7506-pipeline-any-source.ts

@@ -0,0 +1,48 @@
+// pipeline() in TypeScript over sources typed any or unknown, and over a
+// Set, a Map and an async generator object, which are pumped as iterables.
+import { pipeline, Writable } from "node:stream";
+import { pipeline as pipelineP } from "node:stream/promises";
+
+function sink(label: string): Writable {
+  const parts: string[] = [];
+  return new Writable({
+    objectMode: true,
+    write(chunk: unknown, _enc, cb) {
+      parts.push(JSON.stringify(chunk));
+      cb();
+    },
+    final(cb) {
+      console.log(label, parts.join(" "));
+      cb();
+    },
+  });
+}
+
+async function* ticks(): AsyncGenerator<string> {
+  yield "t1";
+  await new Promise((r) => setTimeout(r, 1));
+  yield "t2";
+}
+
+async function main(): Promise<void> {
+  const anySource: any = ["any1", "any2"];
+  await pipelineP(anySource, sink("any"));
+  const unknownSource: unknown = new Set([1, 2, 3]);
+  await pipelineP(unknownSource as any, sink("unknown"));
+  await pipelineP(new Set(["s"]), sink("set"));
+  await pipelineP(
+    new Map([
+      ["a", 1],
+      ["b", 2],
+    ]),
+    sink("map"),
+  );
+  await pipelineP(ticks(), sink("async generator"));
+  await new Promise<void>((resolve) => {
+    pipeline(anySource, sink("callback"), (err?: Error | null) => {
+      console.log("callback done", err);
+      resolve();
+    });
+  });
+}
+void main();

tests/corpus/7507-stream-write-error-drain.mjs

@@ -0,0 +1,37 @@
+// A write whose callback fails synchronously: write() answers false and
+// writableNeedDrain stays false, so no 'drain' follows the error, and a
+// pipeline pumping a generator stops at the failed write.
+import { pipeline, Writable } from "node:stream";
+
+const w = new Writable({
+  write(_chunk, _enc, cb) {
+    cb(new Error("bad write"));
+  },
+});
+w.on("error", (e) => console.log("error event", e.message));
+w.on("drain", () => console.log("drain"));
+const r = w.write("a");
+console.log("returned", r, "needDrain", w.writableNeedDrain, "errored", w.errored.message);
+
+function* source() {
+  try {
+    for (let i = 0; i < 4; i++) {
+      console.log("yield", i);
+      yield `c${i}`;
+    }
+  } finally {
+    console.log("source finally");
+  }
+}
+let writes = 0;
+pipeline(
+  source(),
+  new Writable({
+    write(chunk, _enc, cb) {
+      writes++;
+      console.log("write", String(chunk));
+      cb(writes === 2 ? new Error("second write failed") : null);
+    },
+  }),
+  (err) => console.log("pipeline done", err.message),
+);

tests/harness/corpus-costs/7503-pipeline-untyped-sources.mjs.json

@@ -0,0 +1 @@
+{ "plain": 5499, "san": 7259 }

tests/harness/corpus-costs/7504-pipeline-untyped-source-errors.mjs.json

@@ -0,0 +1 @@
+{ "plain": 3431, "san": 4529 }

tests/harness/corpus-costs/7505-pipeline-untyped-sources.cjs.json

@@ -0,0 +1 @@
+{ "plain": 3197, "san": 4220 }

tests/harness/corpus-costs/7506-pipeline-any-source.ts.json

@@ -0,0 +1 @@
+{ "plain": 3250, "san": 4741 }

tests/harness/corpus-costs/7507-stream-write-error-drain.mjs.json

@@ -0,0 +1 @@
+{ "plain": 2735, "san": 4076 }

tests/harness/self-hosting-native-toolchain.test.ts

@@ -109,7 +109,7 @@ for (const backend of ["llvm"] as const) {
           object_format: target.objectFormat,
           minimum_os: target.minimumOs,
         },
-        runtime_abi: { version: 37, marker: "scr_runtime_abi_v37" },
+        runtime_abi: { version: 42, marker: "scr_runtime_abi_v42" },
         compiler: { command: "fixture", identity: "fixture", target: target.llvmTriple },
         macros: { executable: [], excluded: [], sanitizer: "external-toolchain-required" },
         flavors: {