Pass { signal } to stream operator callbacks and Duplex.from bodies

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

@@ -10,7 +10,7 @@
   The `objectMode` option, and the `readableObjectMode` and `writableObjectMode` options of a `Duplex`, `Transform`, or `PassThrough`, take `true` or `false` literals in an options object literal; options passed as a value are read when the stream is built.
   In typed code, a `write` or `transform` callback, or a `_write` or `_transform` method, may declare its chunk with any type that converts to a checked-dynamic value: the chunk converts when the callback runs, and a value of another type throws a TypeError there.
   An object-mode write passes its encoding argument to `write` and `transform` as Node does (`undefined` when absent, `null` when a callback took its place); a parameter declared `string` receives `""` for `undefined` and `null`, and one declared `string | null | undefined` receives Node's value.
-  `readable.read()` on an object-mode stream answers its Buffers; another value throws a not-supported error (read such values with `'data'` listeners, `for await`, or `pipe`).
+  `readable.read()` on an object-mode stream answers any value it holds.
   `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.
@@ -20,10 +20,11 @@
   `@types/node` types that result as `NodeJS.WritableStream`, so `for await` over it needs the `PassThrough` type (`pipeline(...) as PassThrough`, or a JSDoc `@type`).
   In typed code, the callback of a pipeline with an async function destination receives `null` instead of `undefined` when there is no error.
   `Readable.from` accepts `objectMode` and `highWaterMark` options.
-  The `Readable` operators (`map`, `filter`, `forEach`, `toArray`, `some`, `find`, `every`, `flatMap`, `drop`, `take`, and `reduce`) accept `concurrency`, `highWaterMark`, and `signal` options; their callbacks take only the chunk, `map` returns strings or Buffers, and `flatMap` returns strings or arrays of them.
+  The `Readable` operators (`map`, `filter`, `forEach`, `toArray`, `some`, `find`, `every`, `flatMap`, `drop`, `take`, and `reduce`) accept `concurrency`, `highWaterMark`, and `signal` options.
+  Their callbacks may take the `{ signal }` argument, `map` may return any value, and `flatMap` may return any iterable, async iterable, or stream.
   `Duplex.from` accepts an async generator function, which reads what is written to the Duplex, a `Duplex`, `Transform` or `PassThrough`, which it returns, and the values `Readable.from` accepts, which give a Duplex that is not writable.
   It also accepts a lone `Readable` or `Writable` and a `{ readable, writable }` pair, which the Duplex reads and writes through, a promise, whose value it reads, and an async function, which reads what is written and must resolve to `null` or `undefined`.
-  A web stream and a body function that declares the options argument are unsupported in `Duplex.from`.
+  A `Duplex.from` body function may take the `{ signal }` argument, which is aborted with the error the Duplex is destroyed with.
   `stream.compose` accepts streams and async generator functions in any position, an async function as the last stage, and an array, a string, or a generator as the first stage; one stage is `Duplex.from` of it.
   A web `ReadableStream` is read as an async iterable by `Readable.from` and as the first stage of `compose` or `pipeline`, and through its reader by `Readable.fromWeb` (with `objectMode`, `highWaterMark`, `encoding`, and `signal` in an options object literal) and `Duplex.from`.
   A web stream in a later compose stage, `Readable.toWeb`, and web `WritableStream` and `TransformStream` interop are unsupported.

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: 31, marker: "scr_runtime_abi_v31" },
+      runtime_abi: { version: 32, marker: "scr_runtime_abi_v32" },
       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_v31");
+    expect(error?.stderr).toContain("scr_runtime_abi_v32");
   });
 });

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

@@ -7272,7 +7272,9 @@ declare module "stream" {
     push(chunk: any, encoding?: string): boolean;
     unshift(chunk: Buffer | string, encoding?: string): void;
     unshift(chunk: any, encoding?: string): void;
-    read(size?: number): Buffer | null;
+    /* A Buffer, a string once encoded, an object-mode value, or null (as
+     * @types/node types it). */
+    read(size?: number): any;
     pause(): this;
     resume(): this;
     isPaused(): boolean;

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

@@ -1134,6 +1134,8 @@ export const LIB_FN_SYMS: Record<string, string> = {
   "stream.duplexifyPair": "scr_stream_duplexify_pair",
   "stream.duplexFromPromise": "scr_stream_duplex_from_promise",
   "stream.duplexFromAsyncFn": "scr_stream_duplex_from_async_fn",
+  "stream.duplexFromGenSignal": "scr_stream_duplex_from_gen_signal",
+  "stream.duplexFromAsyncFnSignal": "scr_stream_duplex_from_async_fn_signal",
   "fetch.readableFromWeb": "scr_stream_readable_from_web",
   "writable.writeEncoding": "scr_stream_write_encoding",
   "writable.setDefaultEncoding": "scr_stream_set_default_encoding",
@@ -1200,6 +1202,8 @@ export const LIB_FN_SYMS: Record<string, string> = {
   "readable.opMap": "scr_stream_op_map",
   "readable.opFilter": "scr_stream_op_filter",
   "readable.opFlatMap": "scr_stream_op_flat_map",
+  "readable.opFlatMapAsyncGen": "scr_stream_op_flat_map_async_gen",
+  "readable.opFlatMapSyncGen": "scr_stream_op_flat_map_sync_gen",
   "readable.opDrop": "scr_stream_op_drop",
   "readable.opTake": "scr_stream_op_take",
   "readable.opToArray": "scr_stream_op_to_array",
@@ -1208,6 +1212,11 @@ export const LIB_FN_SYMS: Record<string, string> = {
   "readable.opEvery": "scr_stream_op_every",
   "readable.opFind": "scr_stream_op_find",
   "readable.opReduce": "scr_stream_op_reduce",
+  "readable.opSignalList": "scr_stream_op_signal_list",
+  "readable.opReduceSignal": "scr_stream_op_reduce_signal",
+  "readable.opSignalOf": "scr_stream_op_signal_of",
+  "readable.opOnceOptions": "scr_stream_op_once_options",
+  "readable.opSignalOpts": "scr_stream_op_signal_opts",
   "readable.pipe": "scr_stream_pipe",
   "readable.pipeGeneral": "scr_stream_pipe_general",
   "writable.cork": "scr_stream_cork",
@@ -1711,6 +1720,8 @@ export const USES_TIMERS_LIB_FNS = new Set<string>([
   "readable.opMap",
   "readable.opFilter",
   "readable.opFlatMap",
+  "readable.opFlatMapAsyncGen",
+  "readable.opFlatMapSyncGen",
   "readable.opDrop",
   "readable.opTake",
   "readable.opToArray",
@@ -1719,6 +1730,7 @@ export const USES_TIMERS_LIB_FNS = new Set<string>([
   "readable.opEvery",
   "readable.opFind",
   "readable.opReduce",
+  "readable.opReduceSignal",
   "readable.resume",
   "readable.pipe",
   "readable.pipeGeneral",
@@ -1748,6 +1760,8 @@ export const USES_TIMERS_LIB_FNS = new Set<string>([
   "stream.duplexifyPair",
   "stream.duplexFromPromise",
   "stream.duplexFromAsyncFn",
+  "stream.duplexFromGenSignal",
+  "stream.duplexFromAsyncFnSignal",
   "fetch.readableFromWeb",
   "stream.setWritev",
   "stream.setWritevDefault",

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 = 31 as const;
-export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v31" as const;
+export const RUNTIME_ABI_VERSION = 32 as const;
+export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v32" 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: 31, marker: "scr_runtime_abi_v31" },
+    runtime_abi: { version: 32, marker: "scr_runtime_abi_v32" },
     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

@@ -127,7 +127,7 @@ async function fixture(target: NativeTargetSpec = MACOS_ARM64_TARGET) {
       object_format: target.objectFormat,
       minimum_os: target.minimumOs,
     },
-    runtime_abi: { version: 31, marker: "scr_runtime_abi_v31" },
+    runtime_abi: { version: 32, marker: "scr_runtime_abi_v32" },
     compiler: {
       command: "clang",
       identity: "fixture clang",

packages/compiler/src/frontend/lowering/class-dynamic-dispatch.ts

@@ -85,7 +85,7 @@ import {
   lowerDynamicEmitterOn,
 } from "./lower-event-emitter.js";
 import { protocolArm, protocolFieldArm } from "./class-protocol-dispatch.js";
-import { STREAM_OPERATORS } from "./lower-stream.js";
+import { STREAM_OPERATORS, operatorSignalOptions, reduceSignalSetup } from "./lower-stream.js";
 import { deletedPropertyNames, fixedFieldFamily } from "./js-class-fields.js";
 
 const STREAM_METHODS = new Set([
@@ -5640,13 +5640,16 @@ export class ClassDynamicDispatch {
         const readableT: IrType = { kind: "object", className: "%Readable" };
         const op = (fn: string, args: IrExpr[], type: IrType): IrStmt[] =>
           ret({ kind: "libCall", fn: fn as IrLibFn, args: [receiver, ...args], type, loc });
+        // Node passes every callback its { signal }; the callback's
+        // declared parameters are not known here.
+        const sig = (): IrExpr => operatorSignalOptions(lowerer, arg(1), loc);
         switch (method) {
           case "map":
-            return op("readable.opMap", [arg(0), arg(1)], readableT);
+            return op("readable.opMap", [arg(0), arg(1), sig()], readableT);
           case "filter":
-            return op("readable.opFilter", [arg(0), arg(1)], readableT);
+            return op("readable.opFilter", [arg(0), arg(1), sig()], readableT);
           case "flatMap":
-            return op("readable.opFlatMap", [arg(0), arg(1)], readableT);
+            return op("readable.opFlatMap", [arg(0), arg(1), sig()], readableT);
           case "drop":
             return op("readable.opDrop", [arg(0), arg(1)], readableT);
           case "take":
@@ -5654,25 +5657,49 @@ export class ClassDynamicDispatch {
           case "toArray":
             return op("readable.opToArray", [arg(0)], { kind: "promise", inner: DYN });
           case "forEach":
-            return op("readable.opForEach", [arg(0), arg(1)], { kind: "promise", inner: VOID });
+            return op("readable.opForEach", [arg(0), arg(1), sig()], {
+              kind: "promise",
+              inner: VOID,
+            });
           case "some":
-            return op("readable.opSome", [arg(0), arg(1)], { kind: "promise", inner: BOOL });
+            return op("readable.opSome", [arg(0), arg(1), sig()], { kind: "promise", inner: BOOL });
           case "every":
-            return op("readable.opEvery", [arg(0), arg(1)], { kind: "promise", inner: BOOL });
+            return op("readable.opEvery", [arg(0), arg(1), sig()], {
+              kind: "promise",
+              inner: BOOL,
+            });
           case "find":
-            return op("readable.opFind", [arg(0), arg(1)], { kind: "promise", inner: DYN });
-          case "reduce":
-            return op(
-              "readable.opReduce",
-              [
-                arg(0),
-                arg(1),
-                { kind: "boolLit", value: dispatch.source.args.length >= 2, type: BOOL, loc },
-                arg(2),
-                { kind: "numLit", value: 0, type: F64, loc },
-              ],
-              { kind: "promise", inner: DYN },
-            );
+            return op("readable.opFind", [arg(0), arg(1), sig()], { kind: "promise", inner: DYN });
+          case "reduce": {
+            // the dispatch function's locals: there is no function
+            // context while dispatch bodies are built
+            const setup = reduceSignalSetup(lowerer, arg(2), loc, (name, type) => {
+              const local = {
+                id: `${name}.${dispatch.fn.locals.length}`,
+                name,
+                type,
+                mutable: false,
+              };
+              dispatch.fn.locals.push(local);
+              return local;
+            });
+            return [
+              ...setup.stmts,
+              ...op(
+                "readable.opReduceSignal",
+                [
+                  arg(0),
+                  arg(1),
+                  { kind: "boolLit", value: dispatch.source.args.length >= 2, type: BOOL, loc },
+                  arg(2),
+                  { kind: "numLit", value: 0, type: F64, loc },
+                  setup.abort,
+                  setup.cbOpts,
+                ],
+                { kind: "promise", inner: DYN },
+              ),
+            ];
+          }
         }
       }
       if (method === "destroy" && (!incoming.length || incoming[0]?.kind === "unitLit"))

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

@@ -3923,21 +3923,27 @@ function pipelineBoxHelper(
 
 /** `() => ac.abort()` over the pipeline's AbortController: an interned
  * lifted function takes the controller and answers the closure. */
-function pipelineAbortHelper(lowerer: Lowerer, ac: IrExpr, loc: SrcLoc): IrExpr {
-  const key = "pipelineAbort";
+function pipelineAbortHelper(
+  lowerer: Lowerer,
+  ac: IrExpr,
+  loc: SrcLoc,
+  withReason = false,
+): IrExpr {
+  const key = withReason ? "pipelineAbortReason" : "pipelineAbort";
   let name = lowerer.valueHelpers.get(key);
-  const abortT = funcOf([], VOID);
+  const abortT = funcOf(withReason ? [DYN] : [], VOID);
   if (!name) {
     name = keyedHelperName(lowerer, "%stream.pipelineAbort", key);
     lowerer.valueHelpers.set(key, name);
     const acLocal: IrLocal = { id: "ac.0", name: "ac", type: DYN, mutable: false, boxed: true };
     const acRef = varRef(acLocal.id, DYN, loc);
+    const reasonLocal: IrLocal = { id: "reason.0", name: "reason", type: DYN, mutable: false };
     lowerer.liftedFns.push({
       name: `${name}.abort`,
-      params: [],
+      params: withReason ? [{ localId: reasonLocal.id, name: reasonLocal.name, type: DYN }] : [],
       returnType: VOID,
       captures: [{ localId: acLocal.id, name: acLocal.name, type: DYN }],
-      locals: [acLocal],
+      locals: withReason ? [acLocal, reasonLocal] : [acLocal],
       body: [
         {
           kind: "exprStmt",
@@ -3946,7 +3952,7 @@ function pipelineAbortHelper(lowerer: Lowerer, ac: IrExpr, loc: SrcLoc): IrExpr
             recv: acRef,
             method: "abort",
             calleeName: "ac.abort",
-            args: [],
+            args: withReason ? [varRef(reasonLocal.id, DYN, loc)] : [],
             type: DYN,
             loc,
           },
@@ -4165,11 +4171,11 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE
         "write the body as async function* (source) { ... } or async function (source) { ... }",
       );
     }
-    if (fn.parameters.nodes.length > 1) {
+    if (fn.parameters.nodes.length > 2) {
       lowerer.noLowering(
-        "a Duplex.from body that declares the options argument",
-        fn.parameters.nodes[1]!,
-        "the body takes only its source",
+        "a Duplex.from body with parameters after the options argument",
+        fn.parameters.nodes[2]!,
+        "the body takes the source and { signal }",
       );
     }
     const sourceT: SyncGenType = {
@@ -4180,6 +4186,46 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE
       nextT: UNDEFINED_T,
     };
     const lowered = lowerStageFunction(lowerer, fn, sourceT);
+    const params = lowered.closure.type.kind === "func" ? lowered.closure.type.params : [];
+    // fromAsyncGen calls the body with { signal } of an AbortController
+    // that destroy aborts with its error.
+    let signalArgs: IrExpr[] = [];
+    let signalDecls: IrStmt[] = [];
+    if (params.length > 1) {
+      if (params[1]!.kind !== "dyn") {
+        lowerer.noLowering(
+          "a Duplex.from body's options argument with this type",
+          fn.parameters.nodes[1]!,
+          "type it as { signal: AbortSignal }",
+        );
+      }
+      const acLocal = lowerer.declareHiddenLocal("%duplexAbort", DYN);
+      const ac = varRef(acLocal.id, DYN, loc);
+      signalDecls = [
+        {
+          kind: "varDecl",
+          localId: acLocal.id,
+          init: { kind: "libCall", fn: "fetch.abortControllerNew", args: [], type: DYN, loc },
+          loc,
+        },
+      ];
+      signalArgs = [
+        pipelineAbortHelper(lowerer, ac, loc, true),
+        { kind: "dynKeyGet", value: ac, key: strLit("signal", loc), type: DYN, loc },
+      ];
+    }
+    const withSignal = (fnName: IrLibFn, args: IrExpr[]): IrExpr => {
+      const call: IrExpr = {
+        kind: "libCall",
+        fn: signalArgs.length > 0 ? (`${fnName}Signal` as IrLibFn) : fnName,
+        args: [...args, ...signalArgs],
+        type: duplex,
+        loc,
+      };
+      return signalDecls.length > 0
+        ? { kind: "seqExpr", stmts: signalDecls, result: call, type: duplex, loc }
+        : call;
+    };
     if (!isGen) {
       // an async function: a writable Duplexify whose final waits for
       // the function's promise
@@ -4200,26 +4246,18 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE
           node,
         );
       }
-      return {
-        kind: "libCall",
-        fn: "stream.duplexFromAsyncFn",
-        args: [lowered.closure, asyncSource, pipelineBoxHelper(lowerer, out, loc)],
-        type: duplex,
-        loc,
-      };
+      return withSignal("stream.duplexFromAsyncFn", [
+        lowered.closure,
+        asyncSource,
+        pipelineBoxHelper(lowerer, out, loc),
+      ]);
     }
     const helpers =
       lowered.out.kind === "generator" ? fromGenHelpers(lowerer, lowered.out, loc) : null;
     const source = duplexSourceHelper(lowerer, sourceT, loc);
     if (!helpers || !source)
       lowerer.noLowering("Duplex.from over a function with this generator type", node);
-    return {
-      kind: "libCall",
-      fn: "stream.duplexFromGen",
-      args: [lowered.closure, source!, ...helpers!],
-      type: duplex,
-      loc,
-    };
+    return withSignal("stream.duplexFromGen", [lowered.closure, source!, ...helpers!]);
   }
   // duplexify.js: a stream answers itself (a Duplex) or is wrapped; any
   // other body reads as from(Duplexify, body, { objectMode: true,
@@ -4510,39 +4548,38 @@ export const STREAM_OPERATORS: ReadonlySet<string> = new Set([
   "reduce",
 ]);
 
-/** An operator callback as a checked-dynamic function value. Callbacks
- * that declare the options argument ({ signal }) are refused: the
- * AbortSignal Node passes there has no representation. map and flatMap
- * callbacks must produce strings or Buffers (flatMap: arrays of them, or
- * strings) — or checked-dynamic values, which the runtime checks. */
+/** An operator callback as a checked-dynamic function value, and whether
+ * it declares the options argument ({ signal }), which the runtime then
+ * passes. Its results convert to checked-dynamic values (flatMap's are
+ * iterated with yield*). */
 function lowerOperatorCallback(
   lowerer: Lowerer,
   node: ts.Expression,
   member: string,
   maxParams: number,
-): IrExpr {
+): { fn: IrExpr; takesOptions: boolean; gen?: IrType & { kind: "generator" } } {
   const loc = locOf(node);
   const optionsParam = (at: ts.Node): never =>
     lowerer.noLowering(
-      `${member} callbacks that declare the options argument`,
+      `${member} callbacks with parameters after the options argument`,
       at,
-      "the { signal } argument has no lowering — take only the chunk",
+      "the callback takes the chunk and { signal }",
     );
-  if (
-    (ts.isArrowFunction(node) || ts.isFunctionExpression(node)) &&
-    node.parameters.nodes.length > maxParams
-  ) {
-    optionsParam(node.parameters.nodes[maxParams]!);
+  let takesOptions = false;
+  if (ts.isArrowFunction(node) || ts.isFunctionExpression(node)) {
+    if (node.parameters.nodes.length > maxParams + 1)
+      optionsParam(node.parameters.nodes[maxParams + 1]!);
+    takesOptions = node.parameters.nodes.length > maxParams;
   }
   const v = lowerer.lowerExpr(node);
-  if (v.type.kind === "dyn") return v;
+  if (v.type.kind === "dyn") return { fn: v, takesOptions };
   if (
     v.type.kind !== "func" &&
     isJsSourceFile(ts.getSourceFileOfNode(node)) &&
     lowerer.dynConvertible(v.type)
   ) {
     // JavaScript passes any value; Node's validateFunction throws.
-    return { kind: "dynFrom", value: v, type: DYN, loc };
+    return { fn: { kind: "dynFrom", value: v, type: DYN, loc }, takesOptions: false };
   }
   if (v.type.kind !== "func") {
     lowerer.noLowering(
@@ -4551,25 +4588,8 @@ function lowerOperatorCallback(
       "pass a function",
     );
   }
-  if (v.type.params.length > maxParams) optionsParam(node);
-  if (member === "map" || member === "flatMap") {
-    const ret = v.type.ret.kind === "promise" ? v.type.ret.inner : v.type.ret;
-    const chunk = (t: IrType): boolean =>
-      t.kind === "string" || t.kind === "dyn" || (t.kind === "bytes" && t.elem === "u8");
-    const ok =
-      member === "map"
-        ? chunk(ret)
-        : ret.kind === "dyn" || ret.kind === "string" || (ret.kind === "array" && chunk(ret.elem));
-    if (!ok) {
-      lowerer.noLowering(
-        `${member} callbacks that return '${lowerer.fmt(ret)}'`,
-        node,
-        member === "map"
-          ? "map results are stream chunks: strings or Buffers (object-mode streams of other values are not lowered)"
-          : "flatMap results are arrays of strings or Buffers, or strings",
-      );
-    }
-  }
+  if (v.type.params.length > maxParams + 1) optionsParam(node);
+  if (v.type.params.length > maxParams) takesOptions = true;
   const getRecord = (id: string) => lowerer.shapes.get(id);
   const getUnion = (id: string) => lowerer.unions.get(id);
   if (!canBoxFuncIntoDyn(v.type, getRecord, getUnion)) {
@@ -4579,7 +4599,95 @@ function lowerOperatorCallback(
       "the callback's parameter and result types must cross the checked-dynamic boundary",
     );
   }
-  return { kind: "dynFrom", value: v, type: DYN, loc };
+  // flatMap over a compiled generator: the runtime steps it through the
+  // helpers over its type
+  const gen = member === "flatMap" && v.type.ret.kind === "generator" ? v.type.ret : undefined;
+  return { fn: { kind: "dynFrom", value: v, type: DYN, loc }, takesOptions, ...(gen && { gen }) };
+}
+
+/** The operator callbacks' `{ signal }` (operators.js): map and the
+ * operators over it pass `{ signal: AbortSignal.any([options?.signal]
+ * .filter(Boolean)) }`. options is the operator's options value. */
+export function operatorSignalOptions(lowerer: Lowerer, options: IrExpr, loc: SrcLoc): IrExpr {
+  void lowerer;
+  const list: IrExpr = {
+    kind: "libCall",
+    fn: "readable.opSignalList",
+    args: [options],
+    type: DYN,
+    loc,
+  };
+  const signal: IrExpr = { kind: "libCall", fn: "fetch.abortAny", args: [list], type: DYN, loc };
+  return { kind: "libCall", fn: "readable.opSignalOpts", args: [signal], type: DYN, loc };
+}
+
+/** reduce's signal for a reducer that reads it (operators.js): `const ac
+ * = new AbortController()`, `options.signal.addEventListener('abort', ()
+ * => ac.abort(), { once: true })`, then reduceSignal's ac.abort() closure
+ * and `{ signal: ac.signal }`. options is the bound options value. */
+export function reduceSignalSetup(
+  lowerer: Lowerer,
+  options: IrExpr,
+  loc: SrcLoc,
+  declare: (name: string, type: IrType) => IrLocal = (name, type) =>
+    lowerer.declareHiddenLocal(name, type),
+): { stmts: IrStmt[]; abort: IrExpr; cbOpts: IrExpr } {
+  const acLocal = declare("%opAbort", DYN);
+  const ac = varRef(acLocal.id, DYN, loc);
+  const sigLocal = declare("%opUserSignal", DYN);
+  const sig = varRef(sigLocal.id, DYN, loc);
+  const abort = pipelineAbortHelper(lowerer, ac, loc);
+  const stmts: IrStmt[] = [
+    {
+      kind: "varDecl",
+      localId: acLocal.id,
+      init: { kind: "libCall", fn: "fetch.abortControllerNew", args: [], type: DYN, loc },
+      loc,
+    },
+    {
+      kind: "varDecl",
+      localId: sigLocal.id,
+      init: { kind: "libCall", fn: "readable.opSignalOf", args: [options], type: DYN, loc },
+      loc,
+    },
+    {
+      kind: "if",
+      cond: { kind: "dynTest", test: "truthy", value: sig, type: BOOL, loc },
+      then: [
+        {
+          kind: "exprStmt",
+          expr: {
+            kind: "dynInvoke",
+            recv: sig,
+            method: "addEventListener",
+            calleeName: "options.signal.addEventListener",
+            args: [
+              { kind: "dynFrom", value: strLit("abort", loc), type: DYN, loc },
+              { kind: "dynFrom", value: pipelineAbortHelper(lowerer, ac, loc), type: DYN, loc },
+              { kind: "libCall", fn: "readable.opOnceOptions", args: [], type: DYN, loc },
+            ],
+            type: DYN,
+            loc,
+          },
+          loc,
+        },
+      ],
+      else_: null,
+      loc,
+    },
+  ];
+  const signal: IrExpr = {
+    kind: "dynKeyGet",
+    value: ac,
+    key: strLit("signal", loc),
+    type: DYN,
+    loc,
+  };
+  return {
+    stmts,
+    abort,
+    cbOpts: { kind: "libCall", fn: "readable.opSignalOpts", args: [signal], type: DYN, loc },
+  };
 }
 
 /** An operator's options (or reduce's initial value): an object literal
@@ -4618,21 +4726,37 @@ function lowerStreamOperator(
 ): IrExpr {
   const loc = locOf(call);
   const args = call.arguments.nodes;
-  const recv = streamReceiverExpr(lowerer, access.expression);
+  // JS order: the receiver, the callback, then the options; the
+  // callback's { signal } is built over the options value afterwards.
+  const decls: IrStmt[] = [];
+  const bind = (value: IrExpr, name: string): IrExpr => {
+    const local = lowerer.declareHiddenLocal(name, value.type);
+    decls.push({ kind: "varDecl", localId: local.id, init: value, loc });
+    return varRef(local.id, value.type, loc);
+  };
+  const recv = bind(streamReceiverExpr(lowerer, access.expression), "%opRecv");
   const readable: IrType = { kind: "object", className: "%Readable" };
   const opts = (i: number): IrExpr =>
     lowerOperatorValue(lowerer, args[i], `${member} options`, loc, true);
+  let takesOptions = false;
+  let callbackGen: (IrType & { kind: "generator" }) | undefined;
   const fnArg = (maxParams: number): IrExpr => {
     if (args[0] === undefined) return dynUndefinedExpr(loc);
-    return lowerOperatorCallback(lowerer, args[0], member, maxParams);
+    const lowered = lowerOperatorCallback(lowerer, args[0], member, maxParams);
+    takesOptions = lowered.takesOptions;
+    callbackGen = lowered.gen;
+    return bind(lowered.fn, "%opFn");
+  };
+  const lib = (fn: string, a: IrExpr[], type: IrType): IrExpr => {
+    const call: IrExpr = { kind: "libCall", fn: fn as IrLibFn, args: a, type, loc };
+    return { kind: "seqExpr", stmts: decls, result: call, type, loc };
+  };
+  // the options argument (bound), and the callback's { signal } when it
+  // declares one
+  const withSignal = (optsValue: IrExpr): IrExpr[] => {
+    const o = bind(optsValue, "%opOptions");
+    return [o, takesOptions ? operatorSignalOptions(lowerer, o, loc) : dynUndefinedExpr(loc)];
   };
-  const lib = (fn: string, a: IrExpr[], type: IrType): IrExpr => ({
-    kind: "libCall",
-    fn: fn as IrLibFn,
-    args: a,
-    type,
-    loc,
-  });
   switch (member) {
     case "map":
     case "filter":
@@ -4644,7 +4768,16 @@ function lowerStreamOperator(
           : member === "filter"
             ? "readable.opFilter"
             : "readable.opFlatMap";
-      return lib(fnName, [recv, fn, opts(1)], readable);
+      const rest = withSignal(opts(1));
+      const genHelpers = callbackGen ? fromGenHelpers(lowerer, callbackGen, loc) : null;
+      if (callbackGen && genHelpers) {
+        return lib(
+          callbackGen.async ? "readable.opFlatMapAsyncGen" : "readable.opFlatMapSyncGen",
+          [recv, fn, ...rest, ...genHelpers],
+          readable,
+        );
+      }
+      return lib(fnName, [recv, fn, ...rest], readable);
     }
     case "drop":
     case "take": {
@@ -4657,15 +4790,24 @@ function lowerStreamOperator(
     }
     case "toArray":
       return lib("readable.opToArray", [recv, opts(0)], { kind: "promise", inner: DYN });
-    case "forEach":
-      return lib("readable.opForEach", [recv, fnArg(1), opts(1)], { kind: "promise", inner: VOID });
+    case "forEach": {
+      const fn = fnArg(1);
+      return lib("readable.opForEach", [recv, fn, ...withSignal(opts(1))], {
+        kind: "promise",
+        inner: VOID,
+      });
+    }
     case "some":
     case "every": {
       const fn = fnArg(1);
-      return lib(member === "some" ? "readable.opSome" : "readable.opEvery", [recv, fn, opts(1)], {
-        kind: "promise",
-        inner: BOOL,
-      });
+      return lib(
+        member === "some" ? "readable.opSome" : "readable.opEvery",
+        [recv, fn, ...withSignal(opts(1))],
+        {
+          kind: "promise",
+          inner: BOOL,
+        },
+      );
     }
     case "find": {
       const t = lowerer.typeOf(call);
@@ -4677,7 +4819,11 @@ function lowerStreamOperator(
           "the result is a checked-dynamic value — use a boolean predicate and narrow the result",
         );
       }
-      return lib("readable.opFind", [recv, fnArg(1), opts(1)], { kind: "promise", inner: DYN });
+      const fn = fnArg(1);
+      return lib("readable.opFind", [recv, fn, ...withSignal(opts(1))], {
+        kind: "promise",
+        inner: DYN,
+      });
     }
     case "reduce": {
       const t = lowerer.typeOf(call);
@@ -4704,12 +4850,19 @@ function lowerStreamOperator(
         }
       }
       const fn = fnArg(2);
-      const init = lowerOperatorValue(lowerer, args[1], "reduce's initial value", loc, false);
-      return lib(
-        "readable.opReduce",
-        [recv, fn, init, boolLit(args.length >= 2, loc), opts(2), numLit(kind, loc)],
-        { kind: "promise", inner },
+      const init = bind(
+        lowerOperatorValue(lowerer, args[1], "reduce's initial value", loc, false),
+        "%opInit",
       );
+      const o = bind(opts(2), "%opOptions");
+      const head = [recv, fn, init, boolLit(args.length >= 2, loc), o, numLit(kind, loc)];
+      if (!takesOptions) return lib("readable.opReduce", head, { kind: "promise", inner });
+      const setup = reduceSignalSetup(lowerer, o, loc);
+      decls.push(...setup.stmts);
+      return lib("readable.opReduceSignal", [...head, setup.abort, setup.cbOpts], {
+        kind: "promise",
+        inner,
+      });
     }
   }
   throw new InternalCompilerError(`unknown stream operator ${member}`);
@@ -5253,6 +5406,22 @@ export function lowerStreamMethodCall(
     requireSide(readableSide, "read");
     if (args.length > 1) lowerer.noLowering(`read with ${args.length} arguments`, call);
     const receiver = streamReceiverExpr(lowerer, access.expression);
+    // JavaScript (and a result the declarations type as any) reads any
+    // chunk: a Buffer, a string once encoded, or an object-mode value.
+    const resultTs = lowerer.typeOf(call);
+    if (
+      isJsSourceFile(ts.getSourceFileOfNode(call)) ||
+      (resultTs.flags & (ts.TypeFlags.Any | ts.TypeFlags.Unknown)) !== 0
+    ) {
+      const sizeDyn = args[0] ? lowerer.lowerExprExpecting(args[0], DYN) : dynUndefinedExpr(loc);
+      return {
+        kind: "libCall",
+        fn: "readable.readDyn",
+        args: [receiver, sizeDyn],
+        type: DYN,
+        loc,
+      };
+    }
     const size = args[0] ? lowerer.lowerExprExpecting(args[0], F64) : numLit(-1, loc);
     const type: IrType = unionOf(lowerer, [BYTES, { kind: "nullT" }]);
     const read: IrExpr = {

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

@@ -386,6 +386,8 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "stream.duplexFromDyn",
   "stream.duplexFromPromise",
   "stream.duplexFromAsyncFn",
+  "stream.duplexFromGenSignal",
+  "stream.duplexFromAsyncFnSignal",
   "fetch.readableFromWeb",
   "stream.destroy",
   "stream.destroyErr",
@@ -521,6 +523,8 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "readable.opMap",
   "readable.opFilter",
   "readable.opFlatMap",
+  "readable.opFlatMapAsyncGen",
+  "readable.opFlatMapSyncGen",
   "readable.opDrop",
   "readable.opTake",
   "rl.createSignal",

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

@@ -4082,6 +4082,13 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
   /** Duplex.from(asyncFunction): the stage, its source helper and the
    * promise's box helper. */
   "stream.duplexFromAsyncFn": { argTypes: [null, null, null], result: VOID },
+  /** The same for a body that declares the options argument: then the
+   * AbortController's abort() and its signal. */
+  "stream.duplexFromGenSignal": {
+    argTypes: [null, null, null, null, null, null, null, DYN],
+    result: VOID,
+  },
+  "stream.duplexFromAsyncFnSignal": { argTypes: [null, null, null, null, DYN], result: VOID },
   "writable.writeEncoding": { argTypes: [null, STRING], result: VOID },
   "writable.setDefaultEncoding": { argTypes: [null, DYN], result: VOID },
   /** Whether a readable.nextChunkDyn step answered the end (the done
@@ -4119,17 +4126,39 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
    * forEach nothing; reduce: [stream, fn, init, hasInit, options, kind]
    * with the payload kind (0 dyn, 1 number, 2 boolean, 3 string). The
    * promise-returning ones reject instead of throwing. */
-  "readable.opMap": { argTypes: [null, DYN, DYN], result: VOID },
-  "readable.opFilter": { argTypes: [null, DYN, DYN], result: VOID },
-  "readable.opFlatMap": { argTypes: [null, DYN, DYN], result: VOID },
+  "readable.opMap": { argTypes: [null, DYN, DYN, DYN], result: VOID },
+  "readable.opFilter": { argTypes: [null, DYN, DYN, DYN], result: VOID },
+  "readable.opFlatMap": { argTypes: [null, DYN, DYN, DYN], result: VOID },
+  /** flatMap whose callback returns a compiled generator: the helpers
+   * over its type follow (async: next, read, return, throw; sync: next,
+   * return, throw). */
+  "readable.opFlatMapAsyncGen": {
+    argTypes: [null, DYN, DYN, DYN, null, null, null, null],
+    result: VOID,
+  },
+  "readable.opFlatMapSyncGen": { argTypes: [null, DYN, DYN, DYN, null, null, null], result: VOID },
   "readable.opDrop": { argTypes: [null, DYN, DYN], result: VOID },
   "readable.opTake": { argTypes: [null, DYN, DYN], result: VOID },
   "readable.opToArray": { argTypes: [null, DYN], result: { kind: "promise", inner: DYN } },
-  "readable.opForEach": { argTypes: [null, DYN, DYN], result: { kind: "promise", inner: VOID } },
-  "readable.opSome": { argTypes: [null, DYN, DYN], result: { kind: "promise", inner: BOOL } },
-  "readable.opEvery": { argTypes: [null, DYN, DYN], result: { kind: "promise", inner: BOOL } },
-  "readable.opFind": { argTypes: [null, DYN, DYN], result: { kind: "promise", inner: DYN } },
+  "readable.opForEach": {
+    argTypes: [null, DYN, DYN, DYN],
+    result: { kind: "promise", inner: VOID },
+  },
+  "readable.opSome": { argTypes: [null, DYN, DYN, DYN], result: { kind: "promise", inner: BOOL } },
+  "readable.opEvery": { argTypes: [null, DYN, DYN, DYN], result: { kind: "promise", inner: BOOL } },
+  "readable.opFind": { argTypes: [null, DYN, DYN, DYN], result: { kind: "promise", inner: DYN } },
+  /** The map family's last argument is the callback's { signal } (or
+   * undefined). reduceSignal adds ac.abort() (a closure) and
+   * { signal: ac.signal } for a reducer that reads them. */
   "readable.opReduce": { argTypes: [null, DYN, DYN, BOOL, DYN, F64], result: VOID },
+  "readable.opReduceSignal": {
+    argTypes: [null, DYN, DYN, BOOL, DYN, F64, null, DYN],
+    result: VOID,
+  },
+  "readable.opSignalList": { argTypes: [DYN], result: DYN },
+  "readable.opSignalOpts": { argTypes: [DYN], result: DYN },
+  "readable.opSignalOf": { argTypes: [DYN], result: DYN },
+  "readable.opOnceOptions": { argTypes: [], result: DYN },
   "readable.resume": { argTypes: [null], result: VOID },
   "readable.isPaused": { argTypes: [null], result: BOOL },
   "readable.pipe": { argTypes: [null, null, BOOL], result: VOID },

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

@@ -223,6 +223,8 @@ const STREAM_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "stream.duplexFromDyn",
   "stream.duplexFromPromise",
   "stream.duplexFromAsyncFn",
+  "stream.duplexFromGenSignal",
+  "stream.duplexFromAsyncFnSignal",
   "fetch.readableFromWeb",
   "stream.duplexifyReadable",
   "stream.duplexifyWritable",
@@ -285,6 +287,8 @@ const READABLE_DYN_ASYNC_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "stream.duplexFromDyn",
   "stream.duplexFromPromise",
   "stream.duplexFromAsyncFn",
+  "stream.duplexFromGenSignal",
+  "stream.duplexFromAsyncFnSignal",
 ]);
 
 /** libCall name prefixes of the Readable entry points that await through the dyn-async unit. */

packages/compiler/src/ir/validate.ts

@@ -5520,6 +5520,8 @@ function validateFunction(
       e.fn === "stream.duplexifyPair" ||
       e.fn === "stream.duplexFromPromise" ||
       e.fn === "stream.duplexFromAsyncFn" ||
+      e.fn === "stream.duplexFromGenSignal" ||
+      e.fn === "stream.duplexFromAsyncFnSignal" ||
       e.fn === "stream.setConstruct" ||
       e.fn === "stream.setWritev" ||
       e.fn === "stream.setWritevDefault" ||
@@ -5597,7 +5599,9 @@ function validateFunction(
       if (
         e.fn === "stream.duplexFromGen" ||
         e.fn === "stream.duplexFromDyn" ||
-        e.fn === "stream.duplexFromAsyncFn"
+        e.fn === "stream.duplexFromAsyncFn" ||
+        e.fn === "stream.duplexFromGenSignal" ||
+        e.fn === "stream.duplexFromAsyncFnSignal"
       ) {
         // Duplex.from: (stage, source, helpers...) or (body) → a Duplex.
         if (!isStreamObject(e.type)) {
@@ -5786,6 +5790,8 @@ function validateFunction(
         e.fn === "readable.opMap" ||
         e.fn === "readable.opFilter" ||
         e.fn === "readable.opFlatMap" ||
+        e.fn === "readable.opFlatMapAsyncGen" ||
+        e.fn === "readable.opFlatMapSyncGen" ||
         e.fn === "readable.opDrop" ||
         e.fn === "readable.opTake"
       ) {
@@ -5794,13 +5800,30 @@ function validateFunction(
         }
         return true;
       }
-      if (e.fn === "readable.opReduce") {
+      if (e.fn === "readable.opReduce" || e.fn === "readable.opReduceSignal") {
         const inner = e.type.kind === "promise" ? e.type.inner : null;
         const kind = e.args[5]?.kind === "numLit" ? e.args[5].value : -1;
         const want = [DYN, F64, BOOL, STRING][kind];
         if (!isStreamObject(e.args[0]?.type) || !inner || !want || !typeEquals(inner, want)) {
           err(`libCall readable.opReduce must return the promise its kind names`, e.loc);
         }
+        if (e.fn === "readable.opReduceSignal" && e.args[6]?.type.kind !== "func")
+          err(`libCall ${e.fn} takes ac.abort() as a closure`, e.loc);
+        return true;
+      }
+      if (e.fn === "readable.opOnceOptions") {
+        if (e.args.length !== 0 || e.type.kind !== "dyn")
+          err(`libCall ${e.fn} takes no arguments`, e.loc);
+        return true;
+      }
+      if (
+        e.fn === "readable.opSignalList" ||
+        e.fn === "readable.opSignalOpts" ||
+        e.fn === "readable.opSignalOf"
+      ) {
+        // The operator callbacks' { signal }: (dyn) → dyn.
+        if (e.args.length !== 1 || e.args[0]?.type.kind !== "dyn" || e.type.kind !== "dyn")
+          err(`libCall ${e.fn} takes and returns a checked-dynamic value`, e.loc);
         return true;
       }
       if (e.fn === "readable.fromDone") {

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_v31");
+    expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v32");
 
     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_v31");
+    expect(undefinedSymbols).toContain("_scr_runtime_abi_v32");
   });
 
   test("helper and clang object paths retain the same outbound FFI C ABI", async () => {

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7433-readable-operator-signal.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7433-readable-operator-signal.mjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7434-readable-operator-values.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7434-readable-operator-values.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7435-readable-flatmap-iterables.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7435-readable-flatmap-iterables.mjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7436-readable-read-any-value.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7436-readable-read-any-value.mjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7437-duplex-from-signal.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7437-duplex-from-signal.mjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7438-duplex-from-signal.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7438-duplex-from-signal.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7439-readable-read-string-chunks.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7439-readable-read-string-chunks.ts"
+ ],
+ "diags": []
+}

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

@@ -340,7 +340,7 @@ async function build() {
       package: packageManifest.name,
       version: packageManifest.version,
       target: RUNTIME_PACK_MATRIX.target,
-      runtime_abi: { version: 31, marker: "scr_runtime_abi_v31" },
+      runtime_abi: { version: 32, marker: "scr_runtime_abi_v32" },
       compiler: {
         command: compiler,
         identity: compilerVersion,

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

@@ -437,7 +437,7 @@ async function build() {
       package: packageManifest.name,
       version: packageManifest.version,
       target: matrix.target,
-      runtime_abi: { version: 31, marker: "scr_runtime_abi_v31" },
+      runtime_abi: { version: 32, marker: "scr_runtime_abi_v32" },
       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_v31(void) {}
+void scr_runtime_abi_v32(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_v31(void);
+void scr_runtime_abi_v32(void);
 void scr_runtime_workers_v8(void);
 
 /* ── the trap funnel (scr_console.c; scr_library.c under -DSCR_LIB) ──────

packages/runtime/src/scr_stream.h

@@ -431,18 +431,41 @@ int scr_stream_push_object(ScrStream *s, const ScrDyn *v);
  * dyn array, find and reduce a dyn value — reduce's `kind` picks the
  * payload instead: 0 dyn, 1 number, 2 boolean, 3 string — some and every
  * a boolean, forEach nothing. */
-ScrStream *scr_stream_op_map(ScrStream *s, ScrDyn *fn, ScrDyn *options);
-ScrStream *scr_stream_op_filter(ScrStream *s, ScrDyn *fn, ScrDyn *options);
-ScrStream *scr_stream_op_flat_map(ScrStream *s, ScrDyn *fn, ScrDyn *options);
+ScrStream *scr_stream_op_map(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts);
+ScrStream *scr_stream_op_filter(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts);
+ScrStream *scr_stream_op_flat_map(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts);
+/* flatMap whose callback returns a compiled generator: the helpers over its
+ * type (async: next, read, return, throw; sync: next, return, throw;
+ * borrowed). */
+ScrStream *scr_stream_op_flat_map_async_gen(ScrStream *s, ScrDyn *fn, ScrDyn *options,
+                                            ScrDyn *cb_opts, ScrClosure *next, ScrClosure *read,
+                                            ScrClosure *ret, ScrClosure *thr);
+ScrStream *scr_stream_op_flat_map_sync_gen(ScrStream *s, ScrDyn *fn, ScrDyn *options,
+                                           ScrDyn *cb_opts, ScrClosure *next, ScrClosure *ret,
+                                           ScrClosure *thr);
 ScrStream *scr_stream_op_drop(ScrStream *s, ScrDyn *number, ScrDyn *options);
 ScrStream *scr_stream_op_take(ScrStream *s, ScrDyn *number, ScrDyn *options);
 ScrPromise *scr_stream_op_to_array(ScrStream *s, ScrDyn *options);
-ScrPromise *scr_stream_op_for_each(ScrStream *s, ScrDyn *fn, ScrDyn *options);
-ScrPromise *scr_stream_op_some(ScrStream *s, ScrDyn *fn, ScrDyn *options);
-ScrPromise *scr_stream_op_every(ScrStream *s, ScrDyn *fn, ScrDyn *options);
-ScrPromise *scr_stream_op_find(ScrStream *s, ScrDyn *fn, ScrDyn *options);
+ScrPromise *scr_stream_op_for_each(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts);
+ScrPromise *scr_stream_op_some(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts);
+ScrPromise *scr_stream_op_every(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts);
+ScrPromise *scr_stream_op_find(ScrStream *s, ScrDyn *fn, ScrDyn *options, ScrDyn *cb_opts);
 ScrPromise *scr_stream_op_reduce(ScrStream *s, ScrDyn *fn, ScrDyn *init, bool has_init,
                                  ScrDyn *options, double kind);
+/* reduce whose reducer reads its { signal } (cb_opts): abort is ac.abort()
+ * (borrowed), run when the reduction ends. */
+ScrPromise *scr_stream_op_reduce_signal(ScrStream *s, ScrDyn *fn, ScrDyn *init, bool has_init,
+                                        ScrDyn *options, double kind, ScrClosure *abort,
+                                        ScrDyn *cb_opts);
+/* `options?.signal` (+1, undefined when absent) and `{ once: true }` (+1),
+ * for the listener that makes options.signal abort reduce's controller. */
+ScrDyn *scr_stream_op_signal_of(ScrDyn *options);
+ScrDyn *scr_stream_op_once_options(void);
+/* The operator callbacks' { signal } (the options argument): the signals
+ * AbortSignal.any combines (`[options?.signal].filter(Boolean)`) and the
+ * object around the combined signal. +1 each. */
+ScrDyn *scr_stream_op_signal_list(ScrDyn *options);
+ScrDyn *scr_stream_op_signal_opts(ScrDyn *signal);
 /* finished(s, options[, cb]) — Node's eos with its listeners and state
  * predicates: readable/writable are -1 (the default), 0 or 1; on_error is
  * options.error !== false; signal is NULL or borrowed (ERR_INVALID_ARG_TYPE
@@ -556,6 +579,16 @@ ScrStream *scr_stream_readable_from_web(struct ScrDyn *web, bool object_mode, do
  * generator and answers its promise; source and the promise's box helper
  * borrowed): a writable Duplexify whose final waits for the promise. +1. */
 ScrStream *scr_stream_duplex_from_async_fn(ScrClosure *stage, ScrClosure *source, ScrClosure *box);
+/* The same for a body that declares the options argument: it takes
+ * { signal } (signal borrowed), and destroy calls abort(err) (borrowed),
+ * the AbortController's abort, before it settles (fromAsyncGen's
+ * ac.abort(err)). */
+ScrStream *scr_stream_duplex_from_gen_signal(ScrClosure *stage, ScrClosure *source, ScrClosure *next,
+                                             ScrClosure *read, ScrClosure *ret, ScrClosure *thr,
+                                             ScrClosure *abort, struct ScrDyn *signal);
+ScrStream *scr_stream_duplex_from_async_fn_signal(ScrClosure *stage, ScrClosure *source,
+                                                  ScrClosure *box, ScrClosure *abort,
+                                                  struct ScrDyn *signal);
 /* Duplex.from over a Readable, a Writable, or a { readable, writable }
  * pair (duplexify.js's _duplexify): a Duplexify that reads r and writes w
  * (either NULL in scr_stream_duplexify; streams borrowed). +1. */

tests/corpus/7433-readable-operator-signal.mjs

@@ -0,0 +1,36 @@
+// The Readable operators pass their callbacks { signal } as the second
+// argument; map and filter's signal aborts when the options' signal
+// does, reduce's signal is aborted once reduce settles, and map may
+// return any value.
+import { Readable } from "node:stream";
+const log = [];
+log.push(JSON.stringify(await Readable.from([1, 2, 3]).map((n) => ({ n, sq: n * n })).toArray()));
+log.push(JSON.stringify(await Readable.from(["a", "bb"]).map((s) => s.length).toArray()));
+log.push(JSON.stringify(await Readable.from([1, 2]).map((n, { signal }) => `${n}:${signal.aborted}`).toArray()));
+log.push(JSON.stringify(await Readable.from([1, 2, 3, 4]).filter(async (n, opts) => n % 2 === 0 && !opts.signal.aborted).toArray()));
+log.push(String(await Readable.from([1, 2, 3]).reduce((acc, n, { signal }) => acc + n + (signal.aborted ? 100 : 0), 0)));
+let kept;
+await Readable.from([1]).reduce((acc, n, { signal }) => { kept = signal; return acc + n; }, 0);
+log.push(`reduce signal after ${kept.aborted}`);
+let mapKept;
+await Readable.from([1]).map((n, { signal }) => { mapKept = signal; return n; }).toArray();
+log.push(`map signal after ${mapKept.aborted}`);
+const ac = new AbortController();
+const events = [];
+const p = Readable.from([1, 2, 3]).map(async (n, { signal }) => {
+  signal.addEventListener("abort", () => events.push(`abort ${n}`));
+  await new Promise((r) => setTimeout(r, 10));
+  return n;
+}, { signal: ac.signal }).toArray();
+setTimeout(() => ac.abort(), 5);
+try { await p; } catch (e) { log.push(`map aborted ${e.name}`); }
+log.push(events.join(","));
+const dyn = [Readable.from([4, 5])][0];
+log.push(JSON.stringify(await dyn.map((n, { signal }) => n + Number(signal.aborted)).toArray()));
+let seen = 0;
+await Readable.from([5, 6]).forEach((n, { signal }) => { if (signal) seen++; });
+log.push(`forEach ${seen}`);
+log.push(String(await Readable.from([1, 2, 3]).some((n, { signal }) => n > 2 && !signal.aborted)));
+log.push(String(await Readable.from([1, 2, 3]).find((n, { signal }) => n > 1 && !signal.aborted)));
+log.push(String(await Readable.from([1, 2, 3]).every((n, { signal }) => n > 0 && !signal.aborted)));
+console.log(log.join("\n"));

tests/corpus/7434-readable-operator-values.ts

@@ -0,0 +1,32 @@
+// Typed Readable operators: map returns numbers and records, flatMap
+// takes generator functions and sets, the callbacks may declare the
+// { signal } argument, and read() on an object-mode stream answers the
+// values pushed.
+import { Readable } from "node:stream";
+async function* pair(n: number): AsyncGenerator<number> {
+  yield n;
+  yield n * 100;
+}
+function* letters(s: string): Generator<string> {
+  for (const c of s) yield c;
+}
+async function main(): Promise<void> {
+  const squares = await Readable.from([1, 2, 3]).map((n: number) => n * n).toArray();
+  const records = await Readable.from(["a", "bb"]).map((s: string) => ({ s, len: s.length })).toArray();
+  const pairs = await Readable.from([1, 2]).flatMap(pair).toArray();
+  const chars = await Readable.from(["hi"]).flatMap(letters).toArray();
+  const sets = await Readable.from([7]).flatMap((n: number) => new Set([n, n + 1])).toArray();
+  const signals = await Readable.from([1]).map((n: number, opts?: { signal?: AbortSignal }) => `${n}:${opts?.signal?.aborted}`).toArray();
+  const total = await Readable.from([1, 2]).reduce((acc: number, n: number, opts?: { signal?: AbortSignal }) => acc + n + (opts?.signal?.aborted ? 1 : 0), 0);
+  console.log(JSON.stringify({ squares, records, pairs, chars, sets, signals, total }));
+  const r = new Readable({ objectMode: true, read(): void {} });
+  r.push({ a: 1 });
+  r.push(5);
+  r.push(null);
+  r.once("readable", () => {
+    const first: unknown = r.read();
+    const second: unknown = r.read();
+    console.log(JSON.stringify(first), second, r.read());
+  });
+}
+void main();

tests/corpus/7435-readable-flatmap-iterables.mjs

@@ -0,0 +1,29 @@
+// flatMap over what the callback returns: async and sync generators, a
+// Set, a Map, a Readable, and a value that is not iterable (V8's
+// TypeError); the generator's steps interleave with process.nextTick as
+// in Node, breaking out of the loop runs the inner generator's finally,
+// and an inner throw rejects toArray().
+import { Readable } from "node:stream";
+const log = [];
+log.push(JSON.stringify(await Readable.from(["x"]).flatMap(async function* (s) { yield s; yield s + s; }).toArray()));
+log.push(JSON.stringify(await Readable.from([2]).flatMap(function* (n) { yield n; yield -n; }).toArray()));
+log.push(JSON.stringify(await Readable.from([1, 2]).flatMap((n) => new Set([n, n])).toArray()));
+log.push(JSON.stringify(await Readable.from(["p"]).flatMap((s) => Readable.from([s, s.toUpperCase()])).toArray()));
+log.push(JSON.stringify(await Readable.from([3]).flatMap((n) => new Map([[n, "v"]])).toArray()));
+log.push(JSON.stringify(await Readable.from([{ k: 1 }]).flatMap((o) => [o, { k: o.k + 1 }]).toArray()));
+try { await Readable.from([1]).flatMap(() => 5).toArray(); } catch (e) { log.push(`${e.name} ${e.message}`); }
+try { await Readable.from([1]).flatMap(() => null).toArray(); } catch (e) { log.push(`${e.name} ${e.message}`); }
+const order = [];
+let n = 0;
+const tick = () => { order.push(`t${n}`); if (++n < 5) process.nextTick(tick); };
+process.nextTick(tick);
+for await (const v of Readable.from([1]).flatMap(async function* (k) { order.push("gen start"); yield k; order.push("gen mid"); yield k + 1; })) order.push(`v${v}`);
+log.push(order.join(" "));
+const closed = [];
+const early = Readable.from([1, 2]).flatMap(async function* (k) { try { yield k; yield k * 10; } finally { closed.push(`finally ${k}`); } });
+for await (const v of early) { closed.push(`got ${v}`); break; }
+log.push(closed.join(" "));
+const thrown = [];
+try { await Readable.from([1]).flatMap(async function* () { yield 1; throw new Error("inner boom"); }).toArray(); } catch (e) { thrown.push(e.message); }
+log.push(thrown.join(" "));
+console.log(log.join("\n"));

tests/corpus/7436-readable-read-any-value.mjs

@@ -0,0 +1,17 @@
+// readable.read() answers any value on an object-mode stream, and a
+// Buffer or a string (with an encoding) in byte mode.
+import { Readable, PassThrough } from "node:stream";
+const r = new Readable({ objectMode: true, read() {} });
+r.push({ a: 1 }); r.push(2); r.push("three"); r.push([4]); r.push(null);
+r.once("readable", () => { let v; while ((v = r.read()) !== null) console.log("read", JSON.stringify(v)); });
+const p = new PassThrough({ objectMode: true });
+p.write([1, 2]);
+p.end();
+p.once("readable", () => console.log("pt", JSON.stringify(p.read()), p.read()));
+const b = new PassThrough();
+b.end("bytes!");
+b.once("readable", () => { const c = b.read(3); console.log("bytes", Buffer.isBuffer(c), String(c), String(b.read())); });
+const e = new PassThrough();
+e.setEncoding("utf8");
+e.end("text");
+e.once("readable", () => { const t = e.read(); console.log("encoded", typeof t, t); });

tests/corpus/7437-duplex-from-signal.mjs

@@ -0,0 +1,54 @@
+// A Duplex.from body may take { signal } (duplexify.js's fromAsyncGen):
+// destroy aborts it with the destroy error, or with an AbortError when
+// there is none, and compose passes it to its function stages.
+import { Duplex, Readable, compose } from "node:stream";
+import { once } from "node:events";
+
+const d = Duplex.from(async function* (source, { signal }) {
+  console.log("body", typeof signal, signal.aborted);
+  signal.addEventListener("abort", () => console.log("abort event", signal.aborted, signal.reason));
+  for await (const chunk of source) yield String(chunk).toUpperCase();
+  console.log("source ended");
+});
+d.on("data", (c) => console.log("data", c));
+d.write("a");
+d.write("b");
+await new Promise((r) => setTimeout(r, 10));
+d.destroy();
+await once(d, "close");
+console.log("closed d");
+
+const seen = [];
+const fin = Duplex.from(async function (source, { signal }) {
+  signal.addEventListener("abort", () => console.log("abort", seen.join(",")));
+  for await (const chunk of source) seen.push(String(chunk));
+  console.log("done", signal.aborted);
+});
+fin.on("finish", () => console.log("finish"));
+fin.write("1");
+fin.write("2");
+fin.end();
+await once(fin, "close");
+console.log("closed fin");
+
+const boom = new Error("boom");
+const e = Duplex.from(async function (source, { signal }) {
+  signal.addEventListener("abort", () => console.log("e abort", String(signal.reason), signal.reason === boom));
+  await new Promise((r) => setTimeout(r, 50));
+});
+e.on("error", (err) => console.log("e error", err === boom));
+setTimeout(() => e.destroy(boom), 5);
+await new Promise((r) => e.on("close", r));
+console.log("closed e");
+
+const c = compose(
+  async function* (source, { signal }) {
+    for await (const chunk of source) yield `${chunk}!${signal.aborted}`;
+  },
+  async function* (source, { signal }) {
+    for await (const chunk of source) yield chunk.toUpperCase();
+  },
+);
+Readable.from(["a", "b"]).pipe(c);
+c.on("data", (x) => console.log("compose", String(x)));
+c.on("end", () => console.log("compose end"));

tests/corpus/7438-duplex-from-signal.ts

@@ -0,0 +1,26 @@
+// Typed Duplex.from bodies that declare the options argument: an async
+// generator function and an async function.
+import { Duplex } from "node:stream";
+
+const d = Duplex.from(async function* (source: AsyncIterable<string>, opts: { signal: AbortSignal }) {
+  opts.signal.addEventListener("abort", () => console.log("aborted", opts.signal.aborted));
+  for await (const chunk of source) {
+    yield `<${chunk}>`;
+  }
+});
+d.on("data", (c: string) => console.log("data", c));
+d.on("end", () => console.log("end"));
+d.on("close", () => {
+  console.log("close");
+  const w = Duplex.from(async function (source: AsyncIterable<string>, { signal }: { signal: AbortSignal }): Promise<void> {
+    let n = 0;
+    for await (const chunk of source) n += chunk.length;
+    console.log("total", n, signal.aborted);
+  });
+  w.on("finish", () => console.log("finish"));
+  w.on("close", () => console.log("w close"));
+  w.write("abc");
+  w.end("de");
+});
+d.write("x");
+d.end("y");

tests/corpus/7439-readable-read-string-chunks.ts

@@ -0,0 +1,17 @@
+// A typed read() answers the strings a stream holds: the first value a
+// lazy Readable.from pull over a parsed array produces, and the decoded
+// text once _read calls setEncoding().
+import { Readable } from "node:stream";
+const r = Readable.from(JSON.parse('["j1", "j2"]'));
+const chunk = r.read();
+console.log("first", typeof chunk, chunk === null ? "null" : chunk.toString("hex"));
+r.once("readable", () => console.log("later", JSON.stringify(r.read())));
+const e = new Readable({
+  read() {
+    this.setEncoding("utf8");
+    this.push(Buffer.from("ab"));
+    this.push(null);
+  },
+});
+const text = e.read();
+console.log("encoded", typeof text, text === null ? "null" : text.toString("hex"));

tests/harness/corpus-costs/7433-readable-operator-signal.mjs.json

@@ -0,0 +1 @@
+{ "plain": 3885, "san": 5128 }

tests/harness/corpus-costs/7434-readable-operator-values.ts.json

@@ -0,0 +1 @@
+{ "plain": 2465, "san": 2933 }

tests/harness/corpus-costs/7435-readable-flatmap-iterables.mjs.json

@@ -0,0 +1 @@
+{ "plain": 2265, "san": 2990 }

tests/harness/corpus-costs/7436-readable-read-any-value.mjs.json

@@ -0,0 +1 @@
+{ "plain": 1675, "san": 2211 }

tests/harness/corpus-costs/7437-duplex-from-signal.mjs.json

@@ -0,0 +1 @@
+{ "plain": 2383, "san": 3146 }

tests/harness/corpus-costs/7438-duplex-from-signal.ts.json

@@ -0,0 +1 @@
+{ "plain": 1777, "san": 2722 }

tests/harness/corpus-costs/7439-readable-read-string-chunks.ts.json

@@ -0,0 +1 @@
+{ "plain": 2578, "san": 3403 }

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

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

tests/harness/stream-entry-kinds.test.ts

@@ -1,14 +1,9 @@
 /* Stream chunk kinds Node cannot oracle — scriptc-only.
  *
- * A typed read() answers `Buffer | null`. When the read itself makes the
- * stream hold strings (the first string a lazy Readable.from pull over a
- * dynamic array produces, or setEncoding() inside _read), Node returns
- * the string, which the typed result cannot hold, so the read throws the
- * encoding fence instead (caught here, so the program exits normally).
- * These pins keep that refusal exact and, under
- * SCRIPTC_SAN=1 (ASan and the RC audit, like every harness lane), prove
- * no string reaches Buffer code. The Node-observable cases are corpus
- * programs (stream-dynamic-entry-kinds.*).
+ * Under SCRIPTC_SAN=1 (ASan and the RC audit, like every harness lane)
+ * these pins prove no string reaches Buffer code. The Node-observable
+ * cases are corpus programs (stream-dynamic-entry-kinds.*, and
+ * 7439-readable-read-string-chunks for read() of strings).
  */
 import { execFile } from "node:child_process";
 import { createHash } from "node:crypto";
@@ -73,71 +68,7 @@ function stripAsanFiberWarning(s: string): string {
   );
 }
 
-const READ_FENCE_MESSAGE =
-  "read() on a stream with an encoding set is not supported yet (consume 'data' events, which deliver strings)";
-
 describe(`typed stream reads of string chunks (scriptc-only${sanitize ? ", sanitized" : ""})`, () => {
-  test("a typed read() whose lazy pull yields the first string throws the encoding fence", async () => {
-    const r = await compileAndRun(
-      "typed-read-json-string",
-      `import { Readable } from "node:stream";
-const r = Readable.from(JSON.parse('["j1", "j2"]'));
-try {
-  const chunk = r.read();
-  console.log("hex", chunk === null ? "null" : chunk.toString("hex"));
-} catch (err) {
-  console.log("caught", (err as Error).message);
-}
-`,
-    );
-    expect(r.exitCode).toBe(0);
-    expect(r.stdout).toBe(`caught ${READ_FENCE_MESSAGE}\n`);
-  });
-
-  test("a typed read() whose _read calls setEncoding() throws the encoding fence", async () => {
-    const r = await compileAndRun(
-      "typed-read-encoding-in-read",
-      `import { Readable } from "node:stream";
-const r = new Readable({
-  read() {
-    this.setEncoding("utf8");
-    this.push(Buffer.from("ab"));
-    this.push(null);
-  },
-});
-try {
-  const chunk = r.read();
-  console.log("hex", chunk === null ? "null" : chunk.toString("hex"));
-} catch (err) {
-  console.log("caught", (err as Error).message);
-}
-`,
-    );
-    expect(r.exitCode).toBe(0);
-    expect(r.stdout).toBe(`caught ${READ_FENCE_MESSAGE}\n`);
-  });
-
-  test("the JavaScript lane's read() on a JSON source throws the same fence", async () => {
-    const r = await compileAndRun(
-      "js-read-json-string",
-      `'use strict';
-const { Readable } = require('node:stream');
-const r = Readable.from(JSON.parse('["j1"]'));
-r.once('readable', () => {
-  try {
-    const chunk = r.read();
-    console.log('got', chunk === null ? 'null' : chunk.length);
-  } catch (err) {
-    console.log('caught', err.message);
-  }
-});
-`,
-      "cjs",
-    );
-    expect(r.exitCode).toBe(0);
-    expect(r.stdout).toBe(`caught ${READ_FENCE_MESSAGE}\n`);
-  });
-
   test("a dynamic source that mixes Buffers and strings yields each as it is, as Node does", async () => {
     const r = await compileAndRun(
       "mixed-dynamic-source",