Support writev on Transform and PassThrough, and runtime stream option values

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

@@ -32,9 +32,9 @@
   `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.
   Stream constructors accept a `signal` option in an options object literal, and `stream.addAbortSignal` accepts any stream.
-  The `construct` option and `_construct` method, `decodeStrings`, and `defaultEncoding` work on every stream class, and the `writev` option and `_writev` method on Writable and Duplex; `writev` on a Transform or PassThrough, assigning `_construct` after construction, and a non-literal `decodeStrings` or `defaultEncoding` in an options object literal are unsupported.
+  The `construct` option and `_construct` method, `decodeStrings`, and `defaultEncoding` work on every stream class, and the `writev` option and `_writev` method on every class with a writable side.
+  In typed code, a `writev` entry's `encoding` declared as a string receives `""` where Node passes `undefined` or `null` (an object-mode write).
   `setDefaultEncoding()` takes any encoding Node knows.
-  A `transform` or `flush` option callback whose completion callback types `data` as `any`, as @types/node's `TransformCallback` does, may pass a Buffer, a string, or nothing; any other value throws a TypeError where the callback is called, instead of Node's `ERR_INVALID_ARG_TYPE` stream error.
   `node:readline` and `node:readline/promises` (also reached as `readline.promises`) interfaces, from `createInterface(options)` or `createInterface(input, output, completer, terminal)`, read `process.stdin` or a readable stream: `question`, `prompt`, `write` (with a key object literal of `name`, `ctrl`, `meta`, and `shift`), `pause`, `resume`, `'line'`/`'close'`/`'pause'`/`'resume'`/`'error'`/`'history'`/`'SIGINT'`/`'SIGTSTP'`/`'SIGCONT'` listeners, and `for await` compile.
   `on`, `once`, and `addListener` return the interface, so registrations chain.
   The `history`, `historySize`, `removeHistoryDuplicates`, `completer`, `tabSize`, and `escapeCodeTimeout` options are validated as in Node.

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: 33, marker: "scr_runtime_abi_v33" },
+      runtime_abi: { version: 34, marker: "scr_runtime_abi_v34" },
       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_v33");
+    expect(error?.stderr).toContain("scr_runtime_abi_v34");
   });
 });

packages/compiler/src/backend/llvm/expr-stream-callbacks.ts

@@ -727,6 +727,17 @@ export function streamCbThunkFor(
         const ty = host.llType(p);
         host.declare(`declare zeroext i1 @scr_exc_pending()`);
         if (hasThis) host.declare(`declare void @scr_stream_release_v(ptr)`);
+        // An entry `encoding` declared as a string (the declarations'
+        // `encoding: string`) receives "" where Node passes undefined or
+        // null (an object-mode write), as a write's encoding parameter does.
+        const entry =
+          p.kind === "array" && p.elem.kind === "record"
+            ? host.recordsById.get(p.elem.shapeId)
+            : undefined;
+        if (entry?.fields.some((f) => f.name === "encoding" && f.type.kind === "string")) {
+          host.declare(`declare void @scr_stream_writev_fill_encodings(ptr)`);
+          d.push(`  call void @scr_stream_writev_fill_encodings(ptr %chunks)`);
+        }
         d.push(
           `  %vt${i} = call ${ty === "i1" ? "zeroext i1" : ty} @${helper}(ptr %chunks, ptr null)`,
           `  %vtbad${i} = call zeroext i1 @scr_exc_pending()`,

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

@@ -1135,6 +1135,8 @@ export const LIB_FN_SYMS: Record<string, string> = {
   "readable.pushEncoding": "scr_stream_set_push_encoding",
   "stream.objectMode": "scr_stream_object_mode",
   "stream.decodeStrings": "scr_stream_decode_strings",
+  "stream.optionValues": "scr_stream_option_values",
+  "stream.checkEncodingOption": "scr_stream_check_encoding_option",
   "stream.duplexFromGen": "scr_stream_duplex_from_gen",
   "stream.duplexFromDyn": "scr_stream_duplex_from_dyn",
   "stream.duplexifyReadable": "scr_stream_duplexify_readable",

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 = 33 as const;
-export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v33" as const;
+export const RUNTIME_ABI_VERSION = 34 as const;
+export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v34" 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: 33, marker: "scr_runtime_abi_v33" },
+    runtime_abi: { version: 34, marker: "scr_runtime_abi_v34" },
     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: 33, marker: "scr_runtime_abi_v33" },
+    runtime_abi: { version: 34, marker: "scr_runtime_abi_v34" },
     compiler: {
       command: "clang",
       identity: "fixture clang",

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

@@ -2734,8 +2734,8 @@ export function collectClassShapeInner(
       // state, so an override or shadowing field would split-brain.
       // Underscore methods are the SUPPORTED override form — but only
       // the ones the class's own base consumes (a `_read` on a
-      // Transform, or `_writev`/`_construct` anywhere, would be consumed
-      // by Node machinery that has no lowering here).
+      // Transform, or `_writev` on a Readable, would be consumed by Node
+      // machinery that has no lowering here).
       if (memberName && ts.isIdentifier(memberName)) {
         const streamBase = (() => {
           for (let c = base; c; c = c.base) if (c.builtinStream) return c;
@@ -2754,7 +2754,7 @@ export function collectClassShapeInner(
             lowerer.unsupported(
               "SC1090",
               memberName,
-              `declaring '_writev' on a ${streamBase.def.name.slice(1)} subclass (batched writes are supported on Writable and Duplex subclasses)`,
+              `declaring '_writev' on a ${streamBase.def.name.slice(1)} subclass (a Readable has no writable side)`,
             );
           }
           const accepted = streamCtorShape(streamBase.def.name).accepted;

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

@@ -179,15 +179,15 @@ export const UNDERSCORE_METHODS: ReadonlyMap<string, string> = new Map([
 
 /** The option callbacks applied after construction rather than through
  * the constructor's callback slots (option name → method name): construct
- * on every stream class, and writev on the writable classes whose
- * machinery hands it batches (Writable and Duplex). */
+ * on every stream class, and writev on every class with a writable side
+ * (a Transform's batches reach writev, not the transform, as in Node). */
 export const STREAM_EXTRA_CALLBACKS: ReadonlyMap<string, string> = new Map([
   ["construct", "_construct"],
   ["writev", "_writev"],
 ]);
 
 export function streamExtraCallbacks(cls: string): readonly string[] {
-  return cls === "%Writable" || cls === "%Duplex" ? ["construct", "writev"] : ["construct"];
+  return cls === "%Readable" ? ["construct"] : ["construct", "writev"];
 }
 
 /** The completion callback's parameter position in each underscore
@@ -442,6 +442,13 @@ interface StreamOptions {
    * right after construction (Node's constructors call addAbortSignal
    * once the state exists), or null when absent. */
   signal: IrExpr | null;
+  /** Non-literal decodeStrings and defaultEncoding values: valueDecls
+   * evaluates them into locals before construction (the defaultEncoding
+   * validated there, as the state constructors throw before the stream
+   * exists), and stream.optionValues applies them to the fresh stream. */
+  valueDecls: IrStmt[];
+  decodeStringsValue: IrExpr | null;
+  defaultEncodingValue: IrExpr | null;
   /** The objectMode option's sides (a Duplex's readableObjectMode and
    * writableObjectMode set one side): applied with stream.objectMode
    * right after construction. */
@@ -732,6 +739,12 @@ function pinAnyDataCompletion(
   if (!symbol || lowerer.stageSourceTypes.has(symbol)) return null;
   const declared = lowerer.paramShape(param).type;
   if (declared.kind !== "func" || declared.rest || declared.params[1]?.kind !== "dyn") return null;
+  // Only when every use calls the callback with a Buffer, a string or
+  // nothing as `data`: another value (or a callback that escapes) keeps
+  // `data` checked-dynamic, so the runtime pushes it as push(data) does
+  // and reports ERR_INVALID_ARG_TYPE through 'error', as Node does.
+  if (fnNode.body === undefined || !completionDataIsChunk(lowerer, fnNode.body, symbol, param.name))
+    return null;
   const data = unionOf(lowerer, [BYTES, STRING, { kind: "undefinedT" }]);
   lowerer.stageSourceTypes.set(
     symbol,
@@ -740,6 +753,59 @@ function pinAnyDataCompletion(
   return symbol;
 }
 
+/** Whether every reference to the completion callback `symbol` in `body`
+ * calls it with a `data` argument that is a Buffer, a string, null or
+ * undefined (or with no data). */
+function completionDataIsChunk(
+  lowerer: Lowerer,
+  body: ts.Node,
+  symbol: ts.Symbol,
+  declaration: ts.Node,
+): boolean {
+  let ok = true;
+  const visit = (node: ts.Node): void => {
+    if (!ok) return;
+    if (
+      ts.isIdentifier(node) &&
+      node !== declaration &&
+      lowerer.checker.getSymbolAtLocation(node) === symbol
+    ) {
+      const call = node.parent;
+      if (!call || !ts.isCallExpression(call) || call.expression !== node) {
+        ok = false;
+        return;
+      }
+      const data = call.arguments.nodes[1];
+      if (data !== undefined) {
+        let t: IrType | null;
+        try {
+          t = lowerer.mapTypeOf(lowerer.typeOf(data));
+        } catch {
+          t = null;
+        }
+        const arms =
+          t?.kind === "union" ? (lowerer.unions.get(t.unionId)?.arms ?? null) : t ? [t] : null;
+        if (
+          arms === null ||
+          !arms.every(
+            (a) =>
+              (a.kind === "bytes" && a.elem === "u8") ||
+              a.kind === "string" ||
+              a.kind === "undefinedT" ||
+              a.kind === "nullT",
+          )
+        ) {
+          ok = false;
+          return;
+        }
+      }
+    }
+    ts.forEachChild(node, visit);
+  };
+  visit(body);
+  return ok;
+}
+
 /** Parses the options object literal of a stream constructor. `which`
  * names the callbacks this class accepts, in canonical flag order. */
 function parseStreamOptions(
@@ -767,6 +833,9 @@ function parseStreamOptions(
     badEncoding: null,
     pushEncoding: null,
     signal: null,
+    valueDecls: [],
+    decodeStringsValue: null,
+    defaultEncodingValue: null,
     readableObject: false,
     writableObject: false,
     decodeStringsOff: false,
@@ -802,6 +871,21 @@ function parseStreamOptions(
     optsLit !== undefined &&
     ts.isObjectLiteralExpression(optsLit) &&
     literalWritesValues(optsLit, ctorName !== "Readable" && ctorName !== "Writable");
+  // In JavaScript a decodeStrings value known only at run time may let
+  // string writes through, so the write and transform callbacks' chunk
+  // parameters read as any (the declarations type it Buffer).
+  const runtimeDecodeStrings =
+    optsLit !== undefined &&
+    isJsSourceFile(ts.getSourceFileOfNode(optsLit)) &&
+    ts.isObjectLiteralExpression(optsLit) &&
+    (optsLit.properties?.nodes ?? []).some(
+      (p) =>
+        ts.isPropertyAssignment(p) &&
+        (ts.isIdentifier(p.name) || ts.isStringLiteral(p.name)) &&
+        p.name.text === "decodeStrings" &&
+        p.initializer.kind !== ts.SyntaxKind.TrueKeyword &&
+        p.initializer.kind !== ts.SyntaxKind.FalseKeyword,
+    );
   if (out.readableObject) out.hwmR = numLit(16, loc);
   if (out.writableObject) out.hwmW = numLit(16, loc);
   const tuples = cbTuples(lowerer);
@@ -851,6 +935,7 @@ function parseStreamOptions(
       }
       const full = tuples[name]!.map((t) => (t === "fn" ? null : t));
       const pinned = out.readableObject ? null : pinAnyDataCompletion(lowerer, name, fnNode);
+      let anyChunk: ts.Symbol | undefined;
       try {
         // The completion-callback tail: typed by the checker contextually;
         // accept the user's declared func type as the tuple's entry so the
@@ -867,12 +952,22 @@ function parseStreamOptions(
           }
           return declared ?? funcOf([], VOID);
         });
+        const chunkParam =
+          runtimeDecodeStrings && (name === "write" || name === "transform")
+            ? fnNode.parameters.nodes[0]
+            : undefined;
+        anyChunk =
+          chunkParam !== undefined && ts.isIdentifier(chunkParam.name) && !chunkParam.type
+            ? lowerer.checker.getSymbolAtLocation(chunkParam.name)
+            : undefined;
+        if (anyChunk) lowerer.anySymbols.add(anyChunk);
         present.set(
           name,
           lowerStreamCallback(lowerer, name, fnNode, thisType, tuple, writesValues),
         );
       } finally {
         if (pinned) lowerer.stageSourceTypes.delete(pinned);
+        if (anyChunk) lowerer.anySymbols.delete(anyChunk);
       }
       continue;
     }
@@ -925,11 +1020,15 @@ function parseStreamOptions(
           if (writableClass) out.decodeStringsOff = true;
           break;
         }
-        lowerer.noLowering(
-          `${ctorName} with a runtime 'decodeStrings' value`,
-          prop,
-          "write 'decodeStrings: true' or 'decodeStrings: false'",
-        );
+        // a runtime value: WritableState's `options.decodeStrings !== false`
+        const local = lowerer.declareHiddenLocal("%decodeStrings", DYN);
+        out.valueDecls.push({
+          kind: "varDecl",
+          localId: local.id,
+          init: lowerer.lowerExprExpecting(value, DYN),
+          loc,
+        });
+        if (writableClass) out.decodeStringsValue = varRef(local.id, DYN, loc);
         break;
       }
       case "encoding": {
@@ -973,11 +1072,33 @@ function parseStreamOptions(
             break;
           }
         }
-        lowerer.noLowering(
-          `the ${ctorName} option 'defaultEncoding' with a runtime value`,
-          prop,
-          "write the encoding as a string literal",
-        );
+        {
+          // a runtime value: null, undefined and utf8 are the default,
+          // another name Buffer.isEncoding knows is kept as given, and any
+          // other value throws ERR_UNKNOWN_ENCODING before construction
+          const local = lowerer.declareHiddenLocal("%defaultEncoding", DYN);
+          const ref = varRef(local.id, DYN, loc);
+          out.valueDecls.push(
+            {
+              kind: "varDecl",
+              localId: local.id,
+              init: lowerer.lowerExprExpecting(value, DYN),
+              loc,
+            },
+            {
+              kind: "exprStmt",
+              expr: {
+                kind: "libCall",
+                fn: "stream.checkEncodingOption",
+                args: [ref],
+                type: VOID,
+                loc,
+              },
+              loc,
+            },
+          );
+          out.defaultEncodingValue = ref;
+        }
         break;
       case "signal": {
         // `if (options.signal) addAbortSignal(options.signal, this)`:
@@ -1041,9 +1162,7 @@ function parseStreamOptions(
         lowerer.noLowering(
           `the ${ctorName} option 'writev'`,
           prop,
-          ctorName === "Readable"
-            ? "a Readable has no writable side"
-            : "batched writes on a Transform would bypass its transform — writev is supported on Writable and Duplex",
+          "a Readable has no writable side",
         );
         break;
       default:
@@ -1083,6 +1202,9 @@ function streamCtorArgs(
   badEncoding: string | null;
   pushEncoding: string | null;
   signal: IrExpr | null;
+  valueDecls: IrStmt[];
+  decodeStringsValue: IrExpr | null;
+  defaultEncodingValue: IrExpr | null;
   readableObject: boolean;
   writableObject: boolean;
   decodeStringsOff: boolean;
@@ -1117,6 +1239,9 @@ function streamCtorArgs(
     badEncoding: o.badEncoding,
     pushEncoding: o.pushEncoding,
     signal: o.signal,
+    valueDecls: o.valueDecls,
+    decodeStringsValue: o.decodeStringsValue,
+    defaultEncodingValue: o.defaultEncodingValue,
     readableObject: o.readableObject,
     writableObject: o.writableObject,
     decodeStringsOff: o.decodeStringsOff,
@@ -1237,6 +1362,18 @@ export function lowerStreamNew(lowerer: Lowerer, expr: ts.NewExpression, info: C
       loc,
     };
   }
+  const values = optionValuesArgs(parsed, loc);
+  if (values) {
+    // runtime decodeStrings and defaultEncoding values (answers the
+    // receiver +1)
+    built = {
+      kind: "libCall",
+      fn: "stream.optionValues",
+      args: [built, ...values],
+      type: thisType,
+      loc,
+    };
+  }
   if (parsed.pushEncoding !== null) {
     // { defaultEncoding }: chain the push-side decode default (the
     // setEncoding chaining shape — answers the receiver +1).
@@ -1269,8 +1406,11 @@ export function lowerStreamNew(lowerer: Lowerer, expr: ts.NewExpression, info: C
       loc,
     };
   }
-  if (!parsed.decodeStringsOff && parsed.writeEncoding === null && parsed.extras.length === 0)
-    return built;
+  if (!parsed.decodeStringsOff && parsed.writeEncoding === null && parsed.extras.length === 0) {
+    return parsed.valueDecls.length > 0
+      ? { kind: "seqExpr", stmts: parsed.valueDecls, result: built, type: thisType, loc }
+      : built;
+  }
   const fresh = lowerer.declareHiddenLocal("%stream", thisType);
   const followUps = streamOptionFollowUps(
     parsed,
@@ -1280,13 +1420,30 @@ export function lowerStreamNew(lowerer: Lowerer, expr: ts.NewExpression, info: C
   );
   return {
     kind: "seqExpr",
-    stmts: [{ kind: "varDecl", localId: fresh.id, init: built, loc }, ...followUps],
+    stmts: [
+      ...parsed.valueDecls,
+      { kind: "varDecl", localId: fresh.id, init: built, loc },
+      ...followUps,
+    ],
     result: varRef(fresh.id, thisType, loc),
     type: thisType,
     loc,
   };
 }
 
+/** stream.optionValues' (decodeStrings, defaultEncoding) arguments, or
+ * null when both options are literals or absent. */
+function optionValuesArgs(
+  parsed: { decodeStringsValue: IrExpr | null; defaultEncodingValue: IrExpr | null },
+  loc: SrcLoc,
+): IrExpr[] | null {
+  if (parsed.decodeStringsValue === null && parsed.defaultEncodingValue === null) return null;
+  return [
+    parsed.decodeStringsValue ?? dynUndefinedExpr(loc),
+    parsed.defaultEncodingValue ?? dynUndefinedExpr(loc),
+  ];
+}
+
 /** True when a value of Node's argument type `passed` widens implicitly
  * into the declared parameter type `declared` (one of its union arms, or a
  * sub-union of it). */
@@ -1589,6 +1746,7 @@ export function lowerStreamSuperCall(
     ];
   }
   const out: IrStmt[] = [
+    ...parsed.valueDecls,
     {
       kind: "exprStmt",
       expr: {
@@ -1614,6 +1772,20 @@ export function lowerStreamSuperCall(
       loc,
     });
   }
+  const values = optionValuesArgs(parsed, loc);
+  if (values) {
+    out.push({
+      kind: "exprStmt",
+      expr: {
+        kind: "libCall",
+        fn: "stream.optionValues",
+        args: [thisRef(), ...values],
+        type: thisType,
+        loc,
+      },
+      loc,
+    });
+  }
   if (parsed.pushEncoding !== null) {
     out.push({
       kind: "exprStmt",
@@ -1699,12 +1871,20 @@ export function lowerStreamUnderscoreAssign(
   const clsName = cls.slice(1);
   if (methodName === "_construct") {
     // Node's constructor reads _construct once; a later assignment has no
-    // effect there
-    lowerer.noLowering(
-      `assigning '${methodName}' on a ${clsName}`,
-      expr.left,
-      "declare _construct on the class or pass the construct option",
-    );
+    // effect on the stream, so only the receiver and the value are
+    // evaluated
+    const value = lowerer.lowerExpr(expr.right);
+    return {
+      kind: "exprStmt",
+      expr: {
+        kind: "seqExpr",
+        stmts: [{ kind: "exprStmt", expr: recv, loc }],
+        result: value,
+        type: value.type,
+        loc,
+      },
+      loc,
+    };
   }
   const which = methodName === "_writev" ? "writev" : optionByMethod.get(methodName)!;
   const { accepted } = streamCtorShape(cls);

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

@@ -362,6 +362,7 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "transform.newDyn",
   "passthrough.newDyn",
   "readable.initDyn",
+  "stream.checkEncodingOption",
   "writable.initDyn",
   "duplex.initDyn",
   "transform.initDyn",

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

@@ -4103,6 +4103,11 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
    * after construction; answers the receiver (+1, the chaining shape). */
   "stream.objectMode": { argTypes: [null, BOOL, BOOL], result: VOID },
   "stream.decodeStrings": { argTypes: [null, BOOL], result: VOID },
+  /** Runtime decodeStrings and defaultEncoding option values, applied
+   * after construction (answers the receiver +1); the encoding was
+   * validated by stream.checkEncodingOption before construction. */
+  "stream.optionValues": { argTypes: [null, DYN, DYN], result: VOID },
+  "stream.checkEncodingOption": { argTypes: [DYN], 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,

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

@@ -219,6 +219,8 @@ const STREAM_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "stream.chunkDone",
   "stream.objectMode",
   "stream.decodeStrings",
+  "stream.optionValues",
+  "stream.checkEncodingOption",
   "stream.duplexFromGen",
   "stream.duplexFromDyn",
   "stream.duplexFromPromise",

packages/compiler/src/ir/validate.ts

@@ -5495,6 +5495,7 @@ function validateFunction(
       e.fn === "stream.destroyErr" ||
       e.fn === "stream.objectMode" ||
       e.fn === "stream.decodeStrings" ||
+      e.fn === "stream.optionValues" ||
       e.fn === "stream.duplexFromGen" ||
       e.fn === "stream.duplexFromDyn" ||
       e.fn === "stream.duplexifyReadable" ||
@@ -5867,6 +5868,7 @@ function validateFunction(
         e.fn === "readable.pushEncoding" ||
         e.fn === "stream.objectMode" ||
         e.fn === "stream.decodeStrings" ||
+        e.fn === "stream.optionValues" ||
         e.fn === "writable.writeEncoding" ||
         e.fn === "writable.setDefaultEncoding" ||
         e.fn === "stream.destroy" ||

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

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7460-transform-writev.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7460-transform-writev.mjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7461-transform-writev.cjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7461-transform-writev.cjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7462-transform-writev.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7462-transform-writev.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7463-stream-option-values.ts.json

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

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7464-stream-option-values.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7464-stream-option-values.mjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7465-transform-callback-values.ts.json

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

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

@@ -340,7 +340,7 @@ async function build() {
       package: packageManifest.name,
       version: packageManifest.version,
       target: RUNTIME_PACK_MATRIX.target,
-      runtime_abi: { version: 33, marker: "scr_runtime_abi_v33" },
+      runtime_abi: { version: 34, marker: "scr_runtime_abi_v34" },
       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: 33, marker: "scr_runtime_abi_v33" },
+      runtime_abi: { version: 34, marker: "scr_runtime_abi_v34" },
       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_v33(void) {}
+void scr_runtime_abi_v34(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_v33(void);
+void scr_runtime_abi_v34(void);
 void scr_runtime_workers_v8(void);
 
 /* ── the trap funnel (scr_console.c; scr_library.c under -DSCR_LIB) ──────

packages/runtime/src/scr_dyn_builtin_stream.c

@@ -34,7 +34,7 @@
  * receiver without a reference; the hook closures outlive the record.
  *
  * Not supported here (each throws the "not supported yet" Error where it
- * runs): the options scr_stream.c's dyn option walk refuses, `_writev` and `_construct`, symbol event names, listeners() /
+ * runs): the options scr_stream.c's dyn option walk refuses, symbol event names, listeners() /
  * rawListeners(), `wrap()`, and pipes into values that are neither runtime
  * streams nor constructor-function streams. */
 #include "scr_runtime.h"
@@ -898,20 +898,15 @@ static ScrDyn *bcs_construct(int kind, ScrDyn *self, ScrDyn *const *args, size_t
   }
   bool failed = false;
   bool writes = kind != BCS_READABLE;
-  bool transformish = kind == BCS_TRANSFORM || kind == BCS_PASSTHROUGH;
   ScrDyn *opts = bcs_arg(args, argc, 0);
   /* Node's constructors assign options.writev and options.construct to
    * the instance (this._writev = options.writev), then read _writev and
    * _construct, so an option shadows the prototype's method. */
   if (opts->kind == SCR_DYN_OBJ) {
-    if (writes && !transformish && !bcs_copy_option(self, opts, "writev", "_writev")) return NULL;
+    if (writes && !bcs_copy_option(self, opts, "writev", "_writev")) return NULL;
     if (!bcs_copy_option(self, opts, "construct", "_construct")) return NULL;
   }
   bool has_writev = writes && bcs_is_function(self, "_writev", &failed);
-  if (has_writev && transformish) {
-    bcs_unsupported("_writev on a constructor-function Transform is not supported yet");
-    return NULL;
-  }
   bool has_construct = !failed && bcs_is_function(self, "_construct", &failed);
   if (failed) return NULL;
   ScrStream *s = kind == BCS_READABLE   ? scr_stream_new_readable_dyn(opts)

packages/runtime/src/scr_stream.c

@@ -2157,6 +2157,24 @@ static ScrDyn *scr_stream_wentry_encoding(const ScrStreamState *st, ScrDyn *enc)
   return scr_dyn_retain(enc ? enc : scr_dyn_undefined());
 }
 
+/* A compiled writev whose entries declare `encoding` a string: an entry
+ * without one (an object-mode write passes undefined or null) gets "".
+ * The batch array is the one this call receives. */
+void scr_stream_writev_fill_encodings(ScrDyn *chunks) {
+  if (!chunks || chunks->kind != SCR_DYN_ARR) return;
+  double n = scr_dyn_arr_len(chunks);
+  for (double i = 0; i < n; i++) {
+    ScrDyn *entry = scr_dyn_arr_at(chunks, i); /* +1 */
+    ScrDyn *enc = entry->kind == SCR_DYN_OBJ ? scr_dyn_obj_get(entry, "encoding", 8) : NULL;
+    if (entry->kind == SCR_DYN_OBJ && (!enc || enc->kind == SCR_DYN_UNDEF || enc->kind == SCR_DYN_NULL)) {
+      ScrStr *empty = scr_str_new("", 0);
+      scr_dyn_obj_set(entry, "encoding", 8, scr_dyn_new_str(empty));
+      scr_str_release(empty);
+    }
+    scr_dyn_release(entry);
+  }
+}
+
 /* A batch entry's callback: the write's callback, or Node's nop. */
 static ScrDyn *scr_stream_writev_entry_cb_call(ScrClosure *clo, ScrDyn *const *args, size_t argc) {
   (void)args;
@@ -2675,9 +2693,8 @@ void *scr_stream_native(ScrStream *s, void (*release)(void *)) {
  * way the compiler-emitted dyn thunks do (chunk as Buffer-flavored bytes,
  * encoding 'buffer', the error via the boundary encoding, the completion
  * callback as a callable dyn over the *_done glue). Options that Node's
- * base WOULD consume but have no lowering here (objectMode truthy,
- * decodeStrings:false, non-utf8 defaultEncoding, signal, construct,
- * writev, and read/write/final on the transformish bases) throw the
+ * base WOULD consume but have no lowering here (signal, and
+ * read/write/final on the transformish bases) throw the
  * "not supported yet" Error — the compile-time fence's runtime twin. */
 
 /* The boxed dyn callable (+1 — scr_box_get_ref retains; callers release
@@ -2903,11 +2920,8 @@ static bool scr_stream_parse_dyn_opts(const ScrDyn *opts, char base, ScrStreamDy
     }
   }
   out->construct = scr_stream_dynopt_fn(opts, "construct", 9);
+  /* a Transform's batches reach writev, not the transform, as in Node */
   if (base != 'r') out->writev = scr_stream_dynopt_fn(opts, "writev", 6);
-  if (transformish && out->writev != NULL) {
-    static const char m[] = "the stream option 'writev' on a Transform is not supported yet (batches would bypass the transform)";
-    return scr_stream_dynopt_unsupported(m, sizeof m - 1);
-  }
   {
     /* a known encoding or ERR_UNKNOWN_ENCODING (the states' constructors) */
     ScrDyn *v = scr_dyn_obj_get((ScrDyn *)opts, "defaultEncoding", 15);
@@ -6296,6 +6310,55 @@ ScrStream *scr_stream_set_default_encoding(ScrStream *s, ScrDyn *enc) {
   return scr_stream_retain(s);
 }
 
+/* Whether a defaultEncoding option value is the default: null, undefined,
+ * 'utf8' or 'utf-8' (the state constructors' kDefaultUTF8Encoding). */
+static bool scr_stream_default_encoding_value(const ScrDyn *v) {
+  if (!v || v->kind == SCR_DYN_UNDEF || v->kind == SCR_DYN_NULL) return true;
+  return v->kind == SCR_DYN_STR && ((v->v.str->len == 4 && memcmp(v->v.str->data, "utf8", 4) == 0) ||
+                                    (v->v.str->len == 5 && memcmp(v->v.str->data, "utf-8", 5) == 0));
+}
+
+/* A runtime defaultEncoding option value, checked before construction as
+ * the state constructors check it: the default, a name Buffer.isEncoding
+ * knows, or ERR_UNKNOWN_ENCODING (thrown). Borrows v. */
+void scr_stream_check_encoding_option(ScrDyn *v) {
+  if (scr_stream_default_encoding_value(v)) return;
+  if (v->kind == SCR_DYN_STR) {
+    ScrStr *canonical = scr_bytes_normalize_encoding(v->v.str);
+    if (canonical) scr_str_release(canonical);
+    return;
+  }
+  ScrStr *shown = scr_dyn_string_coerce_js(v);
+  if (!shown) return;
+  char msg[128];
+  int len = snprintf(msg, sizeof msg, "Unknown encoding: %.*s", (int)(shown->len < 64 ? shown->len : 64),
+                     shown->data);
+  scr_str_release(shown);
+  scr_throw_error_msg_code(SCR_ERR_TYPE, msg, (size_t)(len < 0 ? 0 : len), "ERR_UNKNOWN_ENCODING");
+}
+
+/* Runtime decodeStrings and defaultEncoding option values over a fresh
+ * stream (after its objectMode): decodeStrings === false keeps string
+ * writes, and a known defaultEncoding is the write side's default as
+ * given and how a byte-mode readable side decodes push(string). Borrows
+ * both; answers the receiver (+1). */
+ScrStream *scr_stream_option_values(ScrStream *s, ScrDyn *decode_strings, ScrDyn *default_encoding) {
+  ScrStreamState *st = s->st;
+  if (st->has_w && decode_strings && decode_strings->kind == SCR_DYN_BOOL && !decode_strings->v.b)
+    scr_stream_set_decode_strings(s, false);
+  if (!scr_stream_default_encoding_value(default_encoding) && default_encoding->kind == SCR_DYN_STR) {
+    ScrStr *canonical = scr_bytes_normalize_encoding(default_encoding->v.str);
+    if (canonical) {
+      scr_stream_set_write_encoding(s, default_encoding->v.str);
+      if (st->has_r && !st->r.object_entries) scr_stream_release(scr_stream_set_push_encoding(s, canonical));
+      scr_str_release(canonical);
+    } else if (scr_exc_pending()) {
+      scr_exc_clear(); /* checked before construction */
+    }
+  }
+  return scr_stream_retain(s);
+}
+
 /* stream.objectMode: scr_stream_set_object_mode, answering the receiver
  * (+1). */
 ScrStream *scr_stream_object_mode(ScrStream *s, bool readable, bool writable) {

packages/runtime/src/scr_stream.h

@@ -563,6 +563,9 @@ void scr_stream_set_writev_concat(ScrStream *s);
 void scr_stream_set_construct(ScrStream *s, ScrClosure *cb, ScrStreamPlainInv inv);
 void scr_stream_set_writev(ScrStream *s, ScrClosure *cb, ScrStreamWritevInv inv);
 void scr_stream_set_writev_default(ScrStream *s, ScrClosure *cb, ScrStreamWritevInv inv);
+/* A compiled writev whose entries declare `encoding` a string: entries
+ * without an encoding (object mode) get "" before the conversion. */
+void scr_stream_writev_fill_encodings(struct ScrDyn *chunks);
 /* Duplex.from (duplexify.js): over an async generator function (the
  * compiled stage takes the source generator that the compiled source
  * makes from next() and ack(); all borrowed, helpers too), and over a
@@ -605,6 +608,11 @@ ScrStream *scr_stream_duplex_from_promise(struct ScrPromise *p, ScrClosure *box)
  * forms answer the stream (+1). */
 void scr_stream_set_decode_strings(ScrStream *s, bool decode);
 ScrStream *scr_stream_decode_strings(ScrStream *s, bool decode);
+/* Runtime decodeStrings/defaultEncoding option values: checked before
+ * construction (ERR_UNKNOWN_ENCODING thrown), then applied to the fresh
+ * stream, answering it (+1). Values borrowed. */
+void scr_stream_check_encoding_option(struct ScrDyn *v);
+ScrStream *scr_stream_option_values(ScrStream *s, struct ScrDyn *decode_strings, struct ScrDyn *default_encoding);
 void scr_stream_set_write_encoding(ScrStream *s, ScrStr *enc);
 ScrStream *scr_stream_write_encoding(ScrStream *s, ScrStr *enc);
 ScrStream *scr_stream_set_default_encoding(ScrStream *s, struct ScrDyn *enc);

tests/corpus/7460-transform-writev.mjs

@@ -0,0 +1,33 @@
+// writev on a Transform and a PassThrough (the writev option): a batch of
+// buffered writes reaches writev instead of the transform, a lone write
+// still reaches the transform, and the readable side gets what writev
+// pushes.
+import { Transform, PassThrough } from "node:stream";
+const t = new Transform({
+  transform(chunk, enc, cb) { console.log("transform", String(chunk), enc); cb(null, chunk); },
+  writev(chunks, cb) {
+    console.log("writev", chunks.map((c) => `${c.chunk}/${c.encoding}`).join(","));
+    for (const { chunk } of chunks) this.push(String(chunk).toUpperCase());
+    cb();
+  },
+});
+t.on("data", (d) => console.log("data", String(d)));
+t.on("finish", () => console.log("finish"));
+t.on("end", () => console.log("end"));
+t.write("a");
+t.cork();
+t.write("b");
+t.write("c", "latin1");
+process.nextTick(() => { t.uncork(); t.end("d"); });
+
+const p = new PassThrough({
+  objectMode: true,
+  writev(chunks, cb) { console.log("pt writev", JSON.stringify(chunks.map((c) => c.chunk))); cb(); },
+});
+p.on("data", (d) => console.log("pt data", JSON.stringify(d)));
+p.cork();
+p.write({ x: 1 });
+p.write([2]);
+p.uncork();
+p.write("solo");
+p.end();

tests/corpus/7461-transform-writev.cjs

@@ -0,0 +1,33 @@
+// Constructor-function Transform and PassThrough streams with _writev:
+// the prototype's _writev takes the batches, and a writev option shadows
+// it with the stream as this.
+'use strict';
+const { Transform, PassThrough } = require('node:stream');
+function T(opts) { Transform.call(this, opts); }
+Object.setPrototypeOf(T.prototype, Transform.prototype);
+Object.setPrototypeOf(T, Transform);
+T.prototype._transform = function (chunk, enc, cb) { cb(null, 'T(' + chunk + ')'); };
+T.prototype._writev = function (chunks, cb) {
+  this.push('V(' + chunks.map((c) => String(c.chunk)).join('+') + ')');
+  cb();
+};
+const t = Object.create(T.prototype);
+T.call(t, {});
+t.on('data', (d) => console.log('t data', String(d)));
+t.on('end', () => console.log('t end'));
+t.write('a');
+t.cork();
+t.write('b');
+t.write('c');
+process.nextTick(() => { t.uncork(); t.end(); });
+function P(opts) { PassThrough.call(this, opts); }
+Object.setPrototypeOf(P.prototype, PassThrough.prototype);
+Object.setPrototypeOf(P, PassThrough);
+const p = Object.create(P.prototype);
+P.call(p, { writev(chunks, cb) { console.log('p writev', chunks.length, this === p); cb(); } });
+p.on('data', (d) => console.log('p data', String(d)));
+p.cork();
+p.write('x');
+p.write('y');
+p.uncork();
+p.write('z');

tests/corpus/7462-transform-writev.ts

@@ -0,0 +1,109 @@
+// Typed writev on Transform and PassThrough: an options writev with
+// decodeStrings: false and flush, a Transform subclass's _writev, a
+// batch completing with an error, and object-mode entries whose
+// encoding is declared as a string or as Node's string | null |
+// undefined.
+import { Transform, PassThrough, Writable } from "node:stream";
+type WritevChunks = Array<{ chunk: any; encoding: BufferEncoding }>;
+type Cb = (error?: Error | null) => void;
+type TransformCallback = (error?: Error | null, data?: any) => void;
+const log: string[] = [];
+const t = new Transform({
+  decodeStrings: false,
+  transform(chunk: string, enc: BufferEncoding, cb: TransformCallback): void {
+    log.push(`transform ${chunk} ${enc}`);
+    cb(null, chunk);
+  },
+  flush(cb: TransformCallback): void {
+    log.push("flush");
+    cb(null, "!");
+  },
+  writev(chunks: WritevChunks, cb: Cb): void {
+    log.push(`writev ${chunks.map((c) => `${c.chunk}/${c.encoding}`).join(",")}`);
+    this.push(chunks.map((c) => String(c.chunk)).join(""));
+    cb();
+  },
+});
+t.on("data", (d: Buffer) => log.push(`data ${d.toString()}`));
+t.on("end", () => log.push("end"));
+t.on("finish", () => log.push("finish"));
+t.on("close", () => {
+  console.log(log.join("\n"));
+  next();
+});
+t.write("one");
+t.cork();
+t.write("two", "latin1");
+t.write("three");
+t.uncork();
+t.end("four");
+
+class Batcher extends Transform {
+  _transform(chunk: Buffer, _enc: BufferEncoding, cb: TransformCallback): void {
+    console.log("batcher transform", chunk.toString());
+    cb(null, chunk);
+  }
+  _writev(chunks: Array<{ chunk: any; encoding: BufferEncoding }>, cb: Cb): void {
+    console.log("batcher writev", chunks.length);
+    this.push(Buffer.concat(chunks.map((c) => c.chunk as Buffer)));
+    cb();
+  }
+}
+
+function next(): void {
+  const b = new Batcher();
+  b.on("data", (d: Buffer) => console.log("batcher data", d.toString()));
+  b.write("x");
+  b.cork();
+  b.write("y");
+  b.write("z");
+  process.nextTick(() => {
+    b.uncork();
+    b.end();
+  });
+  b.on("finish", () => {
+    const p = new PassThrough({
+      writev(chunks: WritevChunks, cb: Cb): void {
+        console.log("pt writev", chunks.length, String(chunks[0]!.chunk));
+        cb(new Error("batch failed"));
+      },
+    });
+    p.on("error", (e: Error) => {
+      console.log("pt error", e.message);
+      objectEntries();
+    });
+    p.cork();
+    p.write("1");
+    p.write("2");
+    p.uncork();
+  });
+}
+
+function objectEntries(): void {
+  const declared = new Writable({
+    objectMode: true,
+    write(_chunk: any, _enc: BufferEncoding, cb: Cb): void { cb(); },
+    writev(chunks: WritevChunks, cb: Cb): void {
+      console.log("string encodings", JSON.stringify(chunks.map((c) => c.chunk)));
+      cb();
+    },
+  });
+  declared.write({ n: 0 });
+  declared.cork();
+  declared.write({ n: 1 });
+  declared.write({ n: 2 }, () => {});
+  declared.uncork();
+  const nodeTyped = new Writable({
+    objectMode: true,
+    write(_chunk: any, _enc: BufferEncoding, cb: Cb): void { cb(); },
+    writev(chunks: Array<{ chunk: any; encoding: string | null | undefined }>, cb: Cb): void {
+      console.log("node encodings", JSON.stringify(chunks.map((c) => [c.chunk, c.encoding === undefined ? "undefined" : c.encoding])));
+      cb();
+    },
+  });
+  nodeTyped.write({ n: 0 });
+  nodeTyped.cork();
+  nodeTyped.write({ n: 1 });
+  nodeTyped.write({ n: 2 }, () => {});
+  nodeTyped.uncork();
+}

tests/corpus/7463-stream-option-values.ts

@@ -0,0 +1,58 @@
+// Typed decodeStrings and defaultEncoding options whose values are known
+// only at run time: decodeStrings === false keeps string writes, a known
+// defaultEncoding is the write default as given and how push(string)
+// decodes, null and utf8 are the default, an unknown name throws
+// ERR_UNKNOWN_ENCODING before construction, and a subclass's super()
+// options take them the same way.
+import { Writable, Readable, Duplex } from "node:stream";
+type Cb = (error?: Error | null) => void;
+const flags: boolean[] = JSON.parse("[false, true]");
+const encodings: Array<string | null> = JSON.parse('["hex", "latin1", null, "UTF8"]');
+for (const decode of flags) {
+  const w = new Writable({
+    decodeStrings: decode,
+    defaultEncoding: encodings[0] as BufferEncoding,
+    write(chunk: any, enc: string, cb: Cb): void {
+      console.log("write", decode, typeof chunk, enc, String(chunk));
+      cb();
+    },
+  });
+  w.write("6869");
+}
+const r = new Readable({ defaultEncoding: encodings[1] as BufferEncoding, read(): void {} });
+r.push("é");
+r.push(null);
+r.on("data", (d: Buffer) => console.log("readable", d.toString("hex")));
+const dflt = new Writable({
+  defaultEncoding: encodings[2] as any,
+  write(chunk: Buffer, enc: string, cb: Cb): void {
+    console.log("null default", enc, chunk.toString());
+    cb();
+  },
+});
+dflt.write("ok");
+const upper = new Duplex({
+  defaultEncoding: encodings[3] as BufferEncoding,
+  decodeStrings: flags[0],
+  read(): void {},
+  write(chunk: any, enc: string, cb: Cb): void {
+    console.log("duplex", typeof chunk, enc);
+    cb();
+  },
+});
+upper.write("x");
+try {
+  new Writable({ defaultEncoding: JSON.parse('"nope"') });
+} catch (e) {
+  console.log("bad", (e as Error & { code?: string }).code, (e as Error).message);
+}
+class Sink extends Writable {
+  constructor(decode: boolean, encoding: BufferEncoding) {
+    super({ decodeStrings: decode, defaultEncoding: encoding });
+  }
+  _write(chunk: any, enc: string, cb: Cb): void {
+    console.log("sink", typeof chunk, enc);
+    cb();
+  }
+}
+new Sink(flags[0]!, encodings[1] as BufferEncoding).write("y");

tests/corpus/7464-stream-option-values.mjs

@@ -0,0 +1,28 @@
+// Option values known only at run time in JavaScript: decodeStrings and
+// defaultEncoding read from an object, a write callback whose chunk is
+// then a string, decodeStrings: 0 (only false turns decoding off), a
+// defaultEncoding that is not a string, and a _construct assigned after
+// construction, which Node's constructor never reads.
+import { Writable, Readable } from "node:stream";
+const opts = { decode: false, enc: "base64" };
+const w = new Writable({
+  decodeStrings: opts.decode,
+  defaultEncoding: opts.enc,
+  write(c, e, cb) { console.log("js", typeof c, e, String(c)); cb(); },
+});
+w.write("aGk=");
+try { new Writable({ defaultEncoding: 5 }); } catch (e) { console.log("bad", e.code, e.message); }
+const zero = new Writable({ decodeStrings: 0, write(c, e, cb) { console.log("zero", Buffer.isBuffer(c), e); cb(); } });
+zero.write("z");
+const late = new Writable({ write(chunk, enc, cb) { console.log("write", chunk.toString()); cb(); } });
+let ran = false;
+late._construct = (cb) => { ran = true; cb(); };
+late.write("a");
+late.end(() => console.log("ended, construct ran:", ran));
+const r = new Readable({ read() {} });
+const order = [];
+(order.push("receiver"), r)._construct = (order.push("value"), function (cb) { console.log("never"); cb(); });
+console.log(order.join(","));
+r.push("x");
+r.push(null);
+r.on("data", (d) => console.log("data", d.toString()));

tests/corpus/7465-transform-callback-values.ts

@@ -0,0 +1,41 @@
+// A transform or flush callback passing a value that is not a Buffer or a
+// string: the push reports Node's ERR_INVALID_ARG_TYPE through 'error'
+// and destroys the stream, for option callbacks and _transform methods,
+// while Buffer and string values take the typed path.
+import { Transform } from "node:stream";
+type TransformCallback = (error?: Error | null, data?: any) => void;
+const t = new Transform({
+  transform(chunk: Buffer, _enc: BufferEncoding, cb: TransformCallback): void {
+    console.log("transform", chunk.toString());
+    cb(null, 42);
+  },
+});
+t.on("error", (e: Error & { code?: string }) => console.log("error", e.name, e.code, e.message));
+t.on("data", (d: Buffer) => console.log("data", d.toString()));
+t.on("close", () => console.log("close", t.destroyed));
+console.log("write", t.write("a"));
+const f = new Transform({
+  transform(chunk: Buffer, _enc: BufferEncoding, cb: TransformCallback): void { cb(null, chunk); },
+  flush(cb: TransformCallback): void { cb(null, { tail: true }); },
+});
+f.on("error", (e: Error & { code?: string }) => console.log("flush error", e.code, e.message));
+f.on("data", (d: Buffer) => console.log("flush data", d.toString()));
+f.on("close", () => console.log("flush close"));
+f.end("z");
+class Bad extends Transform {
+  _transform(chunk: Buffer, _enc: BufferEncoding, cb: TransformCallback): void {
+    cb(null, chunk.length);
+  }
+}
+const bad = new Bad();
+bad.on("error", (e: Error & { code?: string }) => console.log("error", e.code, e.message));
+bad.on("close", () => console.log("close"));
+bad.write("abc");
+class Good extends Transform {
+  _transform(chunk: Buffer, _enc: BufferEncoding, cb: TransformCallback): void {
+    cb(null, chunk.toString().toUpperCase());
+  }
+}
+const good = new Good();
+good.on("data", (d: Buffer) => console.log("good", d.toString()));
+good.end("xyz");

tests/harness/corpus-costs/7460-transform-writev.mjs.json

@@ -0,0 +1 @@
+{ "plain": 2266, "san": 2991 }

tests/harness/corpus-costs/7461-transform-writev.cjs.json

@@ -0,0 +1 @@
+{ "plain": 1828, "san": 2413 }

tests/harness/corpus-costs/7462-transform-writev.ts.json

@@ -0,0 +1 @@
+{ "plain": 2178, "san": 2875 }

tests/harness/corpus-costs/7463-stream-option-values.ts.json

@@ -0,0 +1 @@
+{ "plain": 1786, "san": 4192 }

tests/harness/corpus-costs/7464-stream-option-values.mjs.json

@@ -0,0 +1 @@
+{ "plain": 1678, "san": 4502 }

tests/harness/corpus-costs/7465-transform-callback-values.ts.json

@@ -0,0 +1 @@
+{ "plain": 1571, "san": 2074 }

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