Give constructor functions native stream state over the stream classes (#1533)

docs/content/limitations/builtin-function-values.mdx

@@ -19,7 +19,11 @@ Native calls accept standard process streams; custom native streams and mutation
 In JavaScript, a constructor function can inherit from `EventEmitter`, the legacy `Stream` constructor, or a stream class through `util.inherits` or `Object.setPrototypeOf`.
 `EventEmitter.call(this)` and `Stream.call(this)` initialize the object as Node does, and its listener methods keep their state in the object's own `_events`, `_eventsCount`, and `_maxListeners` properties.
 The `captureRejections` option and `events.errorMonitor` listeners are not supported on such objects.
-Calling `Readable`, `Writable`, `Duplex`, `Transform`, or `PassThrough` with a constructor function's receiver throws `SC2020` where the call runs.
+`Readable.call(this, options)` and the same call on `Writable`, `Duplex`, `Transform`, or `PassThrough` give the object a native stream.
+The stream calls the object's `_read`, `_write`, `_final`, `_transform`, `_flush`, and `_destroy` with the object as `this`, and the stream prototypes' methods and accessors work on it.
+Its listeners live in the native stream rather than in `_events`, and `_readableState` and `_writableState` are read-only views that answer the commonly read fields.
+Typed code that receives such an object as a stream class uses its native stream, and piping a native stream into it works.
+The options the native streams refuse (`objectMode`, `decodeStrings: false`, `signal`, and others) are refused here too, as are `_writev`, `_construct`, symbol event names, `listeners()`, `rawListeners()`, `wrap()`, strings written in encodings other than UTF-8, and pipes into a destination that overrides `write()` or `end()`.
 Other class constructors passed to `util.inherits` or `util.deprecate` are refused.
 Deprecation warnings are emitted on the next tick and follow assignments to `process.noDeprecation` and `process.throwDeprecation`; compiled binaries take no Node command-line flags such as `--no-deprecation` or `--trace-deprecation`, and `process.traceDeprecation` is unsupported.
 A stored `util.deprecate` reports a `length` of 4, where Node reports 3.

packages/compiler/src/backend/executable-features.ts

@@ -35,6 +35,7 @@ export function executableLinkFeatures(mod: IrModule): NativeLinkFeatures {
     searchParams: features.searchParams,
     qs: features.qs,
     dynBuiltinClass: features.dynBuiltinClass,
+    dynBuiltinStream: features.dynBuiltinStream,
     parseArgs: features.parseArgs,
     stream: features.stream,
     net: features.net,

packages/compiler/src/backend/library-pack.ts

@@ -43,6 +43,7 @@ export async function compilePackedLibrary(
     searchParams: options.searchParams ?? false,
     qs: false,
     dynBuiltinClass: false,
+    dynBuiltinStream: false,
     parseArgs: false,
     stream: false,
     net: false,

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

@@ -130,6 +130,8 @@ export interface DynHost extends WalkerHost {
   /** The shared adapter record: live array capsules read its `arrayOps`, so a
    * narrower return type would hand the native compiler a copy without it. */
   liveDynRefAdapter(t: IrType): LlStreamTypedRefAdapter;
+  /** See moduleUsesDynBuiltinStream. */
+  usesDynBuiltinStream?(): boolean;
   liveDynUnionRefAdapter(t: IrType & { kind: "union" }): string;
   dynPromiseAdapter(t: IrType): string;
   isErrorClass(className: string): boolean;
@@ -140,6 +142,19 @@ export interface DynHost extends WalkerHost {
 
 const FN_ATTRS = "#0";
 
+/** scr_dyn_builtin_stream_check's kind for a runtime stream class (the
+ * numbering of scr_dyn_builtin_class.c), or null. */
+function builtinStreamKind(className: string): number | null {
+  const kinds: Record<string, number> = {
+    "%Readable": 2,
+    "%Writable": 3,
+    "%Duplex": 4,
+    "%Transform": 5,
+    "%PassThrough": 6,
+  };
+  return kinds[className] ?? null;
+}
+
 export class LlDyn {
   private readonly dynMatchers = new Map<string, string>();
   private readonly dynBuilders = new Map<string, string>();
@@ -1491,6 +1506,24 @@ export class LlDyn {
       B.line(`${ref} = call ptr @scr_dyn_typed_ref_unbox(ptr %d)`);
       B.terminate(`ret ptr ${ref}`);
       B.startBlock(lNext);
+      // A constructor-function stream (scr_dyn_builtin_stream.c) carries
+      // its runtime stream.
+      const streamKind = t.kind === "object" ? builtinStreamKind(t.className) : null;
+      if (streamKind !== null && this.host.usesDynBuiltinStream?.() === true) {
+        host.declare(`declare ptr @scr_dyn_builtin_stream_check(ptr, double)`);
+        const own = B.tmp(),
+          hit = B.tmp();
+        B.line(
+          `${own} = call ptr @scr_dyn_builtin_stream_check(ptr %d, double ${f64Lit(streamKind)})`,
+        );
+        B.line(`${hit} = icmp ne ptr ${own}, null`);
+        const lOwn = B.newLabel("dc.ctor"),
+          lNotOwn = B.newLabel("dc.notctor");
+        B.condBr(hit, lOwn, lNotOwn);
+        B.startBlock(lOwn);
+        B.terminate(`ret ptr ${own}`);
+        B.startBlock(lNotOwn);
+      }
       if (t.kind !== "union" && preserveRefs) {
         const kind = this.kindOf(B, "%d");
         const capsule = B.tmp();

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

@@ -53,7 +53,11 @@ import {
   VOID,
   classJsName,
 } from "../../ir/ir.js";
-import { moduleRuntimeFeatures, moduleUsesMockTimers } from "../../ir/runtime-features.js";
+import {
+  moduleRuntimeFeatures,
+  moduleUsesDynBuiltinStream,
+  moduleUsesMockTimers,
+} from "../../ir/runtime-features.js";
 import { captureFreeFunctions, observesStacks } from "../stack-observations.js";
 import { type DenseArrayLoop } from "./dense-array-loops.js";
 import { findInitializedByteLoopBindings } from "./byte-windows.js";
@@ -902,6 +906,11 @@ export class LlEmitter {
   /** Identity-preserving static→dyn capsules for program classes and Web
    * APIs whose values remain directly observable. */
   readonly liveDynRefAdapters = new Map<string, LlStreamTypedRefAdapter>();
+  private dynBuiltinStreamUse: boolean | undefined;
+  /** See moduleUsesDynBuiltinStream. */
+  get usesDynBuiltinStream(): boolean {
+    return (this.dynBuiltinStreamUse ??= moduleUsesDynBuiltinStream(this.mod));
+  }
   /** Runtime-arm dispatchers for live union values. */
   readonly liveDynUnionRefAdapters = new Map<string, string>();
   readonly dynPromiseAdapters = new Map<string, string>();
@@ -1405,6 +1414,7 @@ export class LlEmitter {
       fieldAbsentTestIn: (B, value, unionId) => this.fieldAbsentTestIn(B, value, unionId),
       absentInstanceRef: (unionId) => this.absentInstanceRef(unionId),
       liveDynRefAdapter: (type) => this.liveDynRefAdapter(type),
+      usesDynBuiltinStream: () => this.usesDynBuiltinStream,
       liveDynUnionRefAdapter: (type) => this.liveDynUnionRefAdapter(type),
       dynPromiseAdapter: (type) => this.dynPromiseAdapter(type),
       isErrorClass: (name) => this.classMeta.get(name)?.root.def.name === "%Error",

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

@@ -766,8 +766,13 @@ export function streamTypedRefMaterializeAdapter(
       !meta.def.instancePrototypeHelper &&
       !meta.def.prototypeDataHelper
     ) {
-      host.declare(`declare ptr @scr_stream_dyn_view(ptr)`);
-      B.line(`${out} = call ptr @scr_stream_dyn_view(ptr %p)`);
+      // Programs with constructor-function streams view them through
+      // scr_dyn_builtin_stream.c, which answers their own objects.
+      const view = host.usesDynBuiltinStream
+        ? "scr_dyn_builtin_stream_view"
+        : "scr_stream_dyn_view";
+      host.declare(`declare ptr @${view}(ptr)`);
+      B.line(`${out} = call ptr @${view}(ptr %p)`);
     } else B.line(`${out} = call ptr @scr_dyn_new_obj()`);
     const withAttributes = meta.def.fields.some((field) => field.name === CLASS_ATTRIBUTES);
     if (meta.def.tracksOwnFields) {

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

@@ -548,7 +548,12 @@ export function emitGenericLibCall(host: LlvmEmitterContext, e: LibCallExpr): Ll
     mockTimerSym(host, e) ??
     (e.fn === "timers.queueMicrotaskDyn" && host.schedulesInContext
       ? "scr_queue_microtask_dyn_als"
-      : LIB_FN_SYMS[e.fn]);
+      : // Programs with constructor-function streams unbox untyped pipe
+        // destinations and pipeline stages through scr_dyn_builtin_stream.c,
+        // which also takes those streams.
+        e.fn === "stream.unboxDyn" && host.usesDynBuiltinStream
+        ? "scr_dyn_builtin_stream_unbox"
+        : LIB_FN_SYMS[e.fn]);
   if (sym === undefined) throw new LlvmUnsupportedError(`libCall:${e.fn}`, e.loc);
   const args = READ_ONLY_STRING_LIB_FNS.has(e.fn)
     ? emitBorrowedInputs(host, e.args)

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

@@ -1461,6 +1461,7 @@ export const LIB_FN_SYMS: Record<string, string> = {
   "dyn.classBasePrototype": "scr_dyn_class_base_prototype",
   "dyn.classInherit": "scr_dyn_class_inherit",
   "dyn.builtinClass": "scr_dyn_builtin_class",
+  "dyn.builtinStreamClass": "scr_dyn_builtin_stream_class",
   "dyn.builtinInstanceOf": "scr_dyn_builtin_instanceof",
   "dyn.functionInstanceOf": "scr_dyn_function_instanceof",
   "dyn.classSuper": "scr_dyn_class_super",
@@ -1539,6 +1540,8 @@ export function mockTimerSym(
 
 export const USES_TIMERS_LIB_FNS = new Set<string>([
   "process.emitWarning",
+  // Constructor-function streams defer their events on the stream ticks.
+  "dyn.builtinStreamClass",
   // Decipheriv.setAuthTag reports DEP0182 on the next tick.
   "crypto.cipherNew",
   // WebCrypto results settle through the deferred crypto queue.

packages/compiler/src/backend/native-link-info-core.ts

@@ -56,6 +56,7 @@ export interface NativeLinkFeatures {
   searchParams: boolean;
   qs: boolean;
   dynBuiltinClass: boolean;
+  dynBuiltinStream: boolean;
   parseArgs: boolean;
   stream: boolean;
   net: boolean;

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

@@ -163,6 +163,10 @@ export interface CcOptions {
    * scr_dyn_builtin_class.c, which also references the inspection,
    * dyn-invoke and symbol units. */
   dynBuiltinClass?: boolean;
+  /** The program uses a stream class as an untyped constructor value:
+   * compiles scr_dyn_builtin_stream.c, the constructors' stream half, which
+   * also references the stream unit. */
+  dynBuiltinStream?: boolean;
   /** The program uses native util parsing, comparison, or styling. Optional
    * utility units also require their symbol, inspection, and warning features. */
   parseArgs?: boolean;

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

@@ -250,6 +250,16 @@ const RUNTIME_UNITS: readonly RuntimeUnit[] = [
       (u.opts.emitter || u.opts.workers),
     "scr_dyn_builtin_class.c",
   ),
+  unit(
+    (u) =>
+      u.opts.dynBuiltinStream &&
+      u.opts.dynBuiltinClass &&
+      u.opts.stream &&
+      u.opts.inspect &&
+      u.opts.dynInvoke &&
+      u.opts.symbol,
+    "scr_dyn_builtin_stream.c",
+  ),
   unit((u) => u.opts.stream, "scr_stream.c", "scr_fs_stream.c"),
   // The zlib stream classes reference both the zlib engine and the
   // stream unit.

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

@@ -108,6 +108,7 @@ const features = {
   searchParams: false,
   qs: false,
   dynBuiltinClass: false,
+  dynBuiltinStream: false,
   parseArgs: false,
   stream: false,
   net: false,

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

@@ -45,6 +45,7 @@ const BASE: NativeLinkFeatures = {
   searchParams: false,
   qs: false,
   dynBuiltinClass: false,
+  dynBuiltinStream: false,
   parseArgs: false,
   stream: false,
   net: false,

packages/compiler/src/backend/warm-cache.ts

@@ -51,6 +51,7 @@ export async function warmNativeCaches(
         searchParams: false,
         qs: false,
         dynBuiltinClass: false,
+        dynBuiltinStream: false,
         parseArgs: false,
         stream: false,
         net: false,

packages/compiler/src/executable/executable-cache.test.ts

@@ -46,6 +46,7 @@ const native: EarlyExecutableNativeFeatures = {
   searchParams: false,
   qs: false,
   dynBuiltinClass: false,
+  dynBuiltinStream: false,
   parseArgs: false,
   stream: false,
   net: false,

packages/compiler/src/executable/executable-cache.ts

@@ -75,6 +75,7 @@ export interface EarlyExecutableNativeFeatures {
   searchParams: boolean;
   qs: boolean;
   dynBuiltinClass: boolean;
+  dynBuiltinStream: boolean;
   parseArgs: boolean;
   stream: boolean;
   net: boolean;
@@ -195,6 +196,7 @@ const BOOLEAN_NATIVE_KEYS = [
   "searchParams",
   "qs",
   "dynBuiltinClass",
+  "dynBuiltinStream",
   "parseArgs",
   "stream",
   "net",

packages/compiler/src/frontend/lowering/builtins/builtin-class-dyn.ts

@@ -111,8 +111,18 @@ export function builtinClassDynValueOf(lowerer: Lowerer, node: ts.Expression): I
   };
 }
 
+/** EventEmitter and the legacy Stream are dyn.builtinClass; the stream
+ * classes, which carry native stream state for constructor functions, are
+ * dyn.builtinStreamClass, so only programs that use them link the stream
+ * half. */
 export function builtinClassValue(kind: number, loc: SrcLoc): IrExpr {
-  return { kind: "libCall", fn: "dyn.builtinClass", args: [numLit(kind, loc)], type: DYN, loc };
+  return {
+    kind: "libCall",
+    fn: kind <= 1 ? "dyn.builtinClass" : "dyn.builtinStreamClass",
+    args: [numLit(kind, loc)],
+    type: DYN,
+    loc,
+  };
 }
 
 /** `Base.call(this, …)` / `Base.apply(this, args)` on one of those class

packages/compiler/src/index.ts

@@ -554,6 +554,7 @@ async function compileExecutableNative(
     searchParams: features.searchParams,
     qs: features.qs,
     dynBuiltinClass: features.dynBuiltinClass,
+    dynBuiltinStream: features.dynBuiltinStream,
     parseArgs: features.parseArgs,
     stream: features.stream,
     net: features.net,

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

@@ -3603,6 +3603,10 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
    * Transform (5) and PassThrough (6) as untyped constructor values, one
    * per thread (scr_dyn_builtin_class.c). */
   "dyn.builtinClass": { argTypes: [F64], result: DYN },
+  /** The stream classes (2-6 above) with their stream half: constructor
+   * functions over them get native stream state
+   * (scr_dyn_builtin_stream.c). */
+  "dyn.builtinStreamClass": { argTypes: [F64], result: DYN },
   /** `value instanceof C` for those constructors, over the value's
    * prototype chain. */
   "dyn.builtinInstanceOf": { argTypes: [DYN, F64], result: BOOL },

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

@@ -255,6 +255,8 @@ const CHILD_PROCESS_LIB_PREFIXES: readonly string[] = ["child.", "cp.", "spawnRe
  * emitter unit, so each of them links all five. */
 const BUILTIN_CLASS_LIB_FNS: readonly IrLibFn[] = [
   "dyn.builtinClass",
+  // the stream classes, through scr_dyn_builtin_stream.c
+  "dyn.builtinStreamClass",
   "dyn.builtinInstanceOf",
   "dyn.functionInstanceOf",
 ];
@@ -546,6 +548,10 @@ const RUNTIME_FEATURE_RULES = {
   dynBuiltinClass: rule({
     fns: [...BUILTIN_CLASS_LIB_FNS],
   }),
+  /** The stream classes as untyped constructor values with their stream
+   * half (scr_dyn_builtin_stream.c): native stream state for constructor
+   * functions, over the stream unit. */
+  dynBuiltinStream: rule({ fns: ["dyn.builtinStreamClass"] }),
   dynInvoke: rule({
     initial: (mod) => mod.workers === true,
     fns: [...DYN_INVOKE_LIB_FNS, ...UTIL_CALLABLE_LIB_FNS, ...BUILTIN_CLASS_LIB_FNS],
@@ -675,7 +681,7 @@ const RUNTIME_FEATURE_RULES = {
   }),
   stream: rule({
     initial: (mod) => (mod.classes ?? []).some((c) => RUNTIME_STREAM_CLASSES.has(c.name)),
-    fns: [...STREAM_LIB_FNS],
+    fns: [...STREAM_LIB_FNS, "dyn.builtinStreamClass"],
     prefixes: STREAM_LIB_PREFIXES,
   }),
   /** A Dirent row converted to dyn carries Node's Symbol(type) key
@@ -720,6 +726,7 @@ function noRuntimeFeatures(): RuntimeFeatures {
     dnsPromises: false,
     dynAsync: false,
     dynBuiltinClass: false,
+    dynBuiltinStream: false,
     dynInvoke: false,
     dynTier: false,
     emitter: false,
@@ -887,6 +894,14 @@ export function moduleUsesStream(mod: IrModule): boolean {
   return scanRuntimeFeatures(mod, "stream").stream;
 }
 
+/** True when the program links scr_dyn_builtin_stream.c (stream classes
+ * as untyped constructor values): its runtime streams box through that
+ * unit's view and checked casts to the stream classes accept constructor-
+ * function streams. Programs without it keep their exact code. */
+export function moduleUsesDynBuiltinStream(mod: IrModule): boolean {
+  return scanRuntimeFeatures(mod, "dynBuiltinStream").dynBuiltinStream;
+}
+
 export const NPM_COMPRESS_MIN = 1024;
 
 export function moduleUsesZlib(mod: IrModule): boolean {

packages/compiler/src/native/cli.ts

@@ -90,6 +90,7 @@ async function main(): Promise<number> {
               searchParams: false,
               qs: false,
               dynBuiltinClass: false,
+              dynBuiltinStream: false,
               parseArgs: false,
               stream: false,
               net: false,

packages/compiler/src/native/prepare.ts

@@ -144,6 +144,7 @@ function validFeatures(value: unknown): value is NativeLinkFeatures {
     "searchParams",
     "qs",
     "dynBuiltinClass",
+    "dynBuiltinStream",
     "parseArgs",
     "stream",
     "net",

packages/compiler/src/native/sanitizer.test.ts

@@ -99,6 +99,7 @@ test.skipIf(process.platform !== "linux" && process.platform !== "darwin")(
               searchParams: false,
               qs: false,
               dynBuiltinClass: false,
+              dynBuiltinStream: false,
               parseArgs: false,
               stream: false,
               net: false,

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6560-constructor-function-readable.cjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6560-constructor-function-readable.cjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6561-constructor-function-writable.cjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6561-constructor-function-writable.cjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6562-constructor-function-transform.cjs.json

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

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6563-constructor-function-stream-errors.cjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6563-constructor-function-stream-errors.cjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6564-constructor-function-stream-late-hooks.cjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6564-constructor-function-stream-late-hooks.cjs"
+ ],
+ "diags": []
+}

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

@@ -96,6 +96,19 @@ export const OPTIONAL_RUNTIME_SOURCES = [
   // through the dyn-invoke unit, inspect unhandled 'error' payloads and
   // create EventEmitter's private symbols.
   ["scr_dyn_builtin_class.c", all("dynBuiltinClass", "inspect", "dynInvoke", "symbol", "emitter")],
+  // Their stream half also references the stream unit.
+  [
+    "scr_dyn_builtin_stream.c",
+    all(
+      "dynBuiltinStream",
+      "dynBuiltinClass",
+      "stream",
+      "inspect",
+      "dynInvoke",
+      "symbol",
+      "emitter",
+    ),
+  ],
   ["scr_stream.c", "stream"],
   // The zlib stream classes reference both the zlib engine and the stream unit.
   ["scr_zlib_stream.c", all("zlib", "stream")],

packages/runtime/src/scr_dyn.h

@@ -1159,6 +1159,39 @@ ScrDyn *scr_dyn_new_class(ScrClassObj *cls, const char *type_key);
 ScrDyn *scr_dyn_builtin_class(double kind);
 bool scr_dyn_builtin_instanceof(ScrDyn *value, double kind);
 bool scr_dyn_function_instanceof(ScrDyn *value, ScrDyn *callee);
+/* The stream half of those constructors (scr_dyn_builtin_stream.c, linked
+ * with the dynBuiltinStream feature), registered with
+ * scr_dyn_builtin_class_streams before the first stream class value is
+ * read: construct runs `Readable.call(this, opts)` and the other stream
+ * kinds (+1 or NULL with an exception pending), prototype installs a stream
+ * kind's methods, and event answers EventEmitter.prototype's methods for a
+ * receiver with stream state (true when it handled the call; *result +1 or
+ * NULL with an exception pending). */
+enum {
+  SCR_EE_M_ON, SCR_EE_M_PREPEND, SCR_EE_M_ONCE, SCR_EE_M_PREPEND_ONCE, SCR_EE_M_OFF,
+  SCR_EE_M_REMOVE_ALL, SCR_EE_M_EMIT, SCR_EE_M_LISTENERS, SCR_EE_M_RAW_LISTENERS,
+  SCR_EE_M_LISTENER_COUNT, SCR_EE_M_EVENT_NAMES, SCR_EE_M_SET_MAX, SCR_EE_M_GET_MAX,
+};
+typedef struct ScrDynStreamOps {
+  ScrDyn *(*construct)(int kind, ScrDyn *self, ScrDyn *const *args, size_t argc);
+  void (*prototype)(int kind, ScrDyn *prototype);
+  bool (*event)(int method, ScrDyn *self, ScrDyn *const *args, size_t argc, ScrDyn **result);
+} ScrDynStreamOps;
+void scr_dyn_builtin_class_streams(const ScrDynStreamOps *ops);
+/* A stream class value with its stream half (scr_dyn_builtin_stream.c),
+ * +1; the stream a JavaScript value stands for (a constructor-function
+ * stream or a boxed runtime stream), +1 or NULL; and the constructor-
+ * function object of a runtime stream, +1 or NULL. */
+ScrDyn *scr_dyn_builtin_stream_class(double kind);
+struct ScrStream *scr_dyn_builtin_stream_of(const ScrDyn *value);
+ScrDyn *scr_dyn_builtin_stream_object(struct ScrStream *s);
+/* The untyped view of a runtime stream in programs that link that unit
+ * (+1), a checked cast to a runtime stream class kind (+1 or NULL), and
+ * the stream.unboxDyn libCall in those programs (scr_stream_unbox_dyn's
+ * contract, also taking constructor-function streams). */
+ScrDyn *scr_dyn_builtin_stream_view(struct ScrStream *s);
+struct ScrStream *scr_dyn_builtin_stream_check(const ScrDyn *value, double kind);
+struct ScrStream *scr_dyn_builtin_stream_unbox(const ScrDyn *value, const ScrStr *message);
 bool scr_dyn_class_is_key(const ScrDyn *value, const ScrStr *type_key);
 bool scr_dyn_class_is(const ScrDyn *value, const char *type_key);
 /* Calls a dyn value: a non-function kind throws the catchable TypeError

packages/runtime/src/scr_dyn_builtin_class.c

@@ -15,9 +15,13 @@
  * Stream is Node's legacy constructor: EventEmitter.call(this, opts) and a
  * prototype that inherits EventEmitter's. Readable, Writable, Duplex,
  * Transform and PassThrough have prototypes chained as in Node, so
- * util.inherits onto them and instanceof work, but their native state has
- * no dyn object form: calling one with a receiver throws the "not
- * supported yet" Error where the construction runs.
+ * util.inherits onto them and instanceof work. Their stream half (the
+ * prototypes' methods and `Readable.call(this, opts)`) lives in
+ * scr_dyn_builtin_stream.c, which registers itself through
+ * scr_dyn_builtin_class_streams in programs that link it; without it,
+ * calling one with a receiver throws the "not supported yet" Error where
+ * the construction runs. EventEmitter's methods hand a receiver with
+ * stream state to that unit, whose listeners live in the runtime stream.
  *
  * One constructor per kind and thread, so every boxing of the same class
  * value is the same function. All arguments are borrowed. */
@@ -59,6 +63,14 @@ static SCR_TL ScrDyn *bc_symbols[BC_SYMBOLS];
 
 static ScrDyn *bc_class(int kind);
 
+/* The stream half, once registered (scr_dyn_builtin_stream.c). */
+static SCR_TL const ScrDynStreamOps *bc_stream_ops;
+
+/* EventEmitter.prototype's method on a receiver with stream state. */
+static bool bc_stream_event(int method, ScrDyn *self, ScrDyn *const *args, size_t argc, ScrDyn **result) {
+  return bc_stream_ops != NULL && bc_stream_ops->event(method, self, args, argc, result);
+}
+
 /* ── property helpers ───────────────────────────────────────────────── */
 
 static ScrDyn *bc_key(const char *name) {
@@ -585,7 +597,12 @@ static ScrDyn *bc_once_wrap(ScrDyn *target, ScrDyn *type, ScrDyn *listener) {
 static ScrDyn *bc_on(ScrClosure *closure, ScrDyn *const *args, size_t argc) {
   (void)closure;
   ScrDyn *self = bc_this();
-  ScrDyn *result = bc_require_object(self, "_events")
+  ScrDyn *result;
+  if (bc_stream_event(SCR_EE_M_ON, self, args, argc, &result)) {
+    scr_dyn_release(self);
+    return result;
+  }
+  result = bc_require_object(self, "_events")
       ? bc_add_listener(self, bc_arg(args, argc, 0), bc_arg(args, argc, 1), false) : NULL;
   scr_dyn_release(self);
   return result;
@@ -594,7 +611,12 @@ static ScrDyn *bc_on(ScrClosure *closure, ScrDyn *const *args, size_t argc) {
 static ScrDyn *bc_prepend(ScrClosure *closure, ScrDyn *const *args, size_t argc) {
   (void)closure;
   ScrDyn *self = bc_this();
-  ScrDyn *result = bc_require_object(self, "_events")
+  ScrDyn *result;
+  if (bc_stream_event(SCR_EE_M_PREPEND, self, args, argc, &result)) {
+    scr_dyn_release(self);
+    return result;
+  }
+  result = bc_require_object(self, "_events")
       ? bc_add_listener(self, bc_arg(args, argc, 0), bc_arg(args, argc, 1), true) : NULL;
   scr_dyn_release(self);
   return result;
@@ -605,6 +627,10 @@ static ScrDyn *bc_once_common(ScrDyn *const *args, size_t argc, bool prepend) {
   ScrDyn *type = bc_arg(args, argc, 0);
   ScrDyn *listener = bc_arg(args, argc, 1);
   ScrDyn *result = NULL;
+  if (bc_stream_event(prepend ? SCR_EE_M_PREPEND_ONCE : SCR_EE_M_ONCE, self, args, argc, &result)) {
+    scr_dyn_release(self);
+    return result;
+  }
   if (bc_check_listener(listener) && bc_require_object(self, "_events")) {
     ScrDyn *wrapper = bc_once_wrap(self, type, listener);
     if (wrapper) {
@@ -629,7 +655,12 @@ static ScrDyn *bc_prepend_once(ScrClosure *closure, ScrDyn *const *args, size_t
 static ScrDyn *bc_off(ScrClosure *closure, ScrDyn *const *args, size_t argc) {
   (void)closure;
   ScrDyn *self = bc_this();
-  ScrDyn *result = bc_require_object(self, "_events")
+  ScrDyn *result;
+  if (bc_stream_event(SCR_EE_M_OFF, self, args, argc, &result)) {
+    scr_dyn_release(self);
+    return result;
+  }
+  result = bc_require_object(self, "_events")
       ? bc_remove_listener(self, bc_arg(args, argc, 0), bc_arg(args, argc, 1)) : NULL;
   scr_dyn_release(self);
   return result;
@@ -641,6 +672,11 @@ static ScrDyn *bc_off(ScrClosure *closure, ScrDyn *const *args, size_t argc) {
 static ScrDyn *bc_remove_all(ScrClosure *closure, ScrDyn *const *args, size_t argc) {
   (void)closure;
   ScrDyn *self = bc_this();
+  ScrDyn *handled;
+  if (bc_stream_event(SCR_EE_M_REMOVE_ALL, self, args, argc, &handled)) {
+    scr_dyn_release(self);
+    return handled;
+  }
   if (!bc_require_object(self, "_events")) {
     scr_dyn_release(self);
     return NULL;
@@ -759,6 +795,11 @@ fail:
 static ScrDyn *bc_emit(ScrClosure *closure, ScrDyn *const *args, size_t argc) {
   (void)closure;
   ScrDyn *self = bc_this();
+  ScrDyn *handled;
+  if (bc_stream_event(SCR_EE_M_EMIT, self, args, argc, &handled)) {
+    scr_dyn_release(self);
+    return handled;
+  }
   if (!bc_require_object(self, "_events")) {
     scr_dyn_release(self);
     return NULL;
@@ -844,6 +885,11 @@ fail_self:
 /* _listeners(target, type, unwrap): a fresh array. */
 static ScrDyn *bc_listeners_of(bool unwrap, ScrDyn *const *args, size_t argc) {
   ScrDyn *self = bc_this();
+  ScrDyn *handled;
+  if (bc_stream_event(unwrap ? SCR_EE_M_LISTENERS : SCR_EE_M_RAW_LISTENERS, self, args, argc, &handled)) {
+    scr_dyn_release(self);
+    return handled;
+  }
   if (!bc_require_object(self, "_events")) {
     scr_dyn_release(self);
     return NULL;
@@ -937,7 +983,12 @@ static ScrDyn *bc_count_listeners(ScrDyn *self, ScrDyn *type, ScrDyn *only) {
 static ScrDyn *bc_listener_count(ScrClosure *closure, ScrDyn *const *args, size_t argc) {
   (void)closure;
   ScrDyn *self = bc_this();
-  ScrDyn *result = bc_require_object(self, "_events")
+  ScrDyn *result;
+  if (bc_stream_event(SCR_EE_M_LISTENER_COUNT, self, args, argc, &result)) {
+    scr_dyn_release(self);
+    return result;
+  }
+  result = bc_require_object(self, "_events")
       ? bc_count_listeners(self, bc_arg(args, argc, 0),
                            argc > 1 && !bc_undefined(args[1]) && args[1]->kind != SCR_DYN_NULL ? args[1] : NULL)
       : NULL;
@@ -964,6 +1015,11 @@ static ScrDyn *bc_event_names(ScrClosure *closure, ScrDyn *const *args, size_t a
   (void)args;
   (void)argc;
   ScrDyn *self = bc_this();
+  ScrDyn *handled;
+  if (bc_stream_event(SCR_EE_M_EVENT_NAMES, self, args, argc, &handled)) {
+    scr_dyn_release(self);
+    return handled;
+  }
   if (!bc_require_object(self, "_eventsCount")) {
     scr_dyn_release(self);
     return NULL;
@@ -1008,6 +1064,11 @@ static ScrDyn *bc_set_max(ScrClosure *closure, ScrDyn *const *args, size_t argc)
     scr_dyn_release(self);
     return NULL;
   }
+  ScrDyn *handled;
+  if (bc_stream_event(SCR_EE_M_SET_MAX, self, args, argc, &handled)) {
+    scr_dyn_release(self);
+    return handled;
+  }
   if (!bc_require_object(self, "_maxListeners") || !bc_set(self, "_maxListeners", scr_dyn_retain(n))) {
     scr_dyn_release(self);
     return NULL;
@@ -1056,6 +1117,11 @@ static ScrDyn *bc_get_max(ScrClosure *closure, ScrDyn *const *args, size_t argc)
   (void)args;
   (void)argc;
   ScrDyn *self = bc_this();
+  ScrDyn *handled;
+  if (bc_stream_event(SCR_EE_M_GET_MAX, self, args, argc, &handled)) {
+    scr_dyn_release(self);
+    return handled;
+  }
   double max = bc_require_object(self, "_maxListeners") ? bc_max_listeners(self) : -1;
   scr_dyn_release(self);
   if (scr_exc_pending()) return NULL;
@@ -1321,6 +1387,8 @@ static ScrDyn *bc_construct(ScrClosure *closure, ScrDyn *const *args, size_t arg
   ScrDyn *result = NULL;
   if (kind == BC_EMITTER || kind == BC_STREAM) {
     if (bc_emitter_init(self, bc_arg(args, argc, 0))) result = scr_dyn_retain(scr_dyn_undefined());
+  } else if (bc_stream_ops) {
+    result = bc_stream_ops->construct(kind, self, args, argc);
   } else {
     char message[160];
     snprintf(message, sizeof message,
@@ -1414,6 +1482,8 @@ static ScrDyn *bc_new_class(int kind, ScrDyn *parent) {
   } else if (kind == BC_STREAM) {
     ScrDyn *pipe = scr_dyn_new_func(scr_closure_new(NULL, 0), bc_pipe, 2, "native:stream", "pipe");
     bc_set(prototype, "pipe", pipe);
+  } else if (bc_stream_ops) {
+    bc_stream_ops->prototype(kind, prototype);
   }
   scr_dyn_release(prototype);
   if (scr_exc_pending()) scr_exc_clear();
@@ -1449,6 +1519,22 @@ static ScrDyn *bc_class(int kind) {
   return bc_ctors[kind];
 }
 
+/* Registers the stream half; constructors that already exist get their
+ * stream prototypes' methods now. */
+void scr_dyn_builtin_class_streams(const ScrDynStreamOps *ops) {
+  if (bc_stream_ops) return;
+  bc_stream_ops = ops;
+  for (int kind = BC_READABLE; kind < BC_COUNT; kind++) {
+    if (!bc_ctors[kind]) continue;
+    ScrDyn *prototype = bc_get(bc_ctors[kind], "prototype");
+    if (prototype) {
+      ops->prototype(kind, prototype);
+      scr_dyn_release(prototype);
+    }
+    if (scr_exc_pending()) scr_exc_clear();
+  }
+}
+
 /* The constructor value for kind (the BC_* order), +1. */
 ScrDyn *scr_dyn_builtin_class(double kind) {
   int index = (int)kind;

packages/runtime/src/scr_stream.c

@@ -2254,6 +2254,8 @@ void scr_stream_set_flush(ScrStream *s, ScrClosure *cb /*moves*/, ScrStreamPlain
 
 void scr_stream_set_flush_on_prefinish(ScrStream *s) { s->st->flush_on_prefinish = true; }
 
+void scr_stream_set_destroyed(ScrStream *s, bool destroyed) { s->st->destroyed = destroyed; }
+
 void scr_stream_set_native(ScrStream *s, void *state /*moves*/, void (*release)(void *)) {
   ScrStreamState *st = s->st;
   if (st->native) st->native_release(st->native);

packages/runtime/src/scr_stream.h

@@ -449,6 +449,9 @@ void scr_stream_set_flush(ScrStream *s, ScrClosure *cb, ScrStreamPlainInv inv);
  * released when the stream is freed; scr_stream_native answers it
  * (borrowed) when it was set with `release`, else NULL. */
 void scr_stream_set_flush_on_prefinish(ScrStream *s);
+/* `stream.destroyed = value` (scr_dyn_builtin_stream.c): Node's setter
+ * stores the states' flag without destroying the stream. */
+void scr_stream_set_destroyed(ScrStream *s, bool destroyed);
 void scr_stream_set_native(ScrStream *s, void *state, void (*release)(void *));
 void *scr_stream_native(ScrStream *s, void (*release)(void *));
 

tests/corpus/6560-constructor-function-readable.cjs

@@ -0,0 +1,121 @@
+// Readable streams built by constructor functions, the from2 pattern:
+// util.inherits(Ctor, Readable), Readable.call(this, opts) and _read on
+// the prototype.
+const { Readable } = require("stream");
+const util = require("util");
+
+function Source(list, opts) {
+  if (!(this instanceof Source)) return new Source(list, opts);
+  this._list = list.slice();
+  // Before Readable.call there is no state: Node's setter ignores this.
+  this.destroyed = false;
+  Readable.call(this, opts || {});
+  this._reads = 0;
+  console.log("hwm", this._readableState.highWaterMark, "objectMode", this._readableState.objectMode);
+}
+util.inherits(Source, Readable);
+Source.prototype._read = function (size) {
+  this._reads++;
+  const item = this._list.shift();
+  this.push(item === undefined ? null : Buffer.from(item));
+};
+
+const s = Source(["alpha", "beta", "gamma"]);
+console.log("instanceof", s instanceof Source, s instanceof Readable, Source.super_ === Readable);
+console.log("readable", s.readable, "ended", s.readableEnded, "destroyed", s.destroyed);
+console.log("flowing", s.readableFlowing, s._readableState.flowing);
+s.on("data", function (chunk) {
+  console.log("data", this === s, Buffer.isBuffer(chunk), chunk.toString());
+});
+s.on("end", function () {
+  console.log("end", this === s, s.readableEnded, s._readableState.ended, s._reads);
+});
+s.on("close", () => {
+  console.log("close", s.destroyed, s.closed);
+  paused();
+});
+console.log("sync after on('data')", s.readableFlowing);
+
+// from2's callback-driven _read, asynchronous pushes and a small
+// highWaterMark.
+function From(read, opts) {
+  if (!(this instanceof From)) return new From(read, opts);
+  this._reading = false;
+  this._callback = check;
+  this.destroyed = false;
+  Readable.call(this, opts);
+  const self = this;
+  const hwm = this._readableState.highWaterMark;
+  function check(err, data) {
+    if (self.destroyed) return;
+    if (err) return self.destroy(err);
+    if (data === null) return self.push(null);
+    self._reading = false;
+    if (self.push(data)) self._read(hwm);
+  }
+  this._from = read;
+}
+util.inherits(From, Readable);
+From.prototype._read = function (size) {
+  if (this._reading || this.destroyed) return;
+  this._reading = true;
+  this._from(size, this._callback);
+};
+From.prototype.destroy = function (err) {
+  if (this.destroyed) return;
+  this.destroyed = true;
+  const self = this;
+  process.nextTick(function () {
+    if (err) self.emit("error", err);
+    self.emit("close");
+  });
+};
+
+function paused() {
+  const parts = ["one ", "two ", "three"];
+  const r = new From(function (size, next) {
+    setImmediate(() => next(null, parts.length ? parts.shift() : null));
+  }, { highWaterMark: 4 });
+  console.log("from hwm", r._readableState.highWaterMark, r.readableHighWaterMark);
+  r.setEncoding("utf8");
+  let text = "";
+  r.on("readable", () => {
+    let chunk;
+    while ((chunk = r.read()) !== null) text += chunk;
+  });
+  r.on("end", () => {
+    console.log("paused text", JSON.stringify(text), typeof text);
+    // From() without new runs outside the listener, where `this` is unset.
+    setImmediate(failing);
+  });
+}
+
+function failing() {
+  let calls = 0;
+  const r = From(function (size, next) {
+    calls++;
+    if (calls === 2) return next(new Error("source failed"));
+    next(null, "chunk" + calls);
+  });
+  r.on("data", (d) => console.log("failing data", String(d)));
+  r.on("error", (e) => console.log("failing error", e.message, r.destroyed));
+  r.on("close", () => {
+    console.log("failing close", calls);
+    unshifted();
+  });
+}
+
+function unshifted() {
+  const r = new Source(["xyz"]);
+  r.once("readable", () => {
+    const first = r.read(1);
+    console.log("read(1)", String(first), r.readableLength);
+    r.unshift(first);
+    console.log("after unshift", r.readableLength);
+    r.pause();
+    console.log("isPaused", r.isPaused());
+    r.on("data", (d) => console.log("rest", String(d)));
+    r.resume();
+  });
+  r.on("end", () => console.log("unshifted end"));
+}

tests/corpus/6561-constructor-function-writable.cjs

@@ -0,0 +1,141 @@
+// Writable and Duplex streams built by constructor functions, the bl and
+// tar-stream patterns: Writable.call(this, opts), Duplex.call(this),
+// _write/_final/_read on the prototype, listeners added before the base
+// constructor runs, and an end() override that calls the base's end.
+const stream = require("stream");
+const util = require("util");
+
+function Sink(opts) {
+  if (!(this instanceof Sink)) return new Sink(opts);
+  stream.Writable.call(this, opts);
+  this.chunks = [];
+  this.finalized = false;
+}
+util.inherits(Sink, stream.Writable);
+Sink.prototype._write = function (chunk, enc, cb) {
+  console.log("_write", this instanceof Sink, enc, Buffer.isBuffer(chunk), chunk.toString());
+  this.chunks.push(chunk);
+  if (this.chunks.length === 2) setImmediate(cb);
+  else cb();
+};
+Sink.prototype._final = function (cb) {
+  console.log("_final", this.chunks.length);
+  this.finalized = true;
+  cb();
+};
+
+const sink = new Sink({ highWaterMark: 8 });
+console.log("writable", sink.writable, sink.writableHighWaterMark, sink._writableState.highWaterMark);
+console.log("write a", sink.write("first"));
+console.log("write b", sink.write(Buffer.from("second"), () => console.log("write b callback")));
+console.log("needDrain", sink.writableNeedDrain, sink.writableLength);
+sink.on("drain", () => console.log("drain", sink.writableLength));
+sink.on("finish", function () {
+  console.log("finish", this === sink, sink.finalized, sink.writableFinished, Buffer.concat(sink.chunks).toString());
+});
+sink.on("close", () => {
+  console.log("close", sink.destroyed, sink.closed);
+  duplex();
+});
+sink.end("third", (err) => console.log("end callback", err, sink.writableEnded, sink._writableState.ended));
+console.log("after end", sink.writableEnded, sink.writable);
+
+// A bl-style list: a Duplex that collects writes and reads them back.
+function BufferListStream(callback) {
+  if (!(this instanceof BufferListStream)) return new BufferListStream(callback);
+  if (typeof callback === "function") {
+    this._callback = callback;
+    const piper = function piper(err) {
+      if (this._callback) {
+        this._callback(err);
+        this._callback = null;
+      }
+    }.bind(this);
+    this.on("pipe", function onPipe(src) {
+      console.log("pipe event", src instanceof stream.Readable);
+      src.on("error", piper);
+    });
+    this.on("unpipe", function onUnpipe(src) {
+      console.log("unpipe event");
+      src.removeListener("error", piper);
+    });
+    callback = null;
+  }
+  this._bufs = [];
+  this.length = 0;
+  stream.Duplex.call(this);
+}
+util.inherits(BufferListStream, stream.Duplex);
+BufferListStream.prototype._write = function (buf, encoding, callback) {
+  this._bufs.push(buf);
+  this.length += buf.length;
+  callback();
+};
+BufferListStream.prototype._read = function (size) {
+  if (!this.length) return this.push(null);
+  const all = Buffer.concat(this._bufs);
+  size = Math.min(size, this.length, 4);
+  this.push(all.subarray(0, size));
+  this._bufs = [all.subarray(size)];
+  this.length -= size;
+};
+BufferListStream.prototype.end = function end(chunk) {
+  stream.Duplex.prototype.end.call(this, chunk);
+  if (this._callback) {
+    this._callback(null, Buffer.concat(this._bufs));
+    this._callback = null;
+  }
+};
+BufferListStream.prototype.slice = function () {
+  return Buffer.concat(this._bufs);
+};
+
+function Letters(text) {
+  stream.Readable.call(this, { highWaterMark: 2 });
+  this._text = text;
+}
+util.inherits(Letters, stream.Readable);
+Letters.prototype._read = function () {
+  const next = this._text.slice(0, 3);
+  this._text = this._text.slice(3);
+  this.push(next.length ? next : null);
+};
+
+function duplex() {
+  const list = new BufferListStream((err, data) => {
+    console.log("bl callback", err, data.toString(), data.length);
+  });
+  console.log("duplex", list.readable, list.writable, list.allowHalfOpen, list instanceof stream.Duplex);
+  console.log("pre-constructor listeners", list.listenerCount("pipe"), list.listenerCount("unpipe"));
+  list.write("hello ");
+  list.write("constructor ");
+  list.end("functions");
+  list.on("finish", () => {
+    const out = [];
+    list.on("data", (d) => out.push(d.toString()));
+    list.on("end", () => {
+      console.log("read back", JSON.stringify(out));
+      piping();
+    });
+  });
+}
+
+function piping() {
+  const src = new Letters("abcdefghij");
+  const dst = new Sink({ highWaterMark: 1 });
+  dst.on("pipe", (from) => console.log("dst pipe", from === src));
+  const returned = src.pipe(dst);
+  console.log("pipe returns dest", returned === dst);
+  dst.on("finish", () => {
+    console.log("piped", Buffer.concat(dst.chunks).toString(), src.readableEnded);
+    const keep = new Sink();
+    const from = new Letters("xyz");
+    from.pipe(keep, { end: false });
+    from.on("end", () => {
+      setImmediate(() => {
+        console.log("end:false keeps the destination open", keep.writable, keep.writableEnded);
+        keep.end();
+      });
+    });
+  });
+}

tests/corpus/6562-constructor-function-transform.cjs

@@ -0,0 +1,143 @@
+// Transform and PassThrough streams built by constructor functions (the
+// yazl and tar-stream patterns), piped together with runtime streams:
+// zlib and fs.
+const { Readable, Writable, Transform, PassThrough } = require("stream");
+const util = require("util");
+const zlib = require("zlib");
+const fs = require("fs");
+const os = require("os");
+const path = require("path");
+
+// yazl's ByteCounter
+function ByteCounter(options) {
+  Transform.call(this, options);
+  this.byteCount = 0;
+}
+util.inherits(ByteCounter, Transform);
+ByteCounter.prototype._transform = function (chunk, encoding, cb) {
+  this.byteCount += chunk.length;
+  cb(null, chunk);
+};
+
+// A transform with a flush that appends a trailer.
+function Upper() {
+  if (!(this instanceof Upper)) return new Upper();
+  Transform.call(this);
+  this.seen = 0;
+}
+util.inherits(Upper, Transform);
+Upper.prototype._transform = function (chunk, encoding, cb) {
+  this.seen++;
+  this.push(chunk.toString().toUpperCase());
+  cb();
+};
+Upper.prototype._flush = function (cb) {
+  cb(null, "[" + this.seen + " chunks]");
+};
+
+// tar-stream's entry stream: a PassThrough with its own destroy.
+function Source(parent, offset) {
+  this._parent = parent;
+  this.offset = offset;
+  PassThrough.call(this, { autoDestroy: false });
+}
+util.inherits(Source, PassThrough);
+Source.prototype.destroy = function (err) {
+  console.log("Source.destroy forwarded", err && err.message);
+  this._parent.destroyed = true;
+};
+
+// A PassThrough without its own _transform passes chunks through.
+function Plain() {
+  PassThrough.call(this);
+}
+util.inherits(Plain, PassThrough);
+
+function Words(words) {
+  Readable.call(this);
+  this._words = words.slice();
+}
+util.inherits(Words, Readable);
+Words.prototype._read = function () {
+  this.push(this._words.length ? this._words.shift() : null);
+};
+
+function Collect() {
+  Writable.call(this);
+  this.parts = [];
+}
+util.inherits(Collect, Writable);
+Collect.prototype._write = function (chunk, enc, cb) {
+  this.parts.push(chunk);
+  cb();
+};
+Collect.prototype.text = function () {
+  return Buffer.concat(this.parts).toString();
+};
+
+const upper = Upper();
+const words = [];
+upper.on("data", (d) => words.push(d.toString()));
+upper.on("end", () => {
+  console.log("upper", JSON.stringify(words), upper.seen, upper._readableState.ended, upper._writableState.finished);
+  counter();
+});
+upper.write("one ");
+upper.write(Buffer.from("two "));
+upper.end("three");
+
+function counter() {
+  const counterStream = new ByteCounter();
+  const sink = new Collect();
+  new Words(["hello ", "byte ", "counter"]).pipe(counterStream).pipe(sink);
+  sink.on("finish", () => {
+    console.log("counted", counterStream.byteCount, JSON.stringify(sink.text()));
+    entry();
+  });
+}
+
+function entry() {
+  const parent = { destroyed: false };
+  const source = new Source(parent, 512);
+  const plain = new Plain();
+  console.log("source", source.offset, source instanceof PassThrough, source instanceof Transform);
+  let body = "";
+  source.pipe(plain);
+  plain.on("data", (d) => (body += d));
+  plain.on("end", () => {
+    console.log("entry body", JSON.stringify(body), source.destroyed);
+    source.destroy(new Error("stop"));
+    console.log("parent destroyed", parent.destroyed);
+    gzip();
+  });
+  source.write("file ");
+  source.end("contents");
+}
+
+function gzip() {
+  const counted = new ByteCounter();
+  const out = new Collect();
+  new Words(["compressed ", "through ", "zlib"]).pipe(counted).pipe(zlib.createGzip()).pipe(out);
+  out.on("finish", () => {
+    const back = zlib.gunzipSync(Buffer.concat(out.parts)).toString();
+    console.log("gzip roundtrip", JSON.stringify(back), counted.byteCount);
+    toFile();
+  });
+}
+
+function toFile() {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), "ctor-stream-"));
+  const file = path.join(dir, "out.txt");
+  const ws = fs.createWriteStream(file);
+  new Words(["written ", "by ", "a ", "constructor"]).pipe(Upper()).pipe(ws);
+  ws.on("finish", () => {
+    console.log("file", JSON.stringify(fs.readFileSync(file, "utf8")));
+    // A runtime stream piped into a constructor-function stream.
+    const back = new Collect();
+    fs.createReadStream(file).pipe(new ByteCounter()).pipe(back);
+    back.on("finish", () => {
+      console.log("read back", JSON.stringify(back.text()));
+      fs.rmSync(dir, { recursive: true });
+    });
+  });
+}

tests/corpus/6563-constructor-function-stream-errors.cjs

@@ -0,0 +1,137 @@
+// Errors, destroy and custom events on constructor-function streams.
+const { Readable, Writable, Duplex } = require("stream");
+const util = require("util");
+
+function Failing() {
+  Writable.call(this);
+}
+util.inherits(Failing, Writable);
+Failing.prototype._write = function (chunk, enc, cb) {
+  const err = new Error("cannot store " + chunk);
+  err.code = "E_STORE";
+  this.lastError = err;
+  cb(err);
+};
+
+const failing = new Failing();
+failing.on("error", function (err) {
+  console.log("write error", this === failing, err === failing.lastError, err.message, err.code);
+  console.log("errored", failing.errored === err, failing.destroyed, failing.writable);
+});
+failing.on("close", () => {
+  console.log("write close");
+  destroying();
+});
+failing.write("x", (err) => console.log("write callback", err && err.code));
+
+function Ticker() {
+  Readable.call(this);
+  this.n = 0;
+}
+util.inherits(Ticker, Readable);
+Ticker.prototype._read = function () {
+  this.push(String(this.n++));
+};
+Ticker.prototype._destroy = function (err, cb) {
+  console.log("_destroy", err && err.message, this.n > 0);
+  setImmediate(() => cb(err && new Error("wrapped: " + err.message)));
+};
+
+function destroying() {
+  const t = new Ticker();
+  const order = [];
+  t.on("data", (d) => {
+    if (d.toString() === "3") t.destroy(new Error("enough"), (e) => order.push("destroy callback " + (e && e.message)));
+  });
+  t.on("error", (e) => order.push("error " + e.message));
+  t.on("close", () => {
+    order.push("close");
+    console.log(order.join(" | "), t.destroyed, t.readable);
+    t.destroy(null, () => console.log("destroy after close calls back"));
+    notImplemented();
+  });
+}
+
+function Bare() {
+  Readable.call(this);
+}
+util.inherits(Bare, Readable);
+
+function Throws() {
+  Readable.call(this);
+}
+util.inherits(Throws, Readable);
+Throws.prototype._read = function () {
+  throw new TypeError("read threw");
+};
+
+function notImplemented() {
+  const bare = new Bare();
+  bare.on("error", (e) => {
+    console.log("bare", e.code, e.message);
+    const thrower = new Throws();
+    thrower.on("error", (err) => {
+      console.log("throws", err instanceof TypeError, err.message, thrower.destroyed);
+      afterEnd();
+    });
+    thrower.resume();
+  });
+  bare.resume();
+}
+
+function Quiet() {
+  Writable.call(this);
+}
+util.inherits(Quiet, Writable);
+Quiet.prototype._write = function (chunk, enc, cb) {
+  cb();
+};
+
+function afterEnd() {
+  const q = new Quiet();
+  q.end("done");
+  q.on("error", (e) => {
+    console.log("after end", e.code, e.message);
+    customEvents();
+  });
+  q.write("late");
+}
+
+// tar-stream's Extract emits 'entry' with three arguments.
+function Extract() {
+  if (!(this instanceof Extract)) return new Extract();
+  Writable.call(this);
+}
+util.inherits(Extract, Writable);
+Extract.prototype._write = function (data, enc, cb) {
+  const header = { name: data.toString(), size: data.length };
+  this.emit("entry", header, "stream-for-" + header.name, cb);
+};
+
+function customEvents() {
+  const ex = Extract();
+  const seen = [];
+  function onEntry(header, s, next) {
+    seen.push(header.name + ":" + header.size + ":" + s + ":" + (this === ex));
+    next();
+  }
+  ex.on("entry", onEntry);
+  ex.once("entry", (header) => seen.push("once " + header.name));
+  ex.prependListener("entry", () => seen.push("first"));
+  console.log("listenerCount", ex.listenerCount("entry"), ex.listenerCount("entry", onEntry));
+  ex.write("a.txt");
+  ex.removeListener("entry", onEntry);
+  ex.write("b.txt");
+  console.log(seen.join(" | "));
+  console.log("emit returns", ex.emit("nothing"), ex.emit("entry", { name: "c", size: 0 }, "", () => {}));
+  try {
+    ex.emit("error", new RangeError("nobody listens"));
+  } catch (e) {
+    console.log("unhandled error rethrown", e instanceof RangeError, e.message);
+  }
+  ex.setMaxListeners(1);
+  console.log("max listeners", ex.getMaxListeners());
+  ex.removeAllListeners("entry");
+  console.log("after removeAll", ex.listenerCount("entry"));
+  console.log("Duplex prototype", typeof Duplex.prototype.write, typeof Duplex.prototype.push);
+}

tests/corpus/6564-constructor-function-stream-late-hooks.cjs

@@ -0,0 +1,117 @@
+// Stream hooks a constructor function assigns after the base constructor
+// runs, and unpipe from the destination's 'pipe' listener.
+const { Writable, Transform, PassThrough } = require("stream");
+const util = require("util");
+
+function Sink() {
+  Writable.call(this);
+  this.chunks = [];
+  this._final = function (cb) {
+    console.log("_final", this instanceof Sink, this.chunks.join(","));
+    setImmediate(cb);
+  };
+}
+util.inherits(Sink, Writable);
+Sink.prototype._write = function (chunk, enc, cb) {
+  this.chunks.push(String(chunk));
+  cb();
+};
+
+const sink = new Sink();
+sink.on("prefinish", () => console.log("sink prefinish"));
+sink.on("finish", () => {
+  console.log("sink finish");
+  flushing();
+});
+sink.write("a");
+sink.end("b");
+
+function Summer() {
+  Transform.call(this);
+  this.total = 0;
+  this._flush = function (cb) {
+    console.log("_flush", this.total);
+    cb(null, "total " + this.total);
+  };
+}
+util.inherits(Summer, Transform);
+Summer.prototype._transform = function (chunk, enc, cb) {
+  this.total += Number(String(chunk));
+  cb(null, "seen " + chunk + "\n");
+};
+
+function flushing() {
+  const summer = new Summer();
+  const out = [];
+  summer.on("data", (chunk) => out.push(String(chunk)));
+  summer.on("end", () => {
+    console.log("summer end", JSON.stringify(out));
+    passing();
+  });
+  summer.write("2");
+  summer.write("3");
+  summer.end();
+}
+
+function Upper() {
+  PassThrough.call(this);
+  this._transform = function (chunk, enc, cb) {
+    cb(null, String(chunk).toUpperCase());
+  };
+}
+util.inherits(Upper, PassThrough);
+
+function passing() {
+  const upper = new Upper();
+  const out = [];
+  upper.on("data", (chunk) => out.push(String(chunk)));
+  upper.on("end", () => {
+    console.log("upper end", JSON.stringify(out));
+    plain();
+  });
+  upper.write("x");
+  upper.end("y");
+}
+
+// Without late hooks the defaults still apply.
+function Plain() {
+  PassThrough.call(this);
+}
+util.inherits(Plain, PassThrough);
+
+function plain() {
+  const p = new Plain();
+  const out = [];
+  p.on("data", (chunk) => out.push(String(chunk)));
+  p.on("finish", () => console.log("plain finish"));
+  p.on("end", () => {
+    console.log("plain end", JSON.stringify(out));
+    unpiping();
+  });
+  p.end("same");
+}
+
+function Collector() {
+  Writable.call(this);
+  this.chunks = [];
+}
+util.inherits(Collector, Writable);
+Collector.prototype._write = function (chunk, enc, cb) {
+  this.chunks.push(String(chunk));
+  cb();
+};
+
+function unpiping() {
+  const source = new PassThrough();
+  const collector = new Collector();
+  collector.on("pipe", function (src) {
+    console.log("pipe", typeof src.unpipe, this === collector);
+    src.unpipe(this);
+  });
+  collector.on("unpipe", (src) => {
+    console.log("unpipe", typeof src.pipe);
+    source.write("after");
+    setImmediate(() => console.log("collected", JSON.stringify(collector.chunks), source.readableFlowing));
+  });
+  source.pipe(collector);
+}

tests/harness/corpus-costs/6564-constructor-function-stream-late-hooks.cjs.json

@@ -0,0 +1 @@
+{ "plain": 2750, "san": 3200 }

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

@@ -31,6 +31,7 @@ const features: NativeLinkFeatures = {
   searchParams: false,
   qs: false,
   dynBuiltinClass: false,
+  dynBuiltinStream: false,
   parseArgs: false,
   stream: false,
   net: false,

tests/harness/self-hosting-runtime-features.test.ts

@@ -51,6 +51,7 @@ const featureCalls: readonly [string, IrLibFn][] = [
   ["searchParams", "sp.new"],
   ["qs", "qs.parse"],
   ["dynBuiltinClass", "dyn.builtinClass"],
+  ["dynBuiltinStream", "dyn.builtinStreamClass"],
   ["parseArgs", "util.parseArgs"],
   ["parseArgs", "util.isDeepStrictEqual"],
   ["parseArgs", "util.styleText"],

tests/harness/self-hosting-runtime-pack.test.ts

@@ -32,6 +32,7 @@ const base: NativeLinkFeatures = {
   searchParams: false,
   qs: false,
   dynBuiltinClass: false,
+  dynBuiltinStream: false,
   parseArgs: false,
   stream: false,
   net: false,