Support Duplex.from over streams, pairs, promises and async functions

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

@@ -22,8 +22,9 @@
   `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.
   `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.
-  A lone `Readable` or `Writable`, a promise, a `{ readable, writable }` pair, a web stream, and a body function that declares the options argument are unsupported in `Duplex.from`.
-  `stream.compose` accepts streams and async generator functions in any position, and an array, a string, or a generator as the first stage; one stage is `Duplex.from` of it.
+  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`.
+  `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.
   Web streams as compose stages are unsupported, and the `ERR_INVALID_ARG_VALUE` message for a stream that is no longer readable or writable names the stream's class instead of printing the stream.
   `readable.compose(stage, { signal })` is `stream.compose(readable, stage)` with the signal added; the options are an object literal whose `signal` is a variable or a property read.
   `pipeline` from `node:stream/promises` accepts a trailing `{ signal, end }` object, with or without function stages, and `finished` accepts `readable`, `writable`, `error`, `cleanup`, and `signal` options, with boolean literals for the flags.

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

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

@@ -1129,6 +1129,11 @@ export const LIB_FN_SYMS: Record<string, string> = {
   "stream.decodeStrings": "scr_stream_decode_strings",
   "stream.duplexFromGen": "scr_stream_duplex_from_gen",
   "stream.duplexFromDyn": "scr_stream_duplex_from_dyn",
+  "stream.duplexifyReadable": "scr_stream_duplexify_readable",
+  "stream.duplexifyWritable": "scr_stream_duplexify_writable",
+  "stream.duplexifyPair": "scr_stream_duplexify_pair",
+  "stream.duplexFromPromise": "scr_stream_duplex_from_promise",
+  "stream.duplexFromAsyncFn": "scr_stream_duplex_from_async_fn",
   "writable.writeEncoding": "scr_stream_write_encoding",
   "writable.setDefaultEncoding": "scr_stream_set_default_encoding",
   "stream.chunkDone": "scr_stream_chunk_done",
@@ -1737,6 +1742,11 @@ export const USES_TIMERS_LIB_FNS = new Set<string>([
   "stream.setConstruct",
   "stream.duplexFromGen",
   "stream.duplexFromDyn",
+  "stream.duplexifyReadable",
+  "stream.duplexifyWritable",
+  "stream.duplexifyPair",
+  "stream.duplexFromPromise",
+  "stream.duplexFromAsyncFn",
   "stream.setWritev",
   "stream.setWritevDefault",
   "process.activeResources",

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

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

@@ -3772,11 +3772,11 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE
   if (fn !== null) {
     const isAsync = (fn.modifiers?.nodes ?? []).some((m) => m.kind === ts.SyntaxKind.AsyncKeyword);
     const isGen = !ts.isArrowFunction(fn) && fn.asteriskToken !== undefined;
-    if (!isAsync || !isGen) {
+    if (!isAsync) {
       lowerer.noLowering(
-        "Duplex.from over a function that is not an async generator function",
+        "Duplex.from over a function that is not async",
         node,
-        "write the body as async function* (source) { for await (const chunk of source) ... }",
+        "write the body as async function* (source) { ... } or async function (source) { ... }",
       );
     }
     if (fn.parameters.nodes.length > 1) {
@@ -3794,6 +3794,34 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE
       nextT: UNDEFINED_T,
     };
     const lowered = lowerStageFunction(lowerer, fn, sourceT);
+    if (!isGen) {
+      // an async function: a writable Duplexify whose final waits for
+      // the function's promise
+      const asyncSource = duplexSourceHelper(lowerer, sourceT, loc);
+      const out = lowered.out;
+      if (out.kind !== "promise" || !asyncSource)
+        return lowerer.noLowering("Duplex.from over a function with this return type", node);
+      if (
+        out.inner.kind !== "void" &&
+        !canConvertToDyn(
+          out.inner,
+          (id) => lowerer.shapes.get(id),
+          (id) => lowerer.unions.get(id),
+        )
+      ) {
+        lowerer.noLowering(
+          `Duplex.from over a function resolving to '${lowerer.fmt(out.inner)}'`,
+          node,
+        );
+      }
+      return {
+        kind: "libCall",
+        fn: "stream.duplexFromAsyncFn",
+        args: [lowered.closure, asyncSource, pipelineBoxHelper(lowerer, out, loc)],
+        type: duplex,
+        loc,
+      };
+    }
     const helpers =
       lowered.out.kind === "generator" ? fromGenHelpers(lowerer, lowered.out, loc) : null;
     const source = duplexSourceHelper(lowerer, sourceT, loc);
@@ -3818,6 +3846,45 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE
   }
   const bodySides =
     bodyT?.kind === "object" ? streamSidesOf(lowerer, lowerer.classes.get(bodyT.className)) : null;
+  let literal: ts.Expression = node;
+  while (ts.isParenthesizedExpression(literal)) literal = literal.expression;
+  if (bodySides === null && ts.isObjectLiteralExpression(literal))
+    return duplexFromPair(lowerer, literal, loc);
+  if (bodyT?.kind === "promise") {
+    // a Duplexify that reads the promise's value
+    const valueT = bodyT.inner;
+    if (
+      valueT.kind !== "void" &&
+      !canConvertToDyn(
+        valueT,
+        (id) => lowerer.shapes.get(id),
+        (id) => lowerer.unions.get(id),
+      )
+    ) {
+      lowerer.noLowering(`Duplex.from over a promise of '${lowerer.fmt(valueT)}'`, node);
+    }
+    const promise = lowerer.lowerExpr(node);
+    if (promise.type.kind !== "promise") {
+      // a checked-dynamic promise (an untyped function's result)
+      return {
+        kind: "libCall",
+        fn: "stream.duplexFromDyn",
+        args: [lowerer.coerceToExpected(promise, DYN)],
+        type: { kind: "object", className: "%Duplex" },
+        loc,
+      };
+    }
+    return {
+      kind: "libCall",
+      fn: "stream.duplexFromPromise",
+      args: [
+        promise,
+        pipelineBoxHelper(lowerer, promise.type as IrType & { kind: "promise" }, loc),
+      ],
+      type: { kind: "object", className: "%Duplex" },
+      loc,
+    };
+  }
   if (bodySides === null) return lowerFromSource(lowerer, node, node, undefined, true);
   let stream: IrExpr;
   try {
@@ -3830,16 +3897,88 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE
     stream.type.kind === "object"
       ? streamSidesOf(lowerer, lowerer.classes.get(stream.type.className))
       : null;
+  // _duplexify({ readable }) / _duplexify({ writable })
+  if (sides === "r" || sides === "w") {
+    return {
+      kind: "libCall",
+      fn: sides === "r" ? "stream.duplexifyReadable" : "stream.duplexifyWritable",
+      args: [stream],
+      type: duplex,
+      loc,
+    };
+  }
   if (sides !== "rw") {
     lowerer.noLowering(
-      `Duplex.from over a ${sides === "r" ? "Readable" : sides === "w" ? "Writable" : "value of this type"}`,
+      "Duplex.from over a value of this type",
       node,
-      "an async generator function, an array, a string, a generator, an async generator, or a Duplex, Transform or PassThrough is the lowered body",
+      "an async generator function, an iterable, a stream, or a { readable, writable } pair is the lowered body",
     );
   }
   return stream;
 }
 
+/** Duplex.from({ readable, writable }): each member that is a stream with
+ * that side is used as it is, any other member is Duplex.from of it, and
+ * the pair reads and writes through _duplexify (duplexify.js). The
+ * members evaluate in source order. */
+function duplexFromPair(
+  lowerer: Lowerer,
+  literal: ts.ObjectLiteralExpression,
+  loc: SrcLoc,
+): IrExpr {
+  const duplex: IrType = { kind: "object", className: "%Duplex" };
+  const sides: { readable?: IrExpr; writable?: IrExpr } = {};
+  for (const p of literal.properties.nodes) {
+    const key =
+      (ts.isPropertyAssignment(p) || ts.isShorthandPropertyAssignment(p)) && ts.isIdentifier(p.name)
+        ? p.name.text
+        : null;
+    if (key !== "readable" && key !== "writable") {
+      lowerer.noLowering(
+        "Duplex.from over an object with members other than readable and writable",
+        p,
+        "pass { readable, writable } with streams or Duplex.from bodies",
+      );
+    }
+    const value = ts.isShorthandPropertyAssignment(p)
+      ? (p.name as ts.Identifier)
+      : (p as ts.PropertyAssignment).initializer;
+    let valueT: IrType | null;
+    try {
+      valueT = lowerer.mapTypeOf(lowerer.typeOf(value));
+    } catch {
+      valueT = null;
+    }
+    const valueSides =
+      valueT?.kind === "object"
+        ? streamSidesOf(lowerer, lowerer.classes.get(valueT.className))
+        : null;
+    const wanted = key === "readable" ? "r" : "w";
+    const direct = valueSides === "rw" || valueSides === wanted;
+    sides[key!] = direct
+      ? lowerStreamArg(lowerer, value, "compose")
+      : duplexFromBody(lowerer, value, loc);
+  }
+  const { readable, writable } = sides;
+  if (readable && writable)
+    return {
+      kind: "libCall",
+      fn: "stream.duplexifyPair",
+      args: [readable, writable],
+      type: duplex,
+      loc,
+    };
+  if (readable)
+    return { kind: "libCall", fn: "stream.duplexifyReadable", args: [readable], type: duplex, loc };
+  if (writable)
+    return { kind: "libCall", fn: "stream.duplexifyWritable", args: [writable], type: duplex, loc };
+  return lowerer.noLowering(
+    "Duplex.from over an object with neither readable nor writable",
+    literal,
+    "pass { readable, writable } with streams or Duplex.from bodies",
+  );
+}
+
 /** The checker type of a stage's string or Buffer chunks, or null for
  * other chunk types. */
 function stageChunkCheckerType(lowerer: Lowerer, chunk: IrType, at: ts.Node): ts.Type | null {

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

@@ -384,6 +384,8 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "writable.setDefaultEncoding",
   "stream.duplexFromGen",
   "stream.duplexFromDyn",
+  "stream.duplexFromPromise",
+  "stream.duplexFromAsyncFn",
   "stream.destroy",
   "stream.destroyErr",
   "stream.iteratorClose",

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

@@ -4069,6 +4069,16 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
   "stream.decodeStrings": { argTypes: [null, BOOL], result: VOID },
   "stream.duplexFromGen": { argTypes: [null, null, null, null, null, null], result: VOID },
   "stream.duplexFromDyn": { argTypes: [DYN], result: VOID },
+  /** Duplex.from over a lone Readable or Writable, or a { readable,
+   * writable } pair of streams (duplexify.js's _duplexify). */
+  "stream.duplexifyReadable": { argTypes: [null], result: VOID },
+  "stream.duplexifyWritable": { argTypes: [null], result: VOID },
+  "stream.duplexifyPair": { argTypes: [null, null], result: VOID },
+  /** Duplex.from(promise): the promise and its box helper. */
+  "stream.duplexFromPromise": { argTypes: [null, null], result: VOID },
+  /** Duplex.from(asyncFunction): the stage, its source helper and the
+   * promise's box helper. */
+  "stream.duplexFromAsyncFn": { argTypes: [null, null, null], 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

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

@@ -221,6 +221,11 @@ const STREAM_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "stream.decodeStrings",
   "stream.duplexFromGen",
   "stream.duplexFromDyn",
+  "stream.duplexFromPromise",
+  "stream.duplexFromAsyncFn",
+  "stream.duplexifyReadable",
+  "stream.duplexifyWritable",
+  "stream.duplexifyPair",
   "console.bind",
   "crypto.hashStream",
   "crypto.hashStreamPiped",
@@ -277,6 +282,8 @@ const READABLE_DYN_ASYNC_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "readable.fromIter",
   "stream.duplexFromGen",
   "stream.duplexFromDyn",
+  "stream.duplexFromPromise",
+  "stream.duplexFromAsyncFn",
 ]);
 
 /** libCall name prefixes of the Readable entry points that await through the dyn-async unit. */

packages/compiler/src/ir/validate.ts

@@ -5499,6 +5499,11 @@ function validateFunction(
       e.fn === "stream.decodeStrings" ||
       e.fn === "stream.duplexFromGen" ||
       e.fn === "stream.duplexFromDyn" ||
+      e.fn === "stream.duplexifyReadable" ||
+      e.fn === "stream.duplexifyWritable" ||
+      e.fn === "stream.duplexifyPair" ||
+      e.fn === "stream.duplexFromPromise" ||
+      e.fn === "stream.duplexFromAsyncFn" ||
       e.fn === "stream.setConstruct" ||
       e.fn === "stream.setWritev" ||
       e.fn === "stream.setWritevDefault" ||
@@ -5554,7 +5559,30 @@ function validateFunction(
         }
         return true;
       }
-      if (e.fn === "stream.duplexFromGen" || e.fn === "stream.duplexFromDyn") {
+      if (e.fn === "stream.duplexFromPromise") {
+        // Duplex.from(promise): (promise, box) → a Duplex.
+        if (e.args[0]?.type.kind !== "promise" || e.args[1]?.type.kind !== "func")
+          err(`libCall ${e.fn} takes a promise and its box helper`, e.loc);
+        if (!isStreamObject(e.type)) err(`libCall ${e.fn} must return a stream class`, e.loc);
+        return true;
+      }
+      if (
+        e.fn === "stream.duplexifyReadable" ||
+        e.fn === "stream.duplexifyWritable" ||
+        e.fn === "stream.duplexifyPair"
+      ) {
+        // Duplex.from over streams: (r), (w) or (r, w) → a Duplex.
+        const want = e.fn === "stream.duplexifyPair" ? 2 : 1;
+        if (e.args.length !== want || !e.args.every((a) => isStreamObject(a.type)))
+          err(`libCall ${e.fn} takes ${want} stream argument(s)`, e.loc);
+        if (!isStreamObject(e.type)) err(`libCall ${e.fn} must return a stream class`, e.loc);
+        return true;
+      }
+      if (
+        e.fn === "stream.duplexFromGen" ||
+        e.fn === "stream.duplexFromDyn" ||
+        e.fn === "stream.duplexFromAsyncFn"
+      ) {
         // Duplex.from: (stage, source, helpers...) or (body) → a Duplex.
         if (!isStreamObject(e.type)) {
           err(`libCall ${e.fn} must return a stream class, got ${e.type.kind}`, e.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_v29");
+    expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v30");
 
     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_v29");
+    expect(undefinedSymbols).toContain("_scr_runtime_abi_v30");
   });
 
   test("helper and clang object paths retain the same outbound FFI C ABI", async () => {

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

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

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

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

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

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

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7427-duplex-from-async-function.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7427-duplex-from-async-function.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: 29, marker: "scr_runtime_abi_v29" },
+      runtime_abi: { version: 30, marker: "scr_runtime_abi_v30" },
       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: 29, marker: "scr_runtime_abi_v29" },
+      runtime_abi: { version: 30, marker: "scr_runtime_abi_v30" },
       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_v29(void) {}
+void scr_runtime_abi_v30(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_v29(void);
+void scr_runtime_abi_v30(void);
 void scr_runtime_workers_v8(void);
 
 /* ── the trap funnel (scr_console.c; scr_library.c under -DSCR_LIB) ──────

packages/runtime/src/scr_stream.c

@@ -1769,18 +1769,26 @@ ScrStream *scr_stream_from_dyn(ScrDyn *src, ScrDyn *opts) {
 }
 
 /* Duplex.from(body) over a checked-dynamic body (duplexify.js): a native
- * Duplex answers itself, an iterable reads as Duplex.from does, and the
- * other bodies Node accepts (a lone Readable or Writable, a function, a
- * promise, a { readable, writable } pair) are not supported yet. */
+ * Duplex answers itself, a lone native Readable or Writable is wrapped, a
+ * promise reads as its value, an iterable reads as Duplex.from does, and
+ * the other bodies Node accepts (a function, a thenable, a { readable,
+ * writable } pair) are not supported yet. */
 static ScrStream *scr_stream_boxed(const ScrDyn *value);
+static ScrStream *scr_dupx_from_boxed(ScrDyn *boxed);
 
 ScrStream *scr_stream_duplex_from_dyn(ScrDyn *body) {
   ScrStream *native = scr_stream_of_handle(body); /* +1 */
   if (!native) native = scr_stream_boxed(body);   /* +1 */
-  if (native && native->st->has_r && native->st->has_w) return native;
-  bool unsupported = native != NULL;
-  if (native) scr_stream_release(native);
-  if (!unsupported && (body->kind == SCR_DYN_FUNC || body->kind == SCR_DYN_PROMISE)) unsupported = true;
+  if (native) {
+    if (native->st->has_r && native->st->has_w) return native;
+    /* _duplexify({ readable }) / _duplexify({ writable }) */
+    ScrStream *d = native->st->has_r ? scr_stream_duplexify(native, NULL)
+                                     : scr_stream_duplexify(NULL, native);
+    scr_stream_release(native);
+    return d;
+  }
+  if (body->kind == SCR_DYN_PROMISE) return scr_dupx_from_boxed(scr_dyn_retain(body));
+  bool unsupported = body->kind == SCR_DYN_FUNC;
   if (!unsupported && body->kind == SCR_DYN_OBJ) {
     static const char *const keys[] = {"then", "readable", "writable"};
     for (size_t i = 0; i < 3 && !unsupported; i++) {
@@ -4709,6 +4717,344 @@ ScrStream *scr_stream_compose_stages(double n, ScrStream **streams, double count
   return scr_stream_compose_impl((size_t)n, streams, &fs, abort);
 }
 
+/* ── Duplex.from over a stream or a { readable, writable } pair ─────────
+ * duplexify.js's _duplexify({ readable, writable }): a Duplexify that
+ * writes to w (write() answering false waits for 'drain'; final ends w and
+ * waits for its 'finish') and reads r through 'readable' and read(), with
+ * each side's end of stream (eos) finishing the Duplex. The readable and
+ * writable object modes follow r's and w's. */
+
+enum { SCR_DUPX_UNSET, SCR_DUPX_NULL, SCR_DUPX_WAIT };
+
+typedef struct ScrDuplexify {
+  size_t rc;
+  ScrStream *r, *w;  /* +1 or NULL */
+  ScrStream *d;      /* +1 until teardown */
+  ScrClosure *drain_l, *finish_l, *readable_l, *end_l; /* +1, the listeners, or NULL */
+  bool readable, writable;
+  bool ondrain, onfinish, onreadable;
+  int onclose; /* _duplexify's onclose: undefined, null, or d's destroy callback */
+  struct ScrDuplexify *live_prev, *live_next;
+} ScrDuplexify;
+
+static SCR_TL ScrDuplexify *scr_dupx_live = NULL;
+
+static void *scr_dupx_retain_v(void *p) {
+  ((ScrDuplexify *)p)->rc++;
+  return p;
+}
+
+static void scr_dupx_release_v(void *p) {
+  ScrDuplexify *x = p;
+  if (--x->rc > 0) return;
+  if (x->r) scr_stream_release(x->r);
+  if (x->w) scr_stream_release(x->w);
+  free(x);
+}
+
+static ScrClosure *scr_dupx_closure(void *fn, ScrDuplexify *x) {
+  ScrClosure *cl = scr_closure_new(fn, 1);
+  cl->caps[0] = scr_box_new_obj(&scr_dupx_retain_v, &scr_dupx_release_v, NULL);
+  scr_box_set_ref(cl->caps[0], scr_dupx_retain_v(x));
+  return cl;
+}
+
+static void scr_dupx_on(ScrStream *s, const char *name, ScrClosure **slot, void *fn, ScrDuplexify *x) {
+  ScrStr *n = scr_str_new(name, strlen(name));
+  *slot = scr_dupx_closure(fn, x);
+  scr_emitter_release(
+      scr_emitter_on((ScrEmitter *)s, n, scr_closure_retain(*slot), scr_ee_inv_fixed0, false, false));
+  scr_str_release(n);
+}
+
+/* Drop the Duplex and the listeners on r and w. */
+static void scr_dupx_teardown(ScrDuplexify *x) {
+  if (!x->d) return;
+  if (x->w) {
+    scr_compose_off(x->w, "drain", &x->drain_l);
+    scr_compose_off(x->w, "finish", &x->finish_l);
+  }
+  if (x->r) {
+    scr_compose_off(x->r, "readable", &x->readable_l);
+    scr_compose_off(x->r, "end", &x->end_l);
+  }
+  if (x->live_prev) x->live_prev->live_next = x->live_next;
+  else scr_dupx_live = x->live_next;
+  if (x->live_next) x->live_next->live_prev = x->live_prev;
+  x->live_prev = x->live_next = NULL;
+  ScrStream *d = x->d;
+  x->d = NULL;
+  scr_stream_release(d);
+  scr_dupx_release_v(x); /* the live list's reference */
+}
+
+static void scr_dupx_atexit(void) {
+  while (scr_dupx_live) scr_dupx_teardown(scr_dupx_live);
+}
+
+/* d's destroy callback: callback(err), then the context lets go. */
+static void scr_dupx_destroyed(ScrDuplexify *x, ScrError *err) {
+  ScrStream *d = scr_stream_retain(x->d);
+  scr_dupx_teardown(x);
+  scr_stream_destroy_done(d, err ? scr_error_retain(err) : NULL);
+  scr_stream_release(d);
+}
+
+/* onfinished(err): the pending destroy callback, else d.destroy(err) */
+static void scr_dupx_onfinished(ScrDuplexify *x, ScrError *err) {
+  int onclose = x->onclose;
+  x->onclose = SCR_DUPX_NULL;
+  if (!x->d) return;
+  if (onclose == SCR_DUPX_WAIT) {
+    scr_dupx_destroyed(x, err);
+  } else if (err) {
+    ScrStream *d = scr_stream_retain(x->d);
+    scr_stream_release(scr_stream_destroy(d, err));
+    scr_stream_release(d);
+  }
+}
+
+/* eos(w): writable = false; an error destroys r */
+static void scr_dupx_w_eos(ScrClosure *cb, ScrStream *s, ScrError *err) {
+  (void)s;
+  ScrDuplexify *x = scr_box_get_ref(cb->caps[0]);
+  x->writable = false;
+  if (err && x->r) scr_stream_release(scr_stream_destroy(x->r, err));
+  scr_dupx_onfinished(x, err);
+  scr_dupx_release_v(x);
+}
+
+/* eos(r): readable = false; an error destroys w */
+static void scr_dupx_r_eos(ScrClosure *cb, ScrStream *s, ScrError *err) {
+  (void)s;
+  ScrDuplexify *x = scr_box_get_ref(cb->caps[0]);
+  x->readable = false;
+  if (err && x->w) scr_stream_release(scr_stream_destroy(x->w, err));
+  scr_dupx_onfinished(x, err);
+  scr_dupx_release_v(x);
+}
+
+/* d._write: w.write(chunk) ? callback() : (ondrain = callback) */
+static void scr_dupx_write_inv(ScrClosure *cb, ScrStream *s, ScrBytes *chunk) {
+  ScrDuplexify *x = scr_box_get_ref(cb->caps[0]);
+  bool ok = scr_stream_writes_values(s) ? scr_stream_write_dyn(x->w, (ScrDyn *)chunk, NULL)
+                                        : scr_stream_write(x->w, chunk, NULL);
+  if (ok) {
+    if (!scr_exc_pending()) scr_stream_write_done(s, NULL);
+  } else {
+    x->ondrain = true;
+  }
+  scr_dupx_release_v(x);
+}
+
+/* d._final: w.end(); onfinish = callback */
+static void scr_dupx_final_inv(ScrClosure *cb, ScrStream *s) {
+  (void)s;
+  ScrDuplexify *x = scr_box_get_ref(cb->caps[0]);
+  ScrStream *r = scr_stream_end(x->w, NULL, NULL, NULL);
+  if (r) scr_stream_release(r);
+  x->onfinish = true;
+  scr_dupx_release_v(x);
+}
+
+static void scr_dupx_drain_l(ScrClosure *cb) {
+  ScrDuplexify *x = scr_box_get_ref(cb->caps[0]);
+  if (x->ondrain && x->d) {
+    x->ondrain = false;
+    scr_stream_write_done(x->d, NULL);
+  }
+  scr_dupx_release_v(x);
+}
+
+static void scr_dupx_finish_l(ScrClosure *cb) {
+  ScrDuplexify *x = scr_box_get_ref(cb->caps[0]);
+  if (x->onfinish && x->d) {
+    x->onfinish = false;
+    scr_stream_final_done(x->d, NULL);
+  }
+  scr_dupx_release_v(x);
+}
+
+/* d._read: push r.read() until it answers null (then wait for 'readable')
+ * or d is full */
+static void scr_dupx_pull(ScrDuplexify *x) {
+  ScrStream *d = x->d;
+  while (d && x->r) {
+    ScrEntryKind kind = SCR_ENTRY_BYTES;
+    void *chunk = scr_stream_read_n_kind(x->r, -1, &kind);
+    if (scr_exc_pending()) return;
+    if (!chunk) {
+      x->onreadable = true;
+      return;
+    }
+    bool more;
+    if (d->st->r.object_entries) {
+      more = scr_stream_add_entry(d, chunk, kind) > 0;
+    } else if (kind == SCR_ENTRY_STR) {
+      more = scr_stream_push_str(d, chunk);
+      scr_str_release(chunk);
+    } else if (kind == SCR_ENTRY_DYN) {
+      more = scr_stream_push_dyn(d, chunk);
+      scr_dyn_release(chunk);
+    } else {
+      more = scr_stream_push(d, chunk);
+      scr_bytes_release(chunk);
+    }
+    if (!more || scr_exc_pending()) return;
+  }
+}
+
+static void scr_dupx_read_inv(ScrClosure *cb, ScrStream *s, double size) {
+  (void)s;
+  (void)size;
+  ScrDuplexify *x = scr_box_get_ref(cb->caps[0]);
+  x->onreadable = false;
+  scr_dupx_pull(x);
+  scr_dupx_release_v(x);
+}
+
+static void scr_dupx_readable_l(ScrClosure *cb) {
+  ScrDuplexify *x = scr_box_get_ref(cb->caps[0]);
+  if (x->onreadable && x->d) {
+    x->onreadable = false;
+    scr_dupx_pull(x);
+  }
+  scr_dupx_release_v(x);
+}
+
+static void scr_dupx_end_l(ScrClosure *cb) {
+  ScrDuplexify *x = scr_box_get_ref(cb->caps[0]);
+  if (x->d) scr_stream_push_null(x->d);
+  scr_dupx_release_v(x);
+}
+
+/* d._destroy */
+static void scr_dupx_destroy_inv(ScrClosure *cb, ScrStream *s, ScrError *err) {
+  (void)s;
+  ScrDuplexify *x = scr_box_get_ref(cb->caps[0]);
+  ScrError *e = err ? scr_error_retain(err) : NULL;
+  if (!e && x->onclose != SCR_DUPX_NULL) e = scr_compose_abort_error();
+  x->onreadable = x->ondrain = x->onfinish = false;
+  if (x->onclose == SCR_DUPX_NULL) {
+    if (x->d) scr_dupx_destroyed(x, e);
+  } else {
+    x->onclose = SCR_DUPX_WAIT;
+    if (x->w && !x->w->st->destroyed) scr_stream_release(scr_stream_destroy(x->w, e));
+    if (x->r && !x->r->st->destroyed) scr_stream_release(scr_stream_destroy(x->r, e));
+  }
+  if (e) scr_error_release(e);
+  scr_dupx_release_v(x);
+}
+
+/* _duplexify({ readable: r, writable: w }): r and w borrowed, either NULL.
+ * +1. */
+ScrStream *scr_stream_duplexify(ScrStream *r, ScrStream *w) {
+  ScrDuplexify *x = calloc(1, sizeof *x);
+  if (!x) scr_stream_oom();
+  x->rc = 1;
+  x->r = r ? scr_stream_retain(r) : NULL;
+  x->w = w ? scr_stream_retain(w) : NULL;
+  x->onclose = SCR_DUPX_UNSET;
+  x->readable = r && scr_stream_prop(r, "readable") != 0;
+  x->writable = w && scr_stream_prop(w, "writable") != 0;
+  /* new Duplexify({ readableObjectMode, writableObjectMode, readable, writable }) */
+  bool r_object = r && r->st->has_r && r->st->r.object_entries;
+  bool w_object = w && w->st->has_w && w->st->w.object_mode;
+  ScrStream *d = scr_stream_new_duplex(
+      r_object ? 16 : -1, w_object ? 16 : -1, true, true, true, x->readable, x->writable,
+      x->readable ? scr_dupx_closure((void *)&scr_dupx_read_inv, x) : NULL,
+      x->readable ? &scr_dupx_read_inv : NULL,
+      x->writable ? scr_dupx_closure((void *)&scr_dupx_write_inv, x) : NULL,
+      x->writable ? &scr_dupx_write_inv : NULL,
+      x->writable ? scr_dupx_closure((void *)&scr_dupx_final_inv, x) : NULL,
+      x->writable ? &scr_dupx_final_inv : NULL,
+      scr_dupx_closure((void *)&scr_dupx_destroy_inv, x), &scr_dupx_destroy_inv);
+  if (r_object) d->st->r.object_entries = true;
+  if (w_object) scr_stream_set_object_mode(d, false, true);
+  d->cls = "Duplexify";
+  x->d = scr_stream_retain(d);
+  if (x->writable) {
+    scr_closure_release(
+        scr_stream_finished(w, scr_dupx_closure((void *)&scr_dupx_w_eos, x), &scr_dupx_w_eos));
+    scr_dupx_on(w, "drain", &x->drain_l, (void *)&scr_dupx_drain_l, x);
+    scr_dupx_on(w, "finish", &x->finish_l, (void *)&scr_dupx_finish_l, x);
+  }
+  if (x->readable) {
+    scr_closure_release(
+        scr_stream_finished(r, scr_dupx_closure((void *)&scr_dupx_r_eos, x), &scr_dupx_r_eos));
+    scr_dupx_on(r, "readable", &x->readable_l, (void *)&scr_dupx_readable_l, x);
+    scr_dupx_on(r, "end", &x->end_l, (void *)&scr_dupx_end_l, x);
+  }
+  static SCR_TL bool registered = false;
+  if (!registered) {
+    registered = true;
+    scr_atexit(scr_dupx_atexit);
+  }
+  x->live_next = scr_dupx_live;
+  if (scr_dupx_live) scr_dupx_live->live_prev = x;
+  scr_dupx_live = x; /* the reference x was created with */
+  return d;
+}
+
+/* Duplex.from(promise) (duplexify.js): a Duplexify({ objectMode: true,
+ * writable: false, read() {} }) that pushes the promise's value unless it
+ * is nullish, then ends; a rejection destroys it with the reason. */
+typedef struct {
+  ScrStream *d;  /* +1 */
+  ScrDyn *boxed; /* +1, the promise boxed */
+} ScrDupxPromise;
+
+static void scr_dupx_noop_read(ScrClosure *cb, ScrStream *s, double size) {
+  (void)cb;
+  (void)s;
+  (void)size;
+}
+
+static void scr_dupx_promise_entry(ScrFiber *fiber, void *arg) {
+  (void)fiber;
+  ScrDupxPromise p = *(ScrDupxPromise *)arg;
+  free(arg);
+  /* the boxed promise's payload is a dyn value (NULL: undefined) */
+  ScrDyn *v = scr_await_ref(p.boxed->v.promise);
+  if (!scr_exc_pending()) {
+    if (v && v->kind != SCR_DYN_NULL && v->kind != SCR_DYN_UNDEF) scr_stream_push_object(p.d, v);
+    if (!scr_exc_pending()) scr_stream_push_null(p.d);
+    if (scr_exc_pending()) scr_exc_clear();
+    if (v) scr_dyn_release(v);
+  } else {
+    ScrError *err = scr_pump_caught();
+    scr_stream_release(scr_stream_destroy(p.d, err));
+    if (scr_exc_pending()) scr_exc_clear();
+    scr_error_release(err);
+  }
+  scr_stream_release(p.d);
+  scr_dyn_release(p.boxed);
+}
+
+/* The readable Duplexify over a boxed promise (moves boxed). +1. */
+static ScrStream *scr_dupx_from_boxed(ScrDyn *boxed) {
+  ScrClosure *read = scr_closure_new((void *)&scr_dupx_noop_read, 0);
+  ScrStream *d = scr_stream_new_duplex(16, 16, true, true, true, true, false, read,
+                                       &scr_dupx_noop_read, NULL, NULL, NULL, NULL, NULL, NULL);
+  scr_stream_set_object_mode(d, true, true);
+  d->cls = "Duplexify";
+  ScrDupxPromise *ctx = malloc(sizeof *ctx);
+  if (!ctx) scr_stream_oom();
+  ctx->d = scr_stream_retain(d);
+  ctx->boxed = boxed;
+  scr_promise_release(scr_async_spawn_after(ctx->boxed->v.promise, &scr_dupx_promise_entry, ctx));
+  return d;
+}
+
+/* p and box borrowed (box: the compiler's promise → dyn promise helper). +1. */
+ScrStream *scr_stream_duplex_from_promise(ScrPromise *p, ScrClosure *box) {
+  return scr_dupx_from_boxed(((ScrPipeBoxFn)box->fn)(box, scr_promise_retain(p)));
+}
+
+ScrStream *scr_stream_duplexify_readable(ScrStream *r) { return scr_stream_duplexify(r, NULL); }
+ScrStream *scr_stream_duplexify_writable(ScrStream *w) { return scr_stream_duplexify(NULL, w); }
+ScrStream *scr_stream_duplexify_pair(ScrStream *r, ScrStream *w) { return scr_stream_duplexify(r, w); }
+
 /* ── node:stream/promises ─────────────────────────────────────────────
  * The promise forms ride the callback machinery above with a settling
  * watcher: caps[0] boxes the pending promise; the terminal status

packages/runtime/src/scr_stream.h

@@ -545,6 +545,20 @@ void scr_stream_set_writev_default(ScrStream *s, ScrClosure *cb, ScrStreamWritev
 ScrStream *scr_stream_duplex_from_gen(ScrClosure *stage, ScrClosure *source, ScrClosure *next,
                                       ScrClosure *read, ScrClosure *ret, ScrClosure *thr);
 ScrStream *scr_stream_duplex_from_dyn(struct ScrDyn *body);
+/* Duplex.from over an async function (the compiled stage takes the source
+ * 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);
+/* 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. */
+ScrStream *scr_stream_duplexify(ScrStream *r, ScrStream *w);
+ScrStream *scr_stream_duplexify_readable(ScrStream *r);
+ScrStream *scr_stream_duplexify_writable(ScrStream *w);
+ScrStream *scr_stream_duplexify_pair(ScrStream *r, ScrStream *w);
+/* Duplex.from(promise): a readable Duplexify of the promise's value. p and
+ * box (the compiler's promise → dyn promise helper) borrowed. +1. */
+ScrStream *scr_stream_duplex_from_promise(struct ScrPromise *p, ScrClosure *box);
 /* decodeStrings (false keeps string writes as strings), the writable
  * defaultEncoding option (a known encoding, borrowed), and
  * setDefaultEncoding (validated; answers the stream +1). The chaining

packages/runtime/src/scr_stream_ops.c

@@ -2042,7 +2042,7 @@ ScrStream *scr_stream_from_async_iter(ScrDyn *iter, double flags, double hwm) {
  * (the chunk, or the done marker), and ack() is fromAsyncGen's
  * `process.nextTick(cb)` followed by a fresh deferred. */
 
-enum { CHAN_NONE, CHAN_WRITE, CHAN_FINAL, CHAN_ABORT };
+enum { CHAN_NONE, CHAN_WRITE, CHAN_FINAL, CHAN_ABORT, CHAN_FINAL_ASYNC };
 
 typedef struct OpChan OpChan;
 struct OpChan {
@@ -2052,6 +2052,12 @@ struct OpChan {
   ScrPromise *resolve;  /* +1: the deferred a write or the final settles, NULL once settled */
   int settled;          /* CHAN_*: what settled the deferred, for ack() */
   bool aborted;         /* destroy ran (fromAsyncGen's ac.abort()) */
+  /* An async function body: its promise's then() (duplexify.js) settles
+   * `promise` there; the final waits for it. */
+  bool finalized;       /* final() has started */
+  bool body_settled;    /* the then() callbacks ran */
+  bool final_waiting;   /* the final's callback awaits the settlement */
+  ScrError *body_err;   /* +1: what awaiting the then() result throws, or NULL */
   OpChan *live_prev, *live_next;
 };
 
@@ -2059,6 +2065,7 @@ static SCR_TL OpChan *op_chan_live = NULL;
 
 static void op_chan_drop(OpObj *o) {
   OpChan *c = (OpChan *)o;
+  if (c->body_err) scr_error_release(c->body_err);
   if (c->d) scr_stream_release(c->d);
   if (c->promise) scr_promise_release(c->promise);
   if (c->resolve) scr_promise_release(c->resolve);
@@ -2136,10 +2143,49 @@ static ScrError *op_chan_abort_error(void) {
 
 /* After the source's await: process.nextTick(cb), then (unless done) the
  * aborted check and a fresh deferred. */
+static void op_chan_final_async_wait(OpChan *c);
+
+/* The final's completion after `await promise`: process.nextTick(cb, err) */
+static void op_chan_final_async_done(ScrClosure *job) {
+  OpChan *c = op_chan_of(job);
+  if (c->d) {
+    ScrError *err = c->body_err ? scr_error_retain(c->body_err) : NULL;
+    ScrStream *d = scr_stream_retain(c->d);
+    scr_stream_final_done(d, err);
+    scr_stream_release(d);
+  }
+  op_unref(c);
+}
+
+/* `await promise` resumed: the completion follows on the next tick */
+static void op_chan_final_async_resume(ScrClosure *job) {
+  OpChan *c = op_chan_of(job);
+  scr_next_tick(op_chan_closure((void *)&op_chan_final_async_done, c));
+  op_unref(c);
+}
+
+/* The final's async callback (run by the source's nextTick): it awaits
+ * the then() result, which may still be pending. */
+static void op_chan_final_async_tick(ScrClosure *job) {
+  OpChan *c = op_chan_of(job);
+  op_chan_final_async_wait(c);
+  op_unref(c);
+}
+
+static void op_chan_final_async_wait(OpChan *c) {
+  if (c->body_settled) scr_queue_microtask(op_chan_closure((void *)&op_chan_final_async_resume, c));
+  else c->final_waiting = true;
+}
+
 static void op_chan_ack_inv(ScrClosure *cb) {
   OpChan *c = op_chan_of(cb);
   int settled = c->settled;
   c->settled = CHAN_NONE;
+  if (settled == CHAN_FINAL_ASYNC) {
+    scr_next_tick(op_chan_closure((void *)&op_chan_final_async_tick, c));
+    op_unref(c);
+    return;
+  }
   if (c->d && (settled == CHAN_WRITE || settled == CHAN_FINAL)) {
     ScrClosure *tick = scr_closure_new((void *)&op_chan_done_tick, 2);
     tick->caps[0] = scr_box_new_obj(&scr_stream_retain_v, &scr_stream_release_v, &scr_stream_trace);
@@ -2261,6 +2307,126 @@ ScrStream *scr_stream_duplex_from_gen(ScrClosure *stage, ScrClosure *source, Scr
   return d;
 }
 
+/* ── Duplex.from(asyncFunction) ── duplexify.js: the function's promise
+ * must settle to a nullish value (else ERR_INVALID_RETURN_VALUE when the
+ * final awaits it), a return before the input was consumed destroys the
+ * Duplex, a rejection destroys it with the reason, and the Duplexify is
+ * writable only, its final waiting for the promise. */
+
+typedef ScrPromise *(*OpAsyncStageFn)(ScrClosure *, ScrGen *);
+typedef ScrDyn *(*OpBoxFn)(ScrClosure *, ScrPromise *);
+
+typedef struct {
+  OpChan *c;     /* +1 */
+  ScrDyn *boxed; /* +1, the function's promise boxed */
+} OpAsyncBody;
+
+static void op_async_body_entry(ScrFiber *fiber, void *arg) {
+  (void)fiber;
+  OpAsyncBody b = *(OpAsyncBody *)arg;
+  free(arg);
+  OpChan *c = b.c;
+  ScrDyn *v = scr_await_ref(b.boxed->v.promise); /* NULL: undefined */
+  ScrStream *d = c->d ? scr_stream_retain(c->d) : NULL;
+  if (!scr_exc_pending()) {
+    if (v && v->kind != SCR_DYN_NULL && v->kind != SCR_DYN_UNDEF) {
+      /* the then() callback throws: awaiting its result rejects */
+      char detail[160];
+      const char *got = scr_dyn_specific_type(v, detail, sizeof detail);
+      char msg[320];
+      int n = snprintf(msg, sizeof msg,
+                       "Expected nully to be returned from the \"body\" function but got %s.", got);
+      c->body_err = op_coded_error(SCR_ERR_TYPE, "ERR_INVALID_RETURN_VALUE", msg);
+      (void)n;
+    } else if (!c->finalized && d) {
+      /* returned without consuming the input: destroyer(d) */
+      ScrError *e = op_chan_abort_error();
+      scr_stream_release(scr_stream_destroy(d, e));
+      scr_error_release(e);
+    }
+    if (v) scr_dyn_release(v);
+  } else {
+    ScrCaught *caught = scr_exc_take();
+    ScrError *err = op_caught_error(caught);
+    scr_caught_release(caught);
+    if (d) scr_stream_release(scr_stream_destroy(d, err));
+    scr_error_release(err);
+  }
+  if (scr_exc_pending()) scr_exc_clear();
+  if (d) scr_stream_release(d);
+  c->body_settled = true;
+  if (c->final_waiting) {
+    c->final_waiting = false;
+    op_chan_final_async_wait(c);
+  }
+  op_unref(c);
+  scr_dyn_release(b.boxed);
+}
+
+/* Duplexify's final(cb) for an async function body: finalized, then
+ * fromAsyncGen's final with the awaiting callback */
+static void op_chan_final_async_inv(ScrClosure *cb, ScrStream *s) {
+  (void)s;
+  OpChan *c = op_chan_of(cb);
+  c->finalized = true;
+  op_chan_settle(c, scr_stream_from_done(), CHAN_FINAL_ASYNC);
+  op_unref(c);
+}
+
+/* fromAsyncGen's destroy alone (no from() iterator to close) */
+static void op_chan_destroy_inv(ScrClosure *cb, ScrStream *s, ScrError *err) {
+  OpChan *c = op_chan_of(cb);
+  c->aborted = true;
+  if (c->resolve) op_chan_settle(c, scr_stream_from_done(), CHAN_ABORT);
+  op_chan_retire(c);
+  op_unref(c);
+  scr_stream_destroy_done(s, err ? scr_error_retain(err) : NULL);
+}
+
+/* Duplex.from(fn) for an async function: the compiled stage (borrowed)
+ * takes the source generator and answers its promise; source and box
+ * (promise → dyn promise) borrowed. +1, or NULL with the function's throw
+ * pending. */
+ScrStream *scr_stream_duplex_from_async_fn(ScrClosure *stage, ScrClosure *source, ScrClosure *box) {
+  op_sentinels_init();
+  static SCR_TL bool registered = false;
+  if (!registered) {
+    registered = true;
+    scr_atexit(op_chan_atexit);
+  }
+  OpChan *c = op_alloc(sizeof *c);
+  c->base.rc = 1;
+  c->base.drop = &op_chan_drop;
+  op_chan_defer(c);
+  ScrGen *src = ((OpSourceFn)source->fn)(source, op_chan_closure((void *)&op_chan_next_inv, c),
+                                         op_chan_closure((void *)&op_chan_ack_inv, c));
+  ScrPromise *p = scr_exc_pending() ? NULL : ((OpAsyncStageFn)stage->fn)(stage, src);
+  if (scr_exc_pending()) {
+    if (p) scr_promise_release(p);
+    op_unref(c);
+    return NULL;
+  }
+  /* Duplexify({ objectMode: true, readable: false, write, final, destroy }) */
+  ScrStream *d = scr_stream_new_duplex(16, 16, true, true, true, false, true, NULL, NULL,
+                                       op_chan_closure((void *)&op_chan_write_inv, c), &op_chan_write_inv,
+                                       op_chan_closure((void *)&op_chan_final_async_inv, c),
+                                       &op_chan_final_async_inv,
+                                       op_chan_closure((void *)&op_chan_destroy_inv, c), &op_chan_destroy_inv);
+  scr_stream_set_object_mode(d, true, true);
+  d->cls = "Duplexify";
+  c->d = scr_stream_retain(d);
+  c->live_next = op_chan_live;
+  if (op_chan_live) op_chan_live->live_prev = c;
+  op_chan_live = c; /* the reference c was created with */
+  /* value.then(onFulfilled, onRejected) */
+  OpAsyncBody *b = malloc(sizeof *b);
+  if (!b) op_oom();
+  b->c = op_ref(c);
+  b->boxed = ((OpBoxFn)box->fn)(box, p); /* p moves */
+  scr_promise_release(scr_async_spawn_after(b->boxed->v.promise, &op_async_body_entry, b));
+  return d;
+}
+
 /* ── the promise operators ─────────────────────────────────────────── */
 
 enum { PK_DYN, PK_F64, PK_BOOL, PK_STR, PK_VOID };

tests/corpus/7424-duplex-from-streams.mjs

@@ -0,0 +1,39 @@
+// Duplex.from over a lone Readable, a lone Writable and a { readable,
+// writable } pair, as duplexify.js's _duplexify wraps them: the Duplex
+// reads through the Readable and writes through the Writable, takes each
+// side's object mode from the stream behind it, and closes once both ends
+// finish; destroying it without an error destroys the streams behind it.
+import { Duplex, PassThrough, Readable, Writable } from "node:stream";
+const log = [];
+const r = Readable.from(["a", "b"]);
+const dr = Duplex.from(r);
+log.push(`dr ${dr.readable} ${dr.writable} ${dr.readableObjectMode} ${dr.writableObjectMode}`);
+dr.on("end", () => log.push("dr end"));
+dr.on("close", () => log.push("dr close"));
+for await (const c of dr) log.push(`dr ${c}`);
+const got = [];
+const w = new Writable({ write(c, e, cb) { got.push(String(c)); log.push(`w write ${c}`); cb(); } });
+const dw = Duplex.from(w);
+log.push(`dw ${dw.readable} ${dw.writable} ${dw.writableObjectMode}`);
+dw.on("finish", () => log.push("dw finish"));
+dw.on("close", () => log.push("dw close"));
+dw.write("x", () => log.push("dw cb x"));
+dw.end("y", () => log.push("dw end cb"));
+await new Promise((res) => dw.on("close", res));
+const inner = new PassThrough({ objectMode: true });
+const out = new PassThrough({ objectMode: true });
+const pair = Duplex.from({ readable: out, writable: inner });
+log.push(`pair ${pair.readable} ${pair.writable} ${pair.readableObjectMode} ${pair.writableObjectMode}`);
+inner.on("data", (v) => { log.push(`inner ${JSON.stringify(v)}`); out.write({ echo: v }); });
+inner.on("finish", () => { log.push("inner finish"); out.end(); });
+pair.on("data", (v) => log.push(`pair data ${JSON.stringify(v)}`));
+pair.write({ n: 1 });
+pair.end({ n: 2 });
+await new Promise((res) => pair.on("close", res));
+log.push("pair closed");
+const killed = Duplex.from(new PassThrough());
+killed.on("error", (e) => log.push(`killed error ${e.name}`));
+killed.on("close", () => log.push(`killed close ${killed.destroyed}`));
+killed.destroy();
+await new Promise((res) => setTimeout(res, 5));
+console.log(log.join("\n"));

tests/corpus/7425-duplex-from-streams.ts

@@ -0,0 +1,72 @@
+// Duplex.from over streams in typed code: a Readable that fails destroys
+// the Duplex with its error, a slow Writable's callbacks and the end
+// callback follow its pace, a pair whose writable member is an async
+// generator function's Duplex, compose() with an async function as its
+// last stage, and a Duplex over a promise.
+import { Duplex, PassThrough, Readable, Writable, compose } from "node:stream";
+const log: string[] = [];
+async function main(): Promise<void> {
+  const failing = new Readable({ read(): void { this.destroy(new Error("source failed")); } });
+  const df = Duplex.from(failing);
+  df.on("error", (e: Error) => log.push(`df error ${e.message}`));
+  df.on("close", () => log.push(`df close ${df.destroyed}`));
+  df.resume();
+  await new Promise((res) => df.on("close", res));
+  const slow = new Writable({
+    highWaterMark: 2,
+    write(c: Buffer, _e: string, cb: (err?: Error | null) => void): void {
+      log.push(`slow write ${c.toString()}`);
+      setTimeout(cb, 2);
+    },
+  });
+  const ds = Duplex.from(slow);
+  log.push(`write1 ${ds.write("aa", () => log.push("cb aa"))}`);
+  log.push(`write2 ${ds.write("bb", () => log.push("cb bb"))}`);
+  ds.on("drain", () => log.push("ds drain"));
+  ds.end(() => log.push("ds end cb"));
+  await new Promise((res) => ds.on("finish", res));
+  log.push("ds finished");
+  const upper = new PassThrough();
+  const dp = Duplex.from({
+    readable: Readable.from(["r1", "r2"]),
+    writable: async function* (source: AsyncIterable<unknown>): AsyncGenerator<string> {
+      for await (const c of source) log.push(`gen got ${String(c)}`);
+      yield "unused";
+    },
+  });
+  dp.on("data", (c: unknown) => log.push(`dp data ${String(c)}`));
+  dp.write("w1");
+  dp.end();
+  await new Promise((res) => dp.on("end", res));
+  upper.end();
+  await more();
+  for (const line of log) console.log(line);
+}
+async function more(): Promise<void> {
+  const sink = compose(new PassThrough({ objectMode: true }), async function (source: AsyncIterable<unknown>): Promise<void> {
+    for await (const c of source) log.push(`sink ${String(c)}`);
+  });
+  log.push(`sink r=${sink.readable} w=${sink.writable}`);
+  sink.write("p");
+  sink.end("q");
+  await new Promise((res) => sink.on("finish", res));
+  log.push("sink finished");
+  const fromReadable: Duplex = Duplex.from(Readable.from([1, 2]));
+  const values: unknown[] = [];
+  for await (const v of fromReadable) values.push(v);
+  log.push(`readable ${JSON.stringify(values)}`);
+  const written: string[] = [];
+  const target = new Writable({
+    write(c: Buffer, _e: string, cb: (err?: Error | null) => void): void {
+      written.push(c.toString());
+      cb();
+    },
+  });
+  const fromWritable: Duplex = Duplex.from(target);
+  fromWritable.end("only");
+  await new Promise((res) => fromWritable.on("finish", res));
+  log.push(`writable ${written.join(",")}`);
+  const value: Duplex = Duplex.from(Promise.resolve("resolved"));
+  for await (const v of value) log.push(`promise ${String(v)}`);
+}
+void main();

tests/corpus/7426-duplex-from-promise.mjs

@@ -0,0 +1,27 @@
+// Duplex.from over a promise: a Duplex that is not writable and reads the
+// promise's value once it settles (nothing for null), then ends; a
+// rejection destroys it with the reason. The events interleave with
+// microtasks and nextTick callbacks as in Node.
+import { Duplex } from "node:stream";
+const log = [];
+const tick = (n) => { if (n < 5) { log.push(`tick ${n}`); process.nextTick(() => tick(n + 1)); } };
+const later = (v, ms) => new Promise((res) => setTimeout(() => res(v), ms));
+const a = Duplex.from(Promise.resolve({ answer: 42 }));
+log.push(`a ${a.readable} ${a.writable} ${a.readableObjectMode}`);
+a.on("data", (v) => log.push(`a data ${JSON.stringify(v)}`));
+a.on("end", () => log.push("a end"));
+a.on("close", () => log.push("a close"));
+Promise.resolve().then(() => log.push("microtask 1")).then(() => log.push("microtask 2"));
+tick(0);
+await new Promise((res) => a.on("close", res));
+const b = Duplex.from(later(null, 1));
+b.on("data", () => log.push("b data"));
+b.on("end", () => log.push("b end"));
+await new Promise((res) => b.on("close", res));
+const c = Duplex.from(Promise.reject(new Error("nope")));
+c.on("error", (e) => log.push(`c error ${e.message}`));
+c.on("close", () => log.push(`c close ${c.destroyed}`));
+await new Promise((res) => c.on("close", res));
+const d = Duplex.from(later("text", 1));
+for await (const v of d) log.push(`d ${v}`);
+console.log(log.join("\n"));

tests/corpus/7427-duplex-from-async-function.mjs

@@ -0,0 +1,40 @@
+// Duplex.from over an async function: a Duplex that is only writable, whose
+// writes reach the function's source and whose end waits for the
+// function's promise. A value other than null or undefined is
+// ERR_INVALID_RETURN_VALUE, a return before the input is consumed aborts
+// the Duplex, and a throw destroys it.
+import { Duplex } from "node:stream";
+const log = [];
+const sink = Duplex.from(async function (source) {
+  for await (const c of source) log.push(`sink got ${c}`);
+  log.push("sink done");
+});
+log.push(`sink ${sink.readable} ${sink.writable} ${sink.writableObjectMode}`);
+sink.on("finish", () => log.push("sink finish"));
+sink.on("close", () => log.push("sink close"));
+sink.write("a", () => log.push("cb a"));
+sink.end("b", () => log.push("end cb"));
+await new Promise((res) => sink.on("close", res));
+const bad = Duplex.from(async function (source) {
+  for await (const c of source) log.push(`bad got ${c}`);
+  return 5;
+});
+bad.on("error", (e) => log.push(`bad error ${e.code} ${e.message}`));
+bad.on("close", () => log.push("bad close"));
+bad.end("x");
+await new Promise((res) => bad.on("close", res));
+const early = Duplex.from(async function (source) {
+  for await (const c of source) { log.push(`early got ${c}`); break; }
+});
+early.on("error", (e) => log.push(`early error ${e.name}`));
+early.on("close", () => log.push(`early close ${early.destroyed}`));
+early.write("1");
+early.write("2");
+await new Promise((res) => early.on("close", res));
+const failing = Duplex.from(async function (source) {
+  for await (const c of source) throw new Error("fail " + c);
+});
+failing.on("error", (e) => log.push(`failing error ${e.message}`));
+failing.write("z");
+await new Promise((res) => failing.on("close", res));
+console.log(log.join("\n"));

tests/harness/corpus-costs/7424-duplex-from-streams.mjs.json

@@ -0,0 +1 @@
+{ "plain": 2607, "san": 3440 }

tests/harness/corpus-costs/7425-duplex-from-streams.ts.json

@@ -0,0 +1 @@
+{ "plain": 2009, "san": 2743 }

tests/harness/corpus-costs/7426-duplex-from-promise.mjs.json

@@ -0,0 +1 @@
+{ "plain": 1877, "san": 2357 }

tests/harness/corpus-costs/7427-duplex-from-async-function.mjs.json

@@ -0,0 +1 @@
+{ "plain": 1811, "san": 2417 }

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