Close async generators on early for await exits and iterate untyped native streams

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: 28, marker: "scr_runtime_abi_v28" },
+      runtime_abi: { version: 29, marker: "scr_runtime_abi_v29" },
       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_v28");
+    expect(error?.stderr).toContain("scr_runtime_abi_v29");
   });
 });

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

@@ -972,6 +972,11 @@ export class LlEmitter {
   get usesDynBuiltinStream(): boolean {
     return this.runtimeFeatures.dynBuiltinStream;
   }
+  /** A checked-dynamic for-await may hold a native stream: the program
+   * links the stream unit (and the dyn-async one the loop needs). */
+  get dynForAwaitStreams(): boolean {
+    return this.runtimeFeatures.stream && this.runtimeFeatures.dynAsync;
+  }
   /** Runtime-arm dispatchers for live union values. */
   readonly liveDynUnionRefAdapters = new Map<string, string>();
   readonly dynPromiseAdapters = new Map<string, string>();

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

@@ -553,7 +553,11 @@ export function emitGenericLibCall(host: LlvmEmitterContext, e: LibCallExpr): Ll
         // which also takes those streams.
         e.fn === "stream.unboxDyn" && host.usesDynBuiltinStream
         ? "scr_dyn_builtin_stream_unbox"
-        : LIB_FN_SYMS[e.fn]);
+        : // for await over a checked-dynamic value iterates a native
+          // stream behind it (scr_dyn_for_await_stream.c)
+          e.fn === "dyn.asyncIteratorOpen" && host.dynForAwaitStreams
+          ? "scr_stream_dyn_async_iterator_open"
+          : LIB_FN_SYMS[e.fn]);
   if (sym === undefined) throw new LlvmUnsupportedError(`libCall:${e.fn}`, e.loc);
   const args = READ_ONLY_STRING_LIB_FNS.has(e.fn)
     ? emitBorrowedInputs(host, e.args)

packages/compiler/src/backend/native/executable.ts

@@ -281,6 +281,11 @@ const RUNTIME_UNITS: readonly RuntimeUnit[] = [
   unit((u) => u.opts.handleProps === true, "scr_dyn_handle_props.c"),
   // for await over checked-dynamic values.
   unit((u) => u.opts.dynForAwait === true, "scr_dyn_for_await.c"),
+  // for await over checked-dynamic values that may hold native streams.
+  unit(
+    (u) => u.opts.dynForAwait === true && u.opts.stream && u.opts.dynAsync,
+    "scr_dyn_for_await_stream.c",
+  ),
   // FileHandle streams reference both the stream unit and FileHandle.
   unit((u) => u.opts.stream && u.opts.fileHandle, "scr_fs_stream_fh.c"),
   // The readiness-poller backends (scr_platform.h): kqueue on macOS/BSD,

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 = 28 as const;
-export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v28" as const;
+export const RUNTIME_ABI_VERSION = 29 as const;
+export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v29" 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: 28, marker: "scr_runtime_abi_v28" },
+    runtime_abi: { version: 29, marker: "scr_runtime_abi_v29" },
     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: 28, marker: "scr_runtime_abi_v28" },
+    runtime_abi: { version: 29, marker: "scr_runtime_abi_v29" },
     compiler: {
       command: "clang",
       identity: "fixture clang",

packages/compiler/src/frontend/lowering/lower-async-iteration.ts

@@ -359,13 +359,15 @@ function boolRef(localId: string, loc: ReturnType<typeof locOf>): IrExpr {
 
 /** The loop's stream: a JavaScript value typed as a stream can arrive as a
  * checked-dynamic value (returned through an untyped function), which
- * checks back into the stream class. */
+ * checks back into the stream class, or as a union with it (an element
+ * read), which narrows to it. */
 function streamSource(
   lowerer: Lowerer,
   plan: NodeReadablePlan,
   loc: ReturnType<typeof locOf>,
 ): IrExpr {
   const v = lowerer.lowerExpr(plan.source);
+  if (v.type.kind === "union") return lowerer.coerceToExpected(v, plan.type);
   if (v.type.kind !== "dyn") return v;
   return { kind: "dynCheck", value: v, type: plan.type, loc };
 }

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

@@ -4338,6 +4338,21 @@ export function lowerDynDispatchMethodCall(
     lowerer.chainBlocked(call, access)
   )
     return null;
+  // readable.compose lowers its stages from their syntax, so it needs the
+  // stream's class: a stream a JavaScript function returns untyped has none.
+  if (method === "compose") {
+    const recvT = lowerer.mapTypeOf(lowerer.typeOf(access.expression));
+    if (
+      recvT?.kind === "object" &&
+      streamSidesOf(lowerer, lowerer.classes.get(recvT.className)) !== null
+    ) {
+      lowerer.noLowering(
+        "readable.compose on a stream held as a checked-dynamic value",
+        call,
+        "type the stream where it comes from, such as a JSDoc @returns {Readable} on the function that returns it",
+      );
+    }
+  }
   if (call.arguments.nodes.some((arg) => ts.isSpreadElement(arg))) {
     const loc = locOf(call);
     const local = lowerer.declareHiddenLocal("%spreadReceiver", DYN);

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

@@ -687,8 +687,12 @@ export function lowerForAwaitGenerator(
   lowerer.scopes.push(new Map());
   try {
     const g = lowerer.declareHiddenLocal("%fag", genT);
-    const done = lowerer.declareHiddenLocal("%fagdone", BOOL);
-    done.mutable = true;
+    // Whether a value has arrived and the body holds the generator open:
+    // leaving the loop then (break, return, a throw, or a return() at a
+    // yield in the body) closes it, as AsyncIteratorClose does; an
+    // exhausted generator and a rejected next() are not closed.
+    const active = lowerer.declareHiddenLocal("%fagactive", BOOL);
+    active.mutable = true;
     const p = lowerer.declareHiddenLocal("%fagpromise", promiseT);
     const r = lowerer.declareHiddenLocal("%fagresult", recT);
     const gRef = (): IrExpr => ({ kind: "varRef", localId: g.id, type: genT, loc });
@@ -709,7 +713,14 @@ export function lowerForAwaitGenerator(
         `for-await over an async generator yielding '${lowerer.fmt(genT.yieldT)}' (no per-element extraction exists)`,
       );
     }
+    const setActive = (value: boolean): IrStmt => ({
+      kind: "assign",
+      localId: active.id,
+      value: { kind: "boolLit", value, type: BOOL, loc },
+      loc,
+    });
     const head: IrStmt[] = [
+      setActive(false),
       {
         kind: "varDecl",
         localId: p.id,
@@ -737,19 +748,12 @@ export function lowerForAwaitGenerator(
           type: BOOL,
           loc,
         },
-        then: [
-          {
-            kind: "assign",
-            localId: done.id,
-            value: { kind: "boolLit", value: true, type: BOOL, loc },
-            loc,
-          },
-          { kind: "break", loc },
-        ],
+        then: [{ kind: "break", loc }],
         else_: null,
         loc,
       },
       ...generatorLoopBinding(lowerer, stmt, extracted, genT.yieldT),
+      setActive(true),
     ];
     const body = lowerer.inCtl("loop", () => lowerer.lowerScopedBlock(stmt.statement), labels);
     const closePromise: IrExpr = {
@@ -766,34 +770,40 @@ export function lowerForAwaitGenerator(
         { kind: "varDecl", localId: g.id, init: iterable, loc },
         {
           kind: "varDecl",
-          localId: done.id,
+          localId: active.id,
           init: { kind: "boolLit", value: false, type: BOOL, loc },
           loc,
         },
         {
-          kind: "while",
-          cond: { kind: "boolLit", value: true, type: BOOL, loc },
-          body: [...head, ...body],
-          ...(labels && { labels }),
-          loc,
-        },
-        {
-          kind: "if",
-          cond: {
-            kind: "unary",
-            op: "!",
-            operand: { kind: "varRef", localId: done.id, type: BOOL, loc },
-            type: BOOL,
-            loc,
-          },
-          then: [
+          kind: "tryCatch",
+          tryBody: [
             {
-              kind: "exprStmt",
-              expr: { kind: "awaitExpr", value: closePromise, type: recT, loc },
+              kind: "while",
+              cond: { kind: "boolLit", value: true, type: BOOL, loc },
+              body: [...head, ...body],
+              ...(labels && { labels }),
+              loc,
+            },
+          ],
+          catchBody: null,
+          catchLocalId: null,
+          // After a throw, an error from closing the generator is dropped.
+          dropFinallyErrors: true,
+          finallyBody: [
+            {
+              kind: "if",
+              cond: { kind: "varRef", localId: active.id, type: BOOL, loc },
+              then: [
+                {
+                  kind: "exprStmt",
+                  expr: { kind: "awaitExpr", value: closePromise, type: recT, loc },
+                  loc,
+                },
+              ],
+              else_: null,
               loc,
             },
           ],
-          else_: null,
           loc,
         },
       ],

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

@@ -3944,6 +3944,29 @@ function lowerStreamCompose(lowerer: Lowerer, args: readonly ts.Expression[], lo
 /* ── the Readable operators ──────────────────────────────────────────── */
 
 /** The Readable operators (scr_stream_ops.c — Node's operators.js). */
+/** A stream member's receiver, lowered: a value the checker types as a
+ * stream class whose lowering is checked-dynamic (a JavaScript binding
+ * that holds an untyped call's result) or a union (an element read)
+ * converts to that class, checked where the member runs. */
+function streamReceiverExpr(lowerer: Lowerer, node: ts.Expression): IrExpr {
+  const value = lowerer.lowerExpr(node);
+  if (value.type.kind !== "dyn" && value.type.kind !== "union") return value;
+  let classT: IrType | null;
+  try {
+    classT = lowerer.mapTypeOf(lowerer.typeOf(node));
+  } catch {
+    return value;
+  }
+  if (
+    classT?.kind !== "object" ||
+    streamSidesOf(lowerer, lowerer.classes.get(classT.className)) === null
+  )
+    return value;
+  return value.type.kind === "dyn"
+    ? { kind: "dynCheck", value, type: classT, loc: value.loc }
+    : lowerer.coerceToExpected(value, classT);
+}
+
 export const STREAM_OPERATORS: ReadonlySet<string> = new Set([
   "map",
   "filter",
@@ -4066,7 +4089,7 @@ function lowerStreamOperator(
 ): IrExpr {
   const loc = locOf(call);
   const args = call.arguments.nodes;
-  const recv = lowerer.lowerExpr(access.expression);
+  const recv = streamReceiverExpr(lowerer, access.expression);
   const readable: IrType = { kind: "object", className: "%Readable" };
   const opts = (i: number): IrExpr =>
     lowerOperatorValue(lowerer, args[i], `${member} options`, loc, true);
@@ -4210,7 +4233,7 @@ function pipeIntoStdio(
 ): IrExpr {
   const loc = locOf(call);
   // JS order: the receiver, then the destination, then the options.
-  const receiverValue = lowerer.lowerExpr(access.expression);
+  const receiverValue = streamReceiverExpr(lowerer, access.expression);
   const source = lowerer.declareHiddenLocal("%pipeSource", receiverValue.type);
   const stdio = lowerer.declareHiddenLocal("%pipeStdio", target.type);
   const options = call.arguments.nodes[1];
@@ -4332,7 +4355,7 @@ function pipeIntoDyn(
 ): IrExpr {
   const loc = locOf(call);
   noteUntypedStreamFlow(lowerer, null);
-  const receiverValue = lowerer.lowerExpr(access.expression);
+  const receiverValue = streamReceiverExpr(lowerer, access.expression);
   const source = lowerer.declareHiddenLocal("%pipeSource", receiverValue.type);
   const destination = lowerer.declareHiddenLocal("%pipeDestination", target.type);
   const stmts: IrStmt[] = [
@@ -4621,7 +4644,7 @@ export function lowerStreamMethodCall(
             loc,
           );
         }
-        const receiver = lowerer.lowerExpr(access.expression);
+        const receiver = streamReceiverExpr(lowerer, access.expression);
         const chunk = lowerer.lowerExprExpecting(args[0]!, STRING);
         return {
           kind: "libCall",
@@ -4637,7 +4660,7 @@ export function lowerStreamMethodCall(
     if ((kind === "null" || kind === "union" || kind === "dyn") && member === "unshift") {
       // unshift(value): an object-mode stream's value at the front, null
       // the end of the stream, as Node's readableAddChunk takes them
-      const receiver = lowerer.lowerExpr(access.expression);
+      const receiver = streamReceiverExpr(lowerer, access.expression);
       const chunk = lowerer.lowerExprExpecting(args[0]!, DYN);
       return {
         kind: "libCall",
@@ -4647,7 +4670,7 @@ export function lowerStreamMethodCall(
         loc,
       };
     }
-    const receiver = lowerer.lowerExpr(access.expression);
+    const receiver = streamReceiverExpr(lowerer, access.expression);
     if (kind === "null") {
       return { kind: "libCall", fn: "readable.pushNull", args: [receiver], type: BOOL, loc };
     }
@@ -4700,7 +4723,7 @@ export function lowerStreamMethodCall(
   if (member === "read") {
     requireSide(readableSide, "read");
     if (args.length > 1) lowerer.noLowering(`read with ${args.length} arguments`, call);
-    const receiver = lowerer.lowerExpr(access.expression);
+    const receiver = streamReceiverExpr(lowerer, access.expression);
     const size = args[0] ? lowerer.lowerExprExpecting(args[0], F64) : numLit(-1, loc);
     const type: IrType = unionOf(lowerer, [BYTES, { kind: "nullT" }]);
     const read: IrExpr = {
@@ -4716,7 +4739,7 @@ export function lowerStreamMethodCall(
   if (member === "pause" || member === "resume") {
     requireSide(readableSide, member);
     if (args.length !== 0) lowerer.noLowering(`${member} with arguments`, call);
-    const receiver = lowerer.lowerExpr(access.expression);
+    const receiver = streamReceiverExpr(lowerer, access.expression);
     return {
       kind: "libCall",
       fn: member === "pause" ? "readable.pause" : "readable.resume",
@@ -4729,7 +4752,7 @@ export function lowerStreamMethodCall(
   if (member === "isPaused") {
     requireSide(readableSide, "isPaused");
     if (args.length !== 0) lowerer.noLowering(`isPaused with arguments`, call);
-    const receiver = lowerer.lowerExpr(access.expression);
+    const receiver = streamReceiverExpr(lowerer, access.expression);
     return { kind: "libCall", fn: "readable.isPaused", args: [receiver], type: BOOL, loc };
   }
 
@@ -4746,7 +4769,7 @@ export function lowerStreamMethodCall(
     // Transform, and pipe() answers the value itself (the receiver
     // evaluates first, as in JS).
     if (isValueStream(lowerer, args[0]!)) {
-      const receiverValue = lowerer.lowerExpr(access.expression);
+      const receiverValue = streamReceiverExpr(lowerer, access.expression);
       const source = lowerer.declareHiddenLocal("%pipeSource", receiverValue.type);
       const hashValue = lowerer.lowerExpr(args[0]!);
       const hash = lowerer.declareHiddenLocal("%pipeHash", hashValue.type);
@@ -4796,7 +4819,7 @@ export function lowerStreamMethodCall(
       );
     }
     const end = pipeEndOption(lowerer, args[1], loc);
-    const receiver = lowerer.lowerExpr(access.expression);
+    const receiver = streamReceiverExpr(lowerer, access.expression);
     return {
       kind: "libCall",
       fn: generalPipe(lowerer, dst.type),
@@ -4809,7 +4832,7 @@ export function lowerStreamMethodCall(
   if (member === "unpipe") {
     requireSide(readableSide, "unpipe");
     if (args.length > 1) lowerer.noLowering(`unpipe with ${args.length} arguments`, call);
-    const receiver = lowerer.lowerExpr(access.expression);
+    const receiver = streamReceiverExpr(lowerer, access.expression);
     if (args[0] !== undefined) {
       const dst = lowerer.lowerExpr(args[0]);
       if (dst.type.kind === "dyn") return unpipeDyn(lowerer, receiver, dst, loc);
@@ -4870,7 +4893,7 @@ export function lowerStreamMethodCall(
 
   if (member === "destroy") {
     if (args.length > 1) lowerer.noLowering(`destroy with ${args.length} arguments`, call);
-    const receiver = lowerer.lowerExpr(access.expression);
+    const receiver = streamReceiverExpr(lowerer, access.expression);
     if (args[0] !== undefined) {
       const err = lowerer.lowerExpr(args[0]);
       const rootsAtError = (t: IrType): boolean => {
@@ -4923,7 +4946,7 @@ export function lowerStreamMethodCall(
     // the callback took its place (a byte-mode side encodes a string
     // chunk with it).
     const writeEnc = (encArg: ts.Expression | undefined, cbArg: ts.Expression | undefined) => {
-      const receiver = lowerer.lowerExpr(access.expression);
+      const receiver = streamReceiverExpr(lowerer, access.expression);
       const chunk = lowerer.lowerExprExpecting(args[0]!, DYN);
       const tail: IrExpr[] = [];
       if (encArg !== undefined) tail.push(lowerer.lowerExprExpecting(encArg, DYN));
@@ -4953,7 +4976,7 @@ export function lowerStreamMethodCall(
         }
         return writeEnc(args[1], undefined);
       }
-      const receiver = lowerer.lowerExpr(access.expression);
+      const receiver = streamReceiverExpr(lowerer, access.expression);
       const chunk = lowerer.lowerExprExpecting(args[0]!, DYN);
       return { kind: "libCall", fn: "writable.writeDyn", args: [receiver, chunk], type: BOOL, loc };
     }
@@ -4963,7 +4986,7 @@ export function lowerStreamMethodCall(
       // lowered value may have collapsed to one arm (checker narrowing) —
       // re-dispatch on its IR type.
       if (args.length > 1) lowerer.noLowering("write with a union chunk and more arguments", call);
-      const receiver = lowerer.lowerExpr(access.expression);
+      const receiver = streamReceiverExpr(lowerer, access.expression);
       const chunk = lowerer.lowerExpr(args[0]!);
       const fn =
         chunk.type.kind === "union"
@@ -4986,7 +5009,7 @@ export function lowerStreamMethodCall(
         return writeEnc(args[1], undefined);
       cbArg = args[1];
     }
-    const receiver = lowerer.lowerExpr(access.expression);
+    const receiver = streamReceiverExpr(lowerer, access.expression);
     const chunk =
       kind === "bytes"
         ? lowerer.lowerExprExpecting(args[0]!, BYTES)
@@ -5021,7 +5044,7 @@ export function lowerStreamMethodCall(
       encNode = args[1];
       cbArg = args[2];
     }
-    const receiver = lowerer.lowerExpr(access.expression);
+    const receiver = streamReceiverExpr(lowerer, access.expression);
     let flags = 0;
     const tail: IrExpr[] = [];
     if (chunkNode !== undefined) {
@@ -5068,7 +5091,7 @@ export function lowerStreamMethodCall(
   if (member === "cork" || member === "uncork") {
     requireSide(writableSide, member);
     if (args.length !== 0) lowerer.noLowering(`${member} with arguments`, call);
-    const receiver = lowerer.lowerExpr(access.expression);
+    const receiver = streamReceiverExpr(lowerer, access.expression);
     return {
       kind: "libCall",
       fn: member === "cork" ? "writable.cork" : "writable.uncork",
@@ -5088,7 +5111,7 @@ export function lowerStreamMethodCall(
       );
     }
     const enc = bufEncoding(lowerer, "setEncoding", args[0]!);
-    const receiver = lowerer.lowerExpr(access.expression);
+    const receiver = streamReceiverExpr(lowerer, access.expression);
     return {
       kind: "libCall",
       fn: "readable.setEncoding",
@@ -5104,7 +5127,7 @@ export function lowerStreamMethodCall(
     requireSide(writableSide, "setDefaultEncoding");
     if (args.length > 1)
       lowerer.noLowering(`setDefaultEncoding with ${args.length} arguments`, call);
-    const receiver = lowerer.lowerExpr(access.expression);
+    const receiver = streamReceiverExpr(lowerer, access.expression);
     const enc = args[0] ? lowerer.lowerExprExpecting(args[0], DYN) : dynUndefinedExpr(loc);
     return {
       kind: "libCall",
@@ -5257,7 +5280,7 @@ export function lowerStreamProperty(
   const propCall = (type: IrType): IrExpr => ({
     kind: "libCall",
     fn: "stream.prop",
-    args: [lowerer.lowerExpr(expr.expression), strLit(name, loc)],
+    args: [streamReceiverExpr(lowerer, expr.expression), strLit(name, loc)],
     type,
     loc,
   });
@@ -5280,7 +5303,7 @@ export function lowerStreamProperty(
     const read: IrExpr = {
       kind: "libCall",
       fn: "readable.flowing",
-      args: [lowerer.lowerExpr(expr.expression)],
+      args: [streamReceiverExpr(lowerer, expr.expression)],
       type,
       loc,
     };
@@ -5294,7 +5317,7 @@ export function lowerStreamProperty(
     const read: IrExpr = {
       kind: "libCall",
       fn: "stream.errored",
-      args: [lowerer.lowerExpr(expr.expression)],
+      args: [streamReceiverExpr(lowerer, expr.expression)],
       type,
       loc,
     };

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

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7420-for-await-close.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7420-for-await-close.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7421-for-await-close.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7421-for-await-close.mjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7422-for-await-untyped-streams.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7422-for-await-untyped-streams.mjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7423-stream-untyped-bindings.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7423-stream-untyped-bindings.mjs"
+ ],
+ "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: 28, marker: "scr_runtime_abi_v28" },
+      runtime_abi: { version: 29, marker: "scr_runtime_abi_v29" },
       compiler: {
         command: compiler,
         identity: compilerVersion,

packages/runtime-pack-common/runtime-pack-matrix.mjs

@@ -129,6 +129,7 @@ export const OPTIONAL_RUNTIME_SOURCES = [
   ["scr_dyn_handle_props.c", "handleProps"],
   // for await over checked-dynamic values.
   ["scr_dyn_for_await.c", "dynForAwait"],
+  ["scr_dyn_for_await_stream.c", all("dynForAwait", "stream", "dynAsync")],
   // fs.ReadStream / fs.WriteStream reference only the stream unit and base units.
   ["scr_fs_stream.c", "stream"],
   // FileHandle streams reference both the stream unit and FileHandle.

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: 28, marker: "scr_runtime_abi_v28" },
+      runtime_abi: { version: 29, marker: "scr_runtime_abi_v29" },
       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_v28(void) {}
+void scr_runtime_abi_v29(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_v28(void);
+void scr_runtime_abi_v29(void);
 void scr_runtime_workers_v8(void);
 
 /* ── the trap funnel (scr_console.c; scr_library.c under -DSCR_LIB) ──────

packages/runtime/src/scr_dyn.h

@@ -839,6 +839,7 @@ ScrDyn *scr_dyn_iterator(const ScrDyn *value, const ScrStr *spell);
 /* for await over a checked-dynamic value (scr_dyn_for_await.c). */
 ScrDyn *scr_dyn_async_iterator_open(ScrDyn *value, ScrStr *spell);
 bool scr_dyn_async_iterator_was_sync(void);
+void scr_dyn_async_iterator_set_sync(bool sync);
 ScrDyn *scr_dyn_iterator_optional(const ScrDyn *value);
 /* from.js's probe: 2 when value[Symbol.asyncIterator] is set, 1 when
  * value[Symbol.iterator] is, 0 for neither, -1 with an error pending. */

packages/runtime/src/scr_dyn_for_await.c

@@ -38,6 +38,11 @@ ScrDyn *scr_dyn_async_iterator_open(ScrDyn *value, ScrStr *spell) {
   return iterator;
 }
 
+/* For an iterator another open path made (scr_dyn_for_await_stream.c). */
+void scr_dyn_async_iterator_set_sync(bool sync) {
+  scr_for_await_sync = sync;
+}
+
 /* Whether the iterator the last scr_dyn_async_iterator_open returned is a
  * synchronous one. Read immediately after the open. */
 bool scr_dyn_async_iterator_was_sync(void) {

packages/runtime/src/scr_dyn_for_await_stream.c

@@ -0,0 +1,61 @@
+/* `for await` over a checked-dynamic value in a program that links the
+ * stream unit: a native readable stream behind the value (a boxed stream,
+ * or a Cipheriv/Decipheriv handle's Transform) iterates through an
+ * iterator object whose next() answers the readable iterator's step as
+ * { value, done } and whose return() closes it as a typed loop does
+ * (destroy with an AbortError). Any other value takes
+ * scr_dyn_async_iterator_open. Linked only by programs with such a loop
+ * and streams. */
+#include "scr_runtime.h"
+#include "scr_stream.h"
+
+/* caps[0]: the stream */
+static ScrStream *iter_stream(ScrClosure *clo) {
+  return scr_box_get_ref(clo->caps[0]); /* +1 */
+}
+
+static ScrDyn *iter_next_thunk(ScrClosure *clo, ScrDyn *const *args, size_t argc) {
+  (void)args;
+  (void)argc;
+  ScrStream *s = iter_stream(clo);
+  ScrPromise *p = scr_stream_next_result_dyn(s);
+  scr_stream_release(s);
+  ScrDyn *d = scr_dyn_new_promise(p);
+  scr_promise_release(p);
+  return d;
+}
+
+static ScrDyn *iter_return_thunk(ScrClosure *clo, ScrDyn *const *args, size_t argc) {
+  (void)args;
+  (void)argc;
+  ScrStream *s = iter_stream(clo);
+  scr_stream_release(scr_stream_iterator_close(s));
+  scr_stream_release(s);
+  ScrDyn *result = scr_dyn_new_obj();
+  scr_dyn_obj_set(result, "value", 5, scr_dyn_retain(scr_dyn_undefined()));
+  scr_dyn_obj_set(result, "done", 4, scr_dyn_new_bool(true));
+  ScrPromise *p = scr_promise_new();
+  scr_promise_fulfill_ref(p, result, scr_dyn_retain_v, scr_dyn_release_v, NULL);
+  ScrDyn *d = scr_dyn_new_promise(p);
+  scr_promise_release(p);
+  return d;
+}
+
+static ScrDyn *iter_method(ScrStream *s, ScrDynThunk thunk, const char *name) {
+  ScrClosure *clo = scr_closure_new((void *)thunk, 1);
+  clo->caps[0] = scr_box_new_obj(&scr_stream_retain_v, &scr_stream_release_v, &scr_stream_trace);
+  scr_box_set_ref(clo->caps[0], scr_stream_retain(s));
+  return scr_dyn_new_func(clo, thunk, 0, "()", name);
+}
+
+ScrDyn *scr_stream_dyn_async_iterator_open(ScrDyn *value, ScrStr *spell) {
+  ScrStream *s = scr_stream_native_readable_of_dyn(value);
+  if (!s) return scr_dyn_async_iterator_open(value, spell);
+  /* the loop awaits its results with the async protocol */
+  scr_dyn_async_iterator_set_sync(false);
+  ScrDyn *it = scr_dyn_new_obj();
+  scr_dyn_obj_set(it, "next", 4, iter_method(s, &iter_next_thunk, "next"));
+  scr_dyn_obj_set(it, "return", 6, iter_method(s, &iter_return_thunk, "return"));
+  scr_stream_release(s);
+  return it;
+}

packages/runtime/src/scr_stream.c

@@ -145,9 +145,12 @@ struct ScrStreamState {
     bool object_mode_opt;
     /* The parked for-await next() promise (at most one — the loop awaits
      * each chunk before asking again). next_dyn picks the resolution
-     * shape: a dyn boxed by tag (the JS lane), or bytes. */
+     * shape: a dyn boxed by tag (the JS lane), or bytes; next_result
+     * wraps the dyn in an iterator result ({ value, done }) for the
+     * checked-dynamic for-await protocol. */
     ScrPromise *next_waiter;
     bool next_dyn;
+    bool next_result;
     /* The iterator's own read() is calling _read: synchronous pushes
      * buffer, and the waiter takes the whole buffer afterwards (Node's
      * read() returns everything _read pushed). */
@@ -1226,7 +1229,15 @@ static void scr_stream_next_fulfill(ScrStreamState *st, ScrPromise *w /*moves*/,
     scr_promise_release(w);
     return;
   }
-  if (dyn) {
+  if (dyn && st->r.next_result) {
+    /* { value, done }: the value, or undefined at the end */
+    ScrDyn *result = scr_dyn_new_obj();
+    scr_dyn_obj_set(result, "value", 5,
+                    chunk == NULL ? scr_dyn_retain(scr_dyn_undefined())
+                                  : scr_stream_entry_box(chunk, kind));
+    scr_dyn_obj_set(result, "done", 4, scr_dyn_new_bool(chunk == NULL));
+    payload = result;
+  } else if (dyn) {
     payload = chunk == NULL ? scr_stream_from_done() : scr_stream_entry_box(chunk, kind);
   } else {
     payload = chunk ? (ScrBytes *)chunk : scr_bytes_new(SCR_BYTES_U8, 0);
@@ -1352,11 +1363,12 @@ static void scr_stream_settle_next(ScrStream *s) {
   }
 }
 
-static ScrPromise *scr_stream_next_chunk_impl(ScrStream *s, bool dyn) {
+static ScrPromise *scr_stream_next_chunk_impl(ScrStream *s, bool dyn, bool result) {
   ScrStreamState *st = s->st;
   ScrPromise *p = scr_promise_new();
   if (!st->has_r) {
     st->r.next_dyn = dyn;
+    st->r.next_result = result;
     scr_stream_next_fulfill(st, scr_promise_retain(p), NULL, SCR_ENTRY_BYTES);
     return p;
   }
@@ -1368,6 +1380,7 @@ static ScrPromise *scr_stream_next_chunk_impl(ScrStream *s, bool dyn) {
     return p;
   }
   st->r.next_dyn = dyn;
+  st->r.next_result = result;
   st->r.next_waiter = scr_promise_retain(p);
   st->r.next_parked = false;
   if (!st->r.next_started) {
@@ -1391,11 +1404,15 @@ ScrPromise *scr_stream_next_chunk(ScrStream *s) {
     scr_promise_reject_pending(p);
     return p;
   }
-  return scr_stream_next_chunk_impl(s, false);
+  return scr_stream_next_chunk_impl(s, false, false);
 }
 
 ScrPromise *scr_stream_next_chunk_dyn(ScrStream *s) {
-  return scr_stream_next_chunk_impl(s, true);
+  return scr_stream_next_chunk_impl(s, true, false);
+}
+
+ScrPromise *scr_stream_next_result_dyn(ScrStream *s) {
+  return scr_stream_next_chunk_impl(s, true, true);
 }
 
 /* ── Readable.from (array, string and Buffer sources) ─────────────────── */
@@ -7524,6 +7541,18 @@ static ScrStream *scr_stream_boxed(const ScrDyn *value) {
   return NULL;
 }
 
+/* The native readable stream behind a checked-dynamic value (+1): a
+ * boxed stream or a Cipheriv/Decipheriv handle's Transform, or NULL. */
+ScrStream *scr_stream_native_readable_of_dyn(const ScrDyn *value) {
+  ScrStream *s = scr_stream_of_handle(value);
+  if (!s) s = scr_stream_boxed(value);
+  if (s && !s->st->has_r) {
+    scr_stream_release(s);
+    return NULL;
+  }
+  return s;
+}
+
 /* stream.handleStream: the stream of a value typed as a stream handle
  * class (a Cipheriv or Decipheriv), +1. A value of that type can also be
  * the stream itself, boxed (the stream addAbortSignal answers). */

packages/runtime/src/scr_stream.h

@@ -301,6 +301,14 @@ bool scr_stream_write_dyn(ScrStream *s, const ScrDyn *d, ScrClosure *cb); /* cb
 bool scr_stream_write_enc(ScrStream *s, const ScrDyn *chunk, ScrDyn *enc, ScrClosure *cb);
 ScrPromise *scr_stream_next_chunk(ScrStream *s);
 ScrPromise *scr_stream_next_chunk_dyn(ScrStream *s);
+/* The same step resolving with an iterator result ({ value, done }). */
+ScrPromise *scr_stream_next_result_dyn(ScrStream *s);
+/* The native readable stream behind a checked-dynamic value (+1), or NULL. */
+ScrStream *scr_stream_native_readable_of_dyn(const struct ScrDyn *value);
+/* for await over a checked-dynamic value in a program with streams
+ * (scr_stream_ops.c): a native readable stream iterates as a typed loop
+ * does; any other value takes scr_dyn_async_iterator_open. */
+struct ScrDyn *scr_stream_dyn_async_iterator_open(struct ScrDyn *value, ScrStr *spell);
 /* node:stream/promises — the promise forms over the finished/pipeline
  * machinery above: a pending void promise the terminal watcher settles
  * (fulfilled on a clean finish, rejected with the finish status

tests/corpus/7420-for-await-close.ts

@@ -0,0 +1,100 @@
+// Leaving a for await loop over an async generator early closes it, as
+// AsyncIteratorClose does: return, throw and break inside the body, a
+// labeled continue to an outer loop, and a return() at a yield in the
+// body all await the generator's return(), so its finally runs before the
+// loop's caller continues. An exhausted generator and a rejected next()
+// are not closed. After a throw, an error from closing is dropped; after
+// a break or return it propagates.
+const log: string[] = [];
+
+async function* numbers(tag: string, failClose = false): AsyncGenerator<number> {
+  try {
+    yield 1;
+    yield 2;
+    yield 3;
+  } finally {
+    log.push(`${tag} finally`);
+    if (failClose) throw new Error(`${tag} close failed`);
+  }
+}
+
+async function* slowClose(tag: string): AsyncGenerator<number> {
+  try {
+    yield 1;
+  } finally {
+    await null;
+    log.push(`${tag} finally after await`);
+  }
+}
+
+async function firstOf(tag: string): Promise<number> {
+  for await (const n of numbers(tag)) return n * 10;
+  return -1;
+}
+
+async function throwsInside(tag: string): Promise<void> {
+  for await (const n of numbers(tag)) throw new Error(`${tag} body ${n}`);
+}
+
+async function* takeOne(tag: string): AsyncGenerator<number> {
+  for await (const n of numbers(tag)) {
+    yield n;
+    return;
+  }
+}
+
+async function* rethrowing(tag: string): AsyncGenerator<string> {
+  for await (const n of numbers(tag)) {
+    if (n === 2) throw new Error(`${tag} at ${n}`);
+    yield `${tag} ${n}`;
+  }
+}
+
+async function main(): Promise<void> {
+  log.push(`first ${await firstOf("return")}`);
+  try {
+    await throwsInside("throw");
+  } catch (e) {
+    log.push(`caught ${(e as Error).message}`);
+  }
+  for await (const n of takeOne("generator-return")) log.push(`took ${n}`);
+  try {
+    for await (const s of rethrowing("generator-throw")) log.push(s);
+  } catch (e) {
+    log.push(`caught ${(e as Error).message}`);
+  }
+  const early = takeOne("consumer-return");
+  for await (const n of early) {
+    log.push(`consumer got ${n}`);
+    break;
+  }
+  try {
+    for await (const n of numbers("break-close-fails", true)) {
+      log.push(`break at ${n}`);
+      break;
+    }
+  } catch (e) {
+    log.push(`caught ${(e as Error).message}`);
+  }
+  try {
+    for await (const n of numbers("throw-close-fails", true)) throw new Error(`body ${n}`);
+  } catch (e) {
+    log.push(`caught ${(e as Error).message}`);
+  }
+  void Promise.resolve().then(() => log.push("microtask"));
+  for await (const n of slowClose("slow")) {
+    log.push(`slow ${n}`);
+    break;
+  }
+  log.push("after slow");
+  outer: for (const k of [1, 2]) {
+    for await (const n of numbers(`labeled ${k}`)) {
+      log.push(`labeled ${k} got ${n}`);
+      continue outer;
+    }
+  }
+  for await (const n of numbers("exhausted")) log.push(`exhausted ${n}`);
+  for (const line of log) console.log(line);
+}
+
+void main();

tests/corpus/7421-for-await-close.mjs

@@ -0,0 +1,53 @@
+// The same early exits in JavaScript: a for await loop that returns,
+// throws or breaks closes its async generator, including one that reaches
+// the loop as an untyped parameter, and a source whose return() is still
+// running finishes before the code after the loop.
+const log = [];
+
+async function* letters(tag) {
+  try {
+    yield "a";
+    yield "b";
+  } finally {
+    log.push(`${tag} finally`);
+  }
+}
+
+async function firstLetter(source) {
+  for await (const c of source) return c;
+}
+
+async function failOn(source, target) {
+  for await (const c of source) {
+    if (c === target) throw new Error(`failed on ${c}`);
+    log.push(`kept ${c}`);
+  }
+}
+
+async function* upperUntil(source, stop) {
+  for await (const c of source) {
+    if (c === stop) return;
+    yield c.toUpperCase();
+  }
+}
+
+log.push(`first ${await firstLetter(letters("param-return"))}`);
+try {
+  await failOn(letters("param-throw"), "b");
+} catch (e) {
+  log.push(`caught ${e.message}`);
+}
+for await (const c of upperUntil(letters("stage-return"), "b")) log.push(`upper ${c}`);
+let ticks = 0;
+const tick = () => {
+  log.push(`tick ${ticks}`);
+  if (++ticks < 4) process.nextTick(tick);
+};
+process.nextTick(tick);
+for await (const c of letters("break")) {
+  log.push(`break ${c}`);
+  break;
+}
+log.push("after break");
+await new Promise((resolve) => setTimeout(resolve, 1));
+console.log(log.join("\n"));

tests/corpus/7422-for-await-untyped-streams.mjs

@@ -0,0 +1,38 @@
+// for await over a native stream that reaches the loop as an untyped
+// value iterates it as a typed loop does, in the same tick order as Node:
+// whole reads, an early return (the stream is destroyed), a break, an
+// object-mode Transform and a stream error.
+import { Readable, PassThrough, Transform } from "node:stream";
+const log = [];
+async function readAll(s, tag) {
+  for await (const c of s) log.push(`${tag} ${typeof c} ${String(c)}`);
+  log.push(`${tag} done`);
+}
+async function first(s, tag) {
+  for await (const c of s) return `${tag} first ${String(c)}`;
+}
+async function breaks(s, tag) {
+  for await (const c of s) { log.push(`${tag} got ${String(c)}`); break; }
+}
+async function failing(s, tag) {
+  try { for await (const c of s) log.push(`${tag} ${c}`); } catch (e) { log.push(`${tag} caught ${e.message}`); }
+}
+let n = 0;
+const ticker = () => { log.push(`tick ${n}`); if (++n < 6) process.nextTick(ticker); };
+process.nextTick(ticker);
+await readAll(Readable.from(["a", "b"]), "arr");
+const r = Readable.from(["x", "y", "z"]);
+r.on("close", () => log.push(`r close ${r.destroyed}`));
+log.push(await first(r, "r"));
+const pt = new PassThrough();
+pt.on("close", () => log.push("pt close"));
+pt.end("bytes");
+await breaks(pt, "pt");
+const t = new Transform({ objectMode: true, transform(c, e, cb) { cb(null, { v: c }); } });
+t.end(1);
+await (async (s) => { for await (const c of s) log.push(`t ${JSON.stringify(c)}`); })(t);
+const bad = new Readable({ read() { this.destroy(new Error("boom")); } });
+await failing(bad, "bad");
+await readAll(["plain", "array"], "dyn");
+await new Promise((res) => setTimeout(res, 5));
+console.log(log.join("\n"));

tests/corpus/7423-stream-untyped-bindings.mjs

@@ -0,0 +1,33 @@
+// Streams held where the lowering keeps a checked-dynamic value: a
+// module-level binding of a stream method's result on a value that a
+// JavaScript function returns, an element read of an array of streams,
+// and readable.compose on a stream whose function is typed with JSDoc.
+import { PassThrough, Readable, Transform } from "node:stream";
+
+const words = () => Readable.from(["alpha", "beta"]);
+const shouted = words().map((w) => w.toUpperCase());
+console.log("shouted", shouted.readable, shouted.readableObjectMode);
+for await (const w of shouted) console.log("shouted", w);
+
+const counted = words().filter((w) => w.length > 4);
+counted.on("data", (w) => console.log("counted", w));
+await new Promise((resolve) => counted.on("end", resolve));
+console.log("counted ended", counted.readableEnded);
+
+const streams = [new PassThrough({ objectMode: true }), new Transform({
+  objectMode: true,
+  transform(chunk, _enc, cb) {
+    cb(null, { wrapped: chunk });
+  },
+})];
+streams[1].end(7);
+for await (const v of streams[1]) console.log("element", JSON.stringify(v));
+streams[0].end("done");
+for await (const v of streams[0]) console.log("element", v);
+
+/** @returns {Readable} */
+const typedWords = () => Readable.from(["x", "yy"]);
+const lengths = typedWords().compose(async function* (source) {
+  for await (const w of source) yield `${w}:${w.length}`;
+});
+console.log("composed", (await lengths.toArray()).join(","));

tests/harness/corpus-costs/7420-for-await-close.ts.json

@@ -0,0 +1 @@
+{ "plain": 2100, "san": 2800 }

tests/harness/corpus-costs/7421-for-await-close.mjs.json

@@ -0,0 +1 @@
+{ "plain": 1900, "san": 2600 }

tests/harness/corpus-costs/7422-for-await-untyped-streams.mjs.json

@@ -0,0 +1 @@
+{ "plain": 1900, "san": 2600 }

tests/harness/corpus-costs/7423-stream-untyped-bindings.mjs.json

@@ -0,0 +1 @@
+{ "plain": 4900, "san": 6454 }

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: 28, marker: "scr_runtime_abi_v28" },
+        runtime_abi: { version: 29, marker: "scr_runtime_abi_v29" },
         compiler: { command: "fixture", identity: "fixture", target: target.llvmTriple },
         macros: { executable: [], excluded: [], sanitizer: "external-toolchain-required" },
         flavors: {