diff --git a/docs/content/limitations/builtin-apis/streams-and-readline.mdx b/docs/content/limitations/builtin-apis/streams-and-readline.mdx
index 73b48c88d1..a21f7a617f 100644
--- a/docs/content/limitations/builtin-apis/streams-and-readline.mdx
+++ b/docs/content/limitations/builtin-apis/streams-and-readline.mdx
@@ -1,7 +1,10 @@
 - **Streams and readline:** Stream chunks are Buffers or strings.
   `NodeJS.ReadableStream` and `NodeJS.WritableStream`, and `stream.Readable` and `stream.Writable` under `@types/node`, accept every stream: when a program passes file, zlib, or `node:stream` streams through them, child-process pipes and `process.stdout` and `process.stderr` join those streams as `Readable` and `Writable` instances.
   `readable.pipe(process.stdout)` and `readable.pipe(process.stderr)` compile and, as in Node, do not end the process stream.
-  A `pipeline` stage that arrives as an untyped value must be a native stream at run time; other values throw a not-supported error.
+  When an untyped value reaches `NodeJS.ReadableStream` or `NodeJS.WritableStream`, those two interfaces hold any value, so JavaScript stream objects (constructor-function streams, plain objects with `on` and `pipe`) pass through TypeScript functions typed with them and come back to JavaScript as the same objects.
+  Member calls on such a value use the native stream when it is one and call the object's own members otherwise; `readable.pipe()` into a JavaScript destination follows Node's pipe protocol.
+  A `pipeline` stage and a `for await` source that arrive as untyped values must be native streams at run time; other values throw a not-supported error.
+  Native streams expose no `emit` to JavaScript, so a JavaScript stream whose own `pipe()` emits on its destination (the legacy `Stream` pipe) cannot pipe into one.
   `Readable.from` accepts arrays, Sets, Map and Set iterators, generators, async generators, and class iterables and async iterables whose values are strings or Buffers, and in JavaScript any iterable or async iterable of them; a stream that mixes strings and Buffers is unsupported.
   In JavaScript, a sync iterator value with a `then` method that is not a promise, and an object whose `[Symbol.asyncIterator]` or `[Symbol.iterator]` method is a generator method, are unsupported.
   Object-mode streams of other values (including `Readable.from` over a `Map`, whose values are entry arrays) are unsupported.
diff --git a/packages/compiler/src/backend/executable-features.ts b/packages/compiler/src/backend/executable-features.ts
index 38f44e58e7..9cbf030e77 100644
--- a/packages/compiler/src/backend/executable-features.ts
+++ b/packages/compiler/src/backend/executable-features.ts
@@ -26,6 +26,7 @@ export function executableLinkFeatures(mod: IrModule): NativeLinkFeatures {
     dc: features.dc,
     dynAsync: features.dynAsync,
     ...(features.dynTier ? { dynTier: true } : {}),
+    ...(features.streamDyn ? { streamDyn: true } : {}),
     events: features.processEvents,
     emitter: features.emitter,
     symbol: features.symbol,
diff --git a/packages/compiler/src/backend/llvm/lib-shared.ts b/packages/compiler/src/backend/llvm/lib-shared.ts
index 42cf40a392..ea57ee8259 100644
--- a/packages/compiler/src/backend/llvm/lib-shared.ts
+++ b/packages/compiler/src/backend/llvm/lib-shared.ts
@@ -737,6 +737,8 @@ export const LIB_FN_SYMS: Record<string, string> = {
   "procStream.write": "scr_proc_stream_write",
   "stream.stdioWritable": "scr_stream_stdio_writable",
   "stream.unboxDyn": "scr_stream_unbox_dyn",
+  "stream.pipeDyn": "scr_stream_pipe_dyn",
+  "stream.unpipeDyn": "scr_stream_unpipe_dyn",
   "process.kill": "scr_process_kill_named",
   "process.killNum": "scr_process_kill",
   // fs.promises (scr_bytes_io.c / scr_lib.c): the promise forms settle
diff --git a/packages/compiler/src/backend/llvm/lib-stream.ts b/packages/compiler/src/backend/llvm/lib-stream.ts
index 0ded59a1a6..707fb6dbca 100644
--- a/packages/compiler/src/backend/llvm/lib-stream.ts
+++ b/packages/compiler/src/backend/llvm/lib-stream.ts
@@ -36,6 +36,24 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV
     host.emitPendingCheck();
     return out;
   }
+  if (e.fn === "stream.dynNative") {
+    // Checked-dynamic stream slots: the boxed native stream (+1) or NULL.
+    if (e.type.kind !== "union")
+      throw new InternalCompilerError(`llvm emitter bug: ${e.fn} result is not a union`);
+    const def = host.unionsById.get(e.type.unionId);
+    const streamTag = def ? def.arms.findIndex((a) => a.kind === "object") : -1;
+    const nullTag = def ? def.arms.findIndex((a) => a.kind === "nullT") : -1;
+    if (streamTag < 0 || nullTag < 0)
+      throw new InternalCompilerError(`llvm emitter bug: ${e.fn} union lacks its arms`);
+    const value = host.emitExpr(e.args[0]!);
+    const side = host.emitExpr(e.args[1]!);
+    host.declare(`declare ptr @scr_stream_dyn_native(ptr, double)`);
+    const raw = B.tmp();
+    B.line(
+      `${raw} = call ptr @scr_stream_dyn_native(ptr ${value.name}, double ${side.name}) ; +1 or NULL`,
+    );
+    return host.wrapNullable(raw, raw, def!.arms[streamTag]!, streamTag, e.type, nullTag);
+  }
   if (e.fn === "stream.onData") {
     // The callback MOVES into the stream's listener registry; the
     // adapter is per callback shape — runtime-provided for the
diff --git a/packages/compiler/src/backend/native-link-info-core.ts b/packages/compiler/src/backend/native-link-info-core.ts
index 031f99ddda..704945c95d 100644
--- a/packages/compiler/src/backend/native-link-info-core.ts
+++ b/packages/compiler/src/backend/native-link-info-core.ts
@@ -43,6 +43,9 @@ export interface NativeLinkFeatures {
   /** Dynamic-tier jsval operations: the scr_js_* units. Absent (not false)
    * for programs without them, so their link features stay unchanged. */
   dynTier?: boolean;
+  /** Checked-dynamic stream slots: the scr_stream_dyn.c unit. Absent (not
+   * false) for programs without them. */
+  streamDyn?: boolean;
   events: boolean;
   emitter: boolean;
   symbol: boolean;
diff --git a/packages/compiler/src/backend/native/contracts.ts b/packages/compiler/src/backend/native/contracts.ts
index 8abdc6c38f..c7371077f6 100644
--- a/packages/compiler/src/backend/native/contracts.ts
+++ b/packages/compiler/src/backend/native/contracts.ts
@@ -113,6 +113,9 @@ export interface CcOptions {
    * the IR): compiles the scr_js_* units (two-tier design I3). Programs
    * without them link no scr_js_ symbol. */
   dynTier?: boolean;
+  /** Checked-dynamic stream slots (stream.dynNative / stream.pipeDyn):
+   * compiles scr_stream_dyn.c. */
+  streamDyn?: boolean;
   /** The program constructs EventTarget/Event/CustomEvent values or
    * calls events.addAbortListener (the event.* libCalls), or uses the
    * native fetch AbortSignal built on the same engine: compiles
diff --git a/packages/compiler/src/backend/native/executable.ts b/packages/compiler/src/backend/native/executable.ts
index 33c126de80..5ee6b6abde 100644
--- a/packages/compiler/src/backend/native/executable.ts
+++ b/packages/compiler/src/backend/native/executable.ts
@@ -252,6 +252,9 @@ const RUNTIME_UNITS: readonly RuntimeUnit[] = [
   // stream unit.
   unit((u) => u.opts.zlib && u.opts.stream, "scr_zlib_stream.c"),
   unit((u) => u.opts.stream && u.opts.dynAsync, "scr_stream_ops.c"),
+  // Checked-dynamic stream slots test values and call JavaScript stream
+  // members through the dyn-invoke unit.
+  unit((u) => u.opts.streamDyn === true, "scr_stream_dyn.c"),
   // FileHandle streams reference both the stream unit and FileHandle.
   unit((u) => u.opts.stream && u.opts.fileHandle, "scr_fs_stream_fh.c"),
   // The readiness-poller backends (scr_platform.h): kqueue on macOS/BSD,
diff --git a/packages/compiler/src/backend/runtime-pack-core.ts b/packages/compiler/src/backend/runtime-pack-core.ts
index be94b33b9f..748de64fc4 100644
--- a/packages/compiler/src/backend/runtime-pack-core.ts
+++ b/packages/compiler/src/backend/runtime-pack-core.ts
@@ -84,6 +84,7 @@ export interface RuntimeFeatureSet extends NativeLinkFeatures {
   eventTarget: boolean;
   wasm: boolean;
   dynTier: boolean;
+  streamDyn: boolean;
   warningReplace: boolean;
   stdioExtended: boolean;
   nativeFetch: boolean;
@@ -275,6 +276,7 @@ export function effectiveRuntimeFeatures(
     eventTarget: features.eventTarget === true,
     wasm: features.wasm === true,
     dynTier: features.dynTier === true,
+    streamDyn: features.streamDyn === true,
     warningReplace: features.warningReplace === true,
     stdioExtended: features.stdioExtended === true,
     nativeFetch,
diff --git a/packages/compiler/src/executable/executable-cache.ts b/packages/compiler/src/executable/executable-cache.ts
index fd4a34ec2e..f0be7e11f9 100644
--- a/packages/compiler/src/executable/executable-cache.ts
+++ b/packages/compiler/src/executable/executable-cache.ts
@@ -66,6 +66,7 @@ export interface EarlyExecutableNativeFeatures {
   dc: boolean;
   dynAsync: boolean;
   dynTier?: boolean;
+  streamDyn?: boolean;
   events: boolean;
   emitter: boolean;
   symbol: boolean;
diff --git a/packages/compiler/src/frontend/lowering/builtins/properties.ts b/packages/compiler/src/frontend/lowering/builtins/properties.ts
index e60a14c0cb..70512fed98 100644
--- a/packages/compiler/src/frontend/lowering/builtins/properties.ts
+++ b/packages/compiler/src/frontend/lowering/builtins/properties.ts
@@ -290,6 +290,9 @@ function generalChildStdioType(
   const className = name === "stdin" ? "%Writable" : "%Readable";
   const mapped = lowerer.mapTypeOf(lowerer.typeOf(expr));
   if (!mapped) return null;
+  // Checked-dynamic stream slots: the declared stream type maps to dyn
+  // (dyn-streams.ts); the read still produces the bridged stream.
+  if (lowerer.dynNodeStreams && mapped.kind === "dyn") return { kind: "object", className };
   const arms =
     mapped.kind === "union" ? (lowerer.unions.get(mapped.unionId)?.arms ?? []) : [mapped];
   return arms.some((arm) => arm.kind === "object" && arm.className === className)
diff --git a/packages/compiler/src/frontend/lowering/dyn-streams.ts b/packages/compiler/src/frontend/lowering/dyn-streams.ts
new file mode 100644
index 0000000000..9694e3820f
--- /dev/null
+++ b/packages/compiler/src/frontend/lowering/dyn-streams.ts
@@ -0,0 +1,314 @@
+/* Checked-dynamic stream slots (TypeMapperCtx.dynNodeStreams).
+ *
+ * Under the general stream representation (general-streams.ts),
+ * NodeJS.ReadableStream and NodeJS.WritableStream map to the runtime stream
+ * classes. JavaScript code also hands such slots its own stream objects:
+ * constructor-function streams over EventEmitter or the legacy Stream, plain
+ * objects implementing on/pipe/write/end, streams of bundled npm code. A
+ * program that moves an untyped value into a stream slot therefore lowers
+ * again with checked-dynamic slots: those two interfaces map to dyn, so a
+ * slot holds either a boxed native stream (its typed capsule, which unboxes
+ * to the same stream) or the JavaScript object itself, and a value returned
+ * to JavaScript is the object that came in.
+ *
+ * A member call on such a slot dispatches per value at run time. A native
+ * stream unboxes and takes the ordinary class-method lowering (the same
+ * stream surface a typed stream has); any other value calls its own member
+ * with JavaScript's dynamic semantics (dynInvoke). Programs without such a
+ * flow never take this pass, so their IR is unchanged. */
+import type { IrExpr, IrStmt, IrType } from "../../ir/ir.js";
+import { dynUndefinedExpr, numLit, strLit, varRef } from "../../ir/build.js";
+import { DYN } from "../../ir/ir.js";
+import * as ts from "../ts7/adapter.js";
+import { locOf } from "../program.js";
+import { isDeclaredInAmbientNamespace } from "../type-mapper.js";
+import { resultIsDiscarded } from "./call-position.js";
+import { lowerObjectMethodCall } from "./lower-calls.js";
+import type { Lowerer } from "./lowerer.js";
+
+/** The side a NodeJS stream interface names: "r" for NodeJS.ReadableStream,
+ * "w" for NodeJS.WritableStream (through null/undefined and intersections),
+ * or null for any other type. */
+export function nodeStreamInterfaceSide(lowerer: Lowerer, type: ts.Type): "r" | "w" | null {
+  const parts = (t: ts.Type): ts.Type[] =>
+    (t.flags & ts.TypeFlags.Intersection) !== 0 ? ts.constituentTypes(t).flatMap(parts) : [t];
+  for (const part of parts(lowerer.checker.getNonNullableType(type))) {
+    const psym = part.getSymbol() ?? part.getAliasSymbol();
+    if (psym?.name !== "ReadableStream" && psym?.name !== "WritableStream") continue;
+    const declared = psym.declarations.some(
+      (d) =>
+        ts.isInterfaceDeclaration(d) &&
+        lowerer.isStdlibFile(d.getSourceFile()) &&
+        isDeclaredInAmbientNamespace(d, "NodeJS"),
+    );
+    if (declared) return psym.name === "ReadableStream" ? "r" : "w";
+  }
+  return null;
+}
+
+/** The runtime class a native value of the side unboxes to. */
+export function nativeStreamClass(side: "r" | "w"): IrType {
+  return { kind: "object", className: side === "r" ? "%Readable" : "%Writable" };
+}
+
+/** The native stream of a checked-dynamic value (+1, typed as the side's
+ * class) or null: the stream.dynNative libCall, which tests the value's
+ * typed capsule without throwing. */
+function dynNativeStream(lowerer: Lowerer, value: IrExpr, side: "r" | "w"): IrExpr {
+  const native = nativeStreamClass(side);
+  return {
+    kind: "libCall",
+    fn: "stream.dynNative",
+    args: [value, numLit(side === "r" ? 1 : 2, value.loc)],
+    type: { kind: "union", unionId: lowerer.unions.intern([native, { kind: "nullT" }]) },
+    loc: value.loc,
+  };
+}
+
+/** Runs `lower` with `node` lowering to `value` and its member calls taking
+ * `type` as the receiver representation. */
+function withStreamReceiver<T>(
+  lowerer: Lowerer,
+  node: ts.Expression,
+  value: IrExpr,
+  type: IrType,
+  lower: () => T,
+): T {
+  const previous = lowerer.receiverTypeOverrides.get(node);
+  lowerer.receiverTypeOverrides.set(node, type);
+  try {
+    return lowerer.withExpressionOverride(node, value, lower);
+  } finally {
+    if (previous) lowerer.receiverTypeOverrides.set(node, previous);
+    else lowerer.receiverTypeOverrides.delete(node);
+  }
+}
+
+/** `stream.member(args)` on a NodeJS.ReadableStream / NodeJS.WritableStream
+ * receiver under checked-dynamic stream slots, or null when the call is not
+ * one. */
+export function lowerDynStreamMemberCall(
+  lowerer: Lowerer,
+  call: ts.CallExpression,
+  access: ts.PropertyAccessExpression,
+): IrExpr | null {
+  if (!lowerer.dynNodeStreams) return null;
+  // The native arm re-enters the call lowering with the receiver bound.
+  if (lowerer.chainRecvByNode.has(access.expression)) return null;
+  if (access.questionDotToken || call.questionDotToken || !ts.isIdentifier(access.name))
+    return null;
+  if (call.arguments.nodes.some(ts.isSpreadElement)) return null;
+  const side = nodeStreamInterfaceSide(lowerer, lowerer.typeOf(access.expression));
+  if (side === null) return null;
+  const loc = locOf(call);
+  const receiverValue = lowerer.lowerExpr(access.expression);
+  const receiver = lowerer.declareHiddenLocal("%streamReceiver", receiverValue.type);
+  const receiverRef = varRef(receiver.id, receiver.type, locOf(access.expression));
+  const stmts: IrStmt[] = [{ kind: "varDecl", localId: receiver.id, init: receiverValue, loc }];
+  const seq = (result: IrExpr): IrExpr => ({
+    kind: "seqExpr",
+    stmts,
+    result,
+    type: result.type,
+    loc,
+  });
+  // A typed stream whose declared type is one of the interfaces (a child
+  // stdio read): the class-method path on its own representation.
+  if (receiverValue.type.kind !== "dyn") {
+    const typed =
+      receiverValue.type.kind === "object"
+        ? withStreamReceiver(lowerer, access.expression, receiverRef, receiverValue.type, () =>
+            lowerObjectMethodCall(lowerer, call, access),
+          )
+        : lowerer.withExpressionOverride(access.expression, receiverRef, () =>
+            lowerer.lowerCall(call),
+          );
+    if (typed === null) lowerer.noLowering(`${access.name.text} on this stream`, call);
+    return seq(typed);
+  }
+  const nativeType = nativeStreamClass(side);
+  const nullable = dynNativeStream(lowerer, receiverRef, side);
+  const nativeOrNull = lowerer.declareHiddenLocal("%streamNative", nullable.type);
+  stmts.push({ kind: "varDecl", localId: nativeOrNull.id, init: nullable, loc });
+  const unionId = nullable.type.kind === "union" ? nullable.type.unionId : "";
+  const tag = lowerer.armTag(unionId, nativeType);
+  const native = lowerer.declareHiddenLocal("%streamNativeValue", nativeType);
+  const nativeRef = varRef(native.id, nativeType, locOf(access.expression));
+  const nativeDecl: IrStmt = {
+    kind: "varDecl",
+    localId: native.id,
+    init: {
+      kind: "unionNarrow",
+      unionId,
+      tag,
+      value: varRef(nativeOrNull.id, nullable.type, loc),
+      type: nativeType,
+      loc,
+    },
+    loc,
+  };
+  const typed = withStreamReceiver(lowerer, access.expression, nativeRef, nativeType, () =>
+    lowerObjectMethodCall(lowerer, call, access),
+  );
+  if (typed === null)
+    lowerer.noLowering(
+      `${side === "r" ? "NodeJS.ReadableStream" : "NodeJS.WritableStream"}.${access.name.text}`,
+      call,
+    );
+  // Any other value calls its own member, with the arguments as untyped
+  // values (typed callbacks box behind checked thunks).
+  const invoked: IrExpr = {
+    kind: "dynInvoke",
+    recv: receiverRef,
+    method: access.name.text,
+    calleeName: access.getText(),
+    args: call.arguments.nodes.map((arg) => lowerer.lowerExprExpecting(arg, DYN)),
+    type: DYN,
+    loc,
+  };
+  const cond: IrExpr = {
+    kind: "unionIsTag",
+    unionId,
+    tag,
+    negated: false,
+    value: varRef(nativeOrNull.id, nullable.type, loc),
+    type: { kind: "bool" },
+    loc,
+  };
+  const resultType = resultIsDiscarded(call) ? null : lowerer.irTypeOf(call);
+  if (resultType === null || resultType.kind === "void") {
+    stmts.push({
+      kind: "if",
+      cond,
+      then: [nativeDecl, { kind: "exprStmt", expr: typed, loc }],
+      else_: [{ kind: "exprStmt", expr: invoked, loc }],
+      loc,
+    });
+    return seq(dynUndefinedExpr(loc));
+  }
+  const result = lowerer.declareHiddenLocal("%streamResult", resultType);
+  result.mutable = true;
+  stmts.push(
+    { kind: "varDecl", localId: result.id, init: null, loc },
+    {
+      kind: "if",
+      cond,
+      then: [
+        nativeDecl,
+        {
+          kind: "assign",
+          localId: result.id,
+          value: lowerer.coerceInto(call, typed, resultType),
+          loc,
+        },
+      ],
+      else_: [
+        {
+          kind: "assign",
+          localId: result.id,
+          value: lowerer.coerceInto(call, invoked, resultType),
+          loc,
+        },
+      ],
+      loc,
+    },
+  );
+  return seq(varRef(result.id, resultType, loc));
+}
+
+/** `for await (const chunk of stream)` over a checked-dynamic
+ * NodeJS.ReadableStream: a native stream iterates through the Readable
+ * lowering; any other value throws the not-supported Error (async
+ * iteration of JavaScript stream objects has no lowering). */
+export function lowerForAwaitDynStream(
+  lowerer: Lowerer,
+  stmt: ts.ForOfStatement,
+  lowerLoop: () => IrStmt | null,
+): IrStmt | null {
+  if (!lowerer.dynNodeStreams || lowerer.chainRecvByNode.has(stmt.expression)) return null;
+  if (nodeStreamInterfaceSide(lowerer, lowerer.typeOf(stmt.expression)) !== "r") return null;
+  const loc = locOf(stmt.expression);
+  const value = lowerer.lowerExpr(stmt.expression);
+  const native: IrType =
+    value.type.kind === "dyn" ? { kind: "object", className: "%Duplex" } : value.type;
+  const local = lowerer.declareHiddenLocal("%streamIterated", native);
+  const init: IrExpr =
+    value.type.kind === "dyn"
+      ? {
+          kind: "libCall",
+          fn: "stream.unboxDyn",
+          args: [
+            value,
+            strLit(
+              "for await over stream objects other than native streams is not supported yet",
+              loc,
+            ),
+          ],
+          type: native,
+          loc,
+        }
+      : value;
+  const loop = withStreamReceiver(
+    lowerer,
+    stmt.expression,
+    varRef(local.id, native, loc),
+    native,
+    lowerLoop,
+  );
+  if (loop === null) return null;
+  return {
+    kind: "block",
+    body: [{ kind: "varDecl", localId: local.id, init, loc }, loop],
+    loc: locOf(stmt),
+  };
+}
+
+/** A statement that runs `onNative` with the native stream a
+ * checked-dynamic value boxes (typed as the side's class), or `otherwise`
+ * for any other value. */
+export function dynNativeBranch(
+  lowerer: Lowerer,
+  value: IrExpr,
+  side: "r" | "w",
+  onNative: (native: IrExpr) => IrStmt,
+  otherwise: IrStmt,
+): IrStmt {
+  const loc = value.loc;
+  const nativeType = nativeStreamClass(side);
+  const nullable = dynNativeStream(lowerer, value, side);
+  const nativeOrNull = lowerer.declareHiddenLocal("%streamNative", nullable.type);
+  const unionId = nullable.type.kind === "union" ? nullable.type.unionId : "";
+  const tag = lowerer.armTag(unionId, nativeType);
+  const native = lowerer.declareHiddenLocal("%streamNativeValue", nativeType);
+  const nullableRef = varRef(nativeOrNull.id, nullable.type, loc);
+  return {
+    kind: "block",
+    body: [
+      { kind: "varDecl", localId: nativeOrNull.id, init: nullable, loc },
+      {
+        kind: "if",
+        cond: {
+          kind: "unionIsTag",
+          unionId,
+          tag,
+          negated: false,
+          value: nullableRef,
+          type: { kind: "bool" },
+          loc,
+        },
+        then: [
+          {
+            kind: "varDecl",
+            localId: native.id,
+            init: { kind: "unionNarrow", unionId, tag, value: nullableRef, type: nativeType, loc },
+            loc,
+          },
+          onNative(varRef(native.id, nativeType, loc)),
+        ],
+        else_: [otherwise],
+        loc,
+      },
+    ],
+    loc,
+  };
+}
diff --git a/packages/compiler/src/frontend/lowering/general-streams.ts b/packages/compiler/src/frontend/lowering/general-streams.ts
index 0113a99582..d31439465d 100644
--- a/packages/compiler/src/frontend/lowering/general-streams.ts
+++ b/packages/compiler/src/frontend/lowering/general-streams.ts
@@ -88,9 +88,11 @@ export function stdioIntoGeneralStream(
 ): IrExpr | null {
   if (!lowerer.generalNodeStreams || expr.type.kind !== "procStream") return null;
   const writable: IrType = { kind: "object", className: "%Writable" };
+  // Checked-dynamic stream slots (dyn-streams.ts) hold the Writable boxed.
   const accepts =
     (expected.kind === "object" && expected.className === "%Writable") ||
-    (expected.kind === "union" && lowerer.armTag(expected.unionId, writable) >= 0);
+    (expected.kind === "union" && lowerer.armTag(expected.unionId, writable) >= 0) ||
+    (expected.kind === "dyn" && lowerer.dynNodeStreams);
   if (!accepts) return null;
   return {
     kind: "libCall",
@@ -112,3 +114,23 @@ export function requestGeneralStreamSurface(lowerer: Lowerer, receiver: ts.Expre
   if (isGeneralNodeStreamInterface(type, lowerer.typeCtx))
     lowerer.requiresGeneralNodeStreams = true;
 }
+
+/** Under the general representation, an untyped value reaching a runtime
+ * stream class slot (the checked unwrap) or an untyped pipe/pipeline
+ * stage: the program lowers again with checked-dynamic stream slots
+ * (dyn-streams.ts), where NodeJS.ReadableStream / NodeJS.WritableStream
+ * hold JavaScript stream objects too. */
+export function noteUntypedStreamFlow(lowerer: Lowerer, expected: IrType | null): void {
+  if (lowerer.dynNodeStreams) return;
+  if (!lowerer.generalNodeStreams) {
+    // An untyped pipe or pipeline stage needs both: the next pass lowers
+    // with the general representation and checked-dynamic slots at once.
+    if (expected === null) {
+      lowerer.requiresGeneralNodeStreams = true;
+      lowerer.requiresDynNodeStreams = true;
+    }
+    return;
+  }
+  if (expected === null || containsPart(lowerer, expected, (p) => isGeneralStream(lowerer, p)))
+    lowerer.requiresDynNodeStreams = true;
+}
diff --git a/packages/compiler/src/frontend/lowering/lower-async-iteration.ts b/packages/compiler/src/frontend/lowering/lower-async-iteration.ts
index f095c5b9c5..0e48f20400 100644
--- a/packages/compiler/src/frontend/lowering/lower-async-iteration.ts
+++ b/packages/compiler/src/frontend/lowering/lower-async-iteration.ts
@@ -137,7 +137,9 @@ function literalBooleanOption(
 
 function nodeReadablePlan(lowerer: Lowerer, expression: ts.Expression): NodeReadablePlan | null {
   const direct =
-    lowerer.stageSourceType(expression) ?? lowerer.mapTypeOf(lowerer.typeOf(expression));
+    lowerer.receiverTypeOverrides.get(expression) ??
+    lowerer.stageSourceType(expression) ??
+    lowerer.mapTypeOf(lowerer.typeOf(expression));
   if (direct?.kind === "object") {
     const sides = streamSidesOf(lowerer, lowerer.classes.get(direct.className));
     if (sides === "r" || sides === "rw") {
diff --git a/packages/compiler/src/frontend/lowering/lower-calls.ts b/packages/compiler/src/frontend/lowering/lower-calls.ts
index 1c7c2ac435..58a94b06d5 100644
--- a/packages/compiler/src/frontend/lowering/lower-calls.ts
+++ b/packages/compiler/src/frontend/lowering/lower-calls.ts
@@ -1,5 +1,6 @@
 import { isForwardedTupleSpread, lowerForwardingApply } from "./typed-arguments.js";
 import { guardNullProto, nullProtoRecordShape } from "./null-prototype.js";
+import { lowerDynStreamMemberCall } from "./dyn-streams.js";
 import { lowerBuiltinClassCall } from "./builtins/builtin-class-dyn.js";
 import { refuseModuleMemberThis } from "./module-member-this.js";
 import { lowerUnionEquality, tagEqualityMayMissAlias } from "./strict-equality.js";
@@ -1412,6 +1413,8 @@ export function lowerCall(lowerer: Lowerer, expr: ts.CallExpression): IrExpr {
   }
   if (ts.isPropertyAccessExpression(expr.expression)) {
     const intrinsic =
+      // Checked-dynamic stream slots dispatch per value (dyn-streams.ts).
+      lowerDynStreamMemberCall(lowerer, expr, expr.expression) ??
       fenceNodeModuleMutationCall(lowerer, expr, expr.expression) ??
       // `f.apply(thisArg, args)` forwarding a fixed argument tuple (typed
       // `arguments`, a tuple rest parameter): a direct call.
@@ -9330,6 +9333,7 @@ export function lowerObjectMethodCall(
 ): IrExpr | null {
   if (lowerer.chainBlocked(access, call)) return null;
   let mappedReceiver =
+    lowerer.receiverTypeOverrides.get(access.expression) ??
     structuralReadType(lowerer, access.expression) ??
     lowerer.mapTypeOf(lowerer.typeOf(access.expression));
   // EventEmitter's static side (the class, a value of it, or the events
diff --git a/packages/compiler/src/frontend/lowering/lower-stmts.ts b/packages/compiler/src/frontend/lowering/lower-stmts.ts
index c712541ee6..4f67975630 100644
--- a/packages/compiler/src/frontend/lowering/lower-stmts.ts
+++ b/packages/compiler/src/frontend/lowering/lower-stmts.ts
@@ -2,6 +2,7 @@ import { lowerTupleAliasDecl, stripTypeWrappers } from "./typed-arguments.js";
 import { declaredMemberOf } from "./expressions/property-access.js";
 import { hostsAll } from "./expressions/hosted-narrowing.js";
 import { requestGeneralStreamSurface } from "./general-streams.js";
+import { lowerForAwaitDynStream } from "./dyn-streams.js";
 import {
   matchAllRowsReadByElements,
   storedMatchAllRowsReadByElements,
@@ -8995,6 +8996,11 @@ export function lowerForOf(lowerer: Lowerer, stmt: ts.ForOfStatement): IrStmt {
     ) {
       return lowerForAwaitStdin(lowerer, stmt);
     }
+    // A checked-dynamic NodeJS.ReadableStream iterates its native stream.
+    const dynStream = lowerForAwaitDynStream(lowerer, stmt, () =>
+      lowerForAwaitBuiltin(lowerer, stmt),
+    );
+    if (dynStream) return dynStream;
     const builtin = lowerForAwaitBuiltin(lowerer, stmt);
     if (builtin) return builtin;
     // A NodeJS.ReadableStream iterates as the general Readable.
diff --git a/packages/compiler/src/frontend/lowering/lower-stream.ts b/packages/compiler/src/frontend/lowering/lower-stream.ts
index eb919909bb..44027f65c4 100644
--- a/packages/compiler/src/frontend/lowering/lower-stream.ts
+++ b/packages/compiler/src/frontend/lowering/lower-stream.ts
@@ -49,7 +49,8 @@ import { sideEffectFreeOptionValue } from "./surfaces.js";
 import { lowerDynObjectLiteral } from "./expressions/object-literals.js";
 import { dynUndefinedExpr, varRef } from "../../ir/build.js";
 import { genResultRecord } from "../type-mapper.js";
-import { requestGeneralStreamSurface } from "./general-streams.js";
+import { noteUntypedStreamFlow, requestGeneralStreamSurface } from "./general-streams.js";
+import { dynNativeBranch } from "./dyn-streams.js";
 import { extractIteratorValue, genStep, genValue, stepDrivable } from "./lower-generators.js";
 import {
   BOOL,
@@ -2081,13 +2082,16 @@ function lowerStreamArg(
   // a boxed native stream unwraps at run time; other values (iterables,
   // functions) throw the not-supported Error there.
   if (v.type.kind === "dyn" && what === "pipeline") {
+    noteUntypedStreamFlow(lowerer, null);
     return {
       kind: "libCall",
       fn: "stream.unboxDyn",
       args: [
         v,
         strLit(
-          "pipeline stages other than native streams (iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet",
+          lowerer.dynNodeStreams
+            ? "pipeline stages other than native streams (JavaScript stream objects, iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet"
+            : "pipeline stages other than native streams (iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet",
           locOf(node),
         ),
       ],
@@ -3403,6 +3407,63 @@ function pipeIntoStdio(
   };
 }
 
+/** `readable.unpipe(untyped)`: a native destination (boxed) unpipes as a
+ * typed one does; any other value unpipes the JavaScript destination a
+ * pipe into it registered (scr_stream_dyn.c), or does nothing when there
+ * is none, as Node's unpipe does for a destination it is not piped into.
+ * The call answers the source. */
+function unpipeDyn(lowerer: Lowerer, receiver: IrExpr, dst: IrExpr, loc: SrcLoc): IrExpr {
+  noteUntypedStreamFlow(lowerer, null);
+  const source = lowerer.declareHiddenLocal("%unpipeSource", receiver.type);
+  const destination = lowerer.declareHiddenLocal("%unpipeDestination", dst.type);
+  const sourceRef = varRef(source.id, receiver.type, loc);
+  const destinationRef = varRef(destination.id, dst.type, loc);
+  const stmts: IrStmt[] = [
+    { kind: "varDecl", localId: source.id, init: receiver, loc },
+    { kind: "varDecl", localId: destination.id, init: dst, loc },
+  ];
+  const unpipeNative = (native: IrExpr): IrStmt => ({
+    kind: "exprStmt",
+    expr: {
+      kind: "libCall",
+      fn: "readable.unpipe",
+      args: [sourceRef, native],
+      type: receiver.type,
+      loc,
+    },
+    loc,
+  });
+  if (lowerer.dynNodeStreams) {
+    stmts.push(
+      dynNativeBranch(lowerer, destinationRef, "w", unpipeNative, {
+        kind: "exprStmt",
+        expr: {
+          kind: "libCall",
+          fn: "stream.unpipeDyn",
+          args: [sourceRef, destinationRef],
+          type: VOID,
+          loc,
+        },
+        loc,
+      }),
+    );
+  } else {
+    stmts.push(
+      unpipeNative({
+        kind: "libCall",
+        fn: "stream.unboxDyn",
+        args: [
+          destinationRef,
+          strLit("unpipe destinations other than native streams are not supported yet", loc),
+        ],
+        type: { kind: "object", className: "%Duplex" },
+        loc,
+      }),
+    );
+  }
+  return { kind: "seqExpr", stmts, result: sourceRef, type: receiver.type, loc };
+}
+
 /** The pipe() libCall for a destination of type `dst`. Under the general
  * stream representation a Writable slot can hold the converted
  * process.stdout / process.stderr, which a pipe never ends (as in Node), so
@@ -3423,6 +3484,7 @@ function pipeIntoDyn(
   target: IrExpr,
 ): IrExpr {
   const loc = locOf(call);
+  noteUntypedStreamFlow(lowerer, null);
   const receiverValue = lowerer.lowerExpr(access.expression);
   const source = lowerer.declareHiddenLocal("%pipeSource", receiverValue.type);
   const destination = lowerer.declareHiddenLocal("%pipeDestination", target.type);
@@ -3432,6 +3494,59 @@ function pipeIntoDyn(
   ];
   const end = pipeEndOption(lowerer, call.arguments.nodes[1], loc);
   const duplex: IrType = { kind: "object", className: "%Duplex" };
+  // Checked-dynamic stream slots (dyn-streams.ts): a JavaScript stream
+  // object is a destination too, piped with Node's listener protocol.
+  if (lowerer.dynNodeStreams) {
+    const sourceRef: IrExpr = { kind: "varRef", localId: source.id, type: receiverValue.type, loc };
+    const destinationRef: IrExpr = {
+      kind: "varRef",
+      localId: destination.id,
+      type: target.type,
+      loc,
+    };
+    const ending = lowerer.declareHiddenLocal("%pipeEnd", BOOL);
+    stmts.push({ kind: "varDecl", localId: ending.id, init: end, loc });
+    const endRef: IrExpr = { kind: "varRef", localId: ending.id, type: BOOL, loc };
+    stmts.push(
+      dynNativeBranch(
+        lowerer,
+        destinationRef,
+        "w",
+        (native) => ({
+          kind: "exprStmt",
+          expr: {
+            kind: "libCall",
+            // Checked-dynamic slots come with the general representation,
+            // where a boxed native destination can be the converted
+            // process.stdout (generalPipe).
+            fn: "readable.pipeGeneral",
+            args: [sourceRef, native, endRef],
+            type: native.type,
+            loc,
+          },
+          loc,
+        }),
+        {
+          kind: "exprStmt",
+          expr: {
+            kind: "libCall",
+            fn: "stream.pipeDyn",
+            args: [sourceRef, lowerer.coerceToExpected(sourceRef, DYN), destinationRef, endRef],
+            type: VOID,
+            loc,
+          },
+          loc,
+        },
+      ),
+    );
+    return {
+      kind: "seqExpr",
+      stmts,
+      result: destinationRef,
+      type: target.type,
+      loc,
+    };
+  }
   stmts.push({
     kind: "exprStmt",
     expr: {
@@ -3746,6 +3861,7 @@ export function lowerStreamMethodCall(
     const receiver = lowerer.lowerExpr(access.expression);
     if (args[0] !== undefined) {
       const dst = lowerer.lowerExpr(args[0]);
+      if (dst.type.kind === "dyn") return unpipeDyn(lowerer, receiver, dst, loc);
       const dstInfo =
         dst.type.kind === "object" ? lowerer.classes.get(dst.type.className) : undefined;
       const dstSides = streamSidesOf(lowerer, dstInfo);
@@ -3760,6 +3876,44 @@ export function lowerStreamMethodCall(
         loc,
       };
     }
+    // Checked-dynamic slots: unpipe() also unpipes the JavaScript
+    // destinations stream.pipeDyn registered (stream.unpipeDyn with an
+    // undefined destination), as Node's unpipe() empties every pipe.
+    if (lowerer.dynNodeStreams) {
+      const source = lowerer.declareHiddenLocal("%unpipeSource", receiver.type);
+      const sourceRef = varRef(source.id, receiver.type, loc);
+      return {
+        kind: "seqExpr",
+        stmts: [
+          { kind: "varDecl", localId: source.id, init: receiver, loc },
+          {
+            kind: "exprStmt",
+            expr: {
+              kind: "libCall",
+              fn: "readable.unpipe",
+              args: [sourceRef],
+              type: receiver.type,
+              loc,
+            },
+            loc,
+          },
+          {
+            kind: "exprStmt",
+            expr: {
+              kind: "libCall",
+              fn: "stream.unpipeDyn",
+              args: [sourceRef, dynUndefinedExpr(loc)],
+              type: VOID,
+              loc,
+            },
+            loc,
+          },
+        ],
+        result: sourceRef,
+        type: receiver.type,
+        loc,
+      };
+    }
     return { kind: "libCall", fn: "readable.unpipe", args: [receiver], type: receiver.type, loc };
   }
 
diff --git a/packages/compiler/src/frontend/lowering/lowerer.ts b/packages/compiler/src/frontend/lowering/lowerer.ts
index 064756d333..47b5f83398 100644
--- a/packages/compiler/src/frontend/lowering/lowerer.ts
+++ b/packages/compiler/src/frontend/lowering/lowerer.ts
@@ -1,5 +1,9 @@
 import { onlyForwarded } from "./typed-arguments.js";
-import { needsGeneralNodeStreams, stdioIntoGeneralStream } from "./general-streams.js";
+import {
+  needsGeneralNodeStreams,
+  noteUntypedStreamFlow,
+  stdioIntoGeneralStream,
+} from "./general-streams.js";
 import { extendAssertedRecordMembers } from "./assertion-members.js";
 import { jsDynamicDeclarations } from "./js-dynamic-objects.js";
 import { builtinClassValue, builtinMemberClassKind } from "./builtins/builtin-class-dyn.js";
@@ -890,6 +894,11 @@ export interface LowererMode {
    * generalNodeStreams). lowerToIr turns it on only after a pass without
    * it saw a stream flow that needs it (requiresGeneralNodeStreams). */
   generalNodeStreams?: boolean;
+  /** With generalNodeStreams, the checked-dynamic stream slots
+   * (TypeMapperCtx.dynNodeStreams). lowerToIr turns it on only after a
+   * pass saw an untyped value flow into a stream slot
+   * (requiresDynNodeStreams). */
+  dynNodeStreams?: boolean;
   /** LoadResult.inlineRequireCycleCalls. */
   inlineRequireCycleCalls?: ReadonlySet<ts.Node> | undefined;
   /** LowerOptions.libraryCallbacks (see there). */
@@ -1055,7 +1064,11 @@ export function lowerToIr(
   // a general stream into a stdio-represented stream slot lowers again
   // with it. Every other program keeps the single pass above unchanged.
   const generalNodeStreams = reachableEmit.requiresGeneralNodeStreams;
-  if (generalNodeStreams) {
+  // Checked-dynamic stream slots (dyn-streams.ts) are a further per-program
+  // decision under the general representation: only a program that moves an
+  // untyped value into a stream slot lowers again with them.
+  let dynNodeStreams = reachableEmit.requiresDynNodeStreams;
+  const streamPass = (dyn: boolean): Lowerer => {
     const general = new Lowerer(program, entry, moduleOrder, {
       frontendServices: options.frontendServices,
       projectCheckJs: options.projectCheckJs ?? false,
@@ -1071,6 +1084,7 @@ export function lowerToIr(
       tierMode,
       ffiBindingSymbols: ffiValidation.symbolsByName,
       generalNodeStreams: true,
+      ...(dyn ? { dynNodeStreams: true } : {}),
     });
     for (const d of extraModuleCycleDiags) general.pushDiag(d);
     for (const d of ffiValidation.diagnostics) general.pushDiag(d);
@@ -1078,7 +1092,15 @@ export function lowerToIr(
     reachable = generalEmitted.reachable;
     result = generalEmitted.result;
     resultLowerer = general;
-    timing("general-streams-relower", { reachable: reachable.size });
+    timing(dyn ? "dyn-streams-relower" : "general-streams-relower", { reachable: reachable.size });
+    return general;
+  };
+  if (generalNodeStreams) {
+    const general = streamPass(dynNodeStreams);
+    if (!dynNodeStreams && general.requiresDynNodeStreams) {
+      dynNodeStreams = true;
+      streamPass(true);
+    }
   }
   // A generic class-rest support decision can depend on record metadata
   // whose historical owner is discovered only while retained bodies lower.
@@ -1092,6 +1114,7 @@ export function lowerToIr(
       frontendServices: options.frontendServices,
       projectCheckJs: options.projectCheckJs ?? false,
       ...(generalNodeStreams ? { generalNodeStreams: true } : {}),
+      ...(dynNodeStreams ? { dynNodeStreams: true } : {}),
       reachable,
       targetPlatform,
       startupCrash,
@@ -1182,11 +1205,12 @@ export function lowerToIr(
     timing("wasm-discovery", { modules: wasmModules.length });
   }
   if (options.coverage !== true) return result;
-  const remainderPass = (general: boolean): Lowerer =>
+  const remainderPass = (general: boolean, dyn = dynNodeStreams): Lowerer =>
     new Lowerer(program, entry, moduleOrder, {
       frontendServices: options.frontendServices,
       projectCheckJs: options.projectCheckJs ?? false,
       ...(general ? { generalNodeStreams: true } : {}),
+      ...(general && dyn ? { dynNodeStreams: true } : {}),
       reachable,
       remainder: true,
       alreadyFlushed: resultLowerer.flushedSymbols,
@@ -1207,7 +1231,11 @@ export function lowerToIr(
   // stream flow there that needs the general representation lowers the
   // remainder again with it, as reachable emit would.
   if (!generalNodeStreams && remainder.requiresGeneralNodeStreams) {
-    remainder = remainderPass(true);
+    remainder = remainderPass(true, remainder.requiresDynNodeStreams);
+    rem = remainder.run();
+  }
+  if (!dynNodeStreams && remainder.generalNodeStreams && remainder.requiresDynNodeStreams) {
+    remainder = remainderPass(true, true);
     rem = remainder.run();
   }
   timing("coverage-remainder");
@@ -2594,6 +2622,15 @@ export class Lowerer {
    * a stdio handle reached an 'unknown' slot. lowerToIr then lowers the
    * program again under the general stream representation. */
   requiresGeneralNodeStreams = false;
+  /** An untyped value reached a stream slot (a checked unwrap into a
+   * runtime stream class, or an untyped pipe/pipeline stage). lowerToIr
+   * then lowers the program again with checked-dynamic stream slots
+   * (dyn-streams.ts). */
+  requiresDynNodeStreams = false;
+  /** Receiver representations forced by the per-value stream dispatch
+   * (dyn-streams.ts): the native arm lowers a member call on the unboxed
+   * stream through the ordinary class-method path. */
+  readonly receiverTypeOverrides = new Map<ts.Node, IrType>();
   /** Helpers that snapshot shape declaration order into their bodies.
    * Reachability lowers inits before the declarations they discover, so
    * these rebuild after the worklist restores historical shape metadata. */
@@ -3053,6 +3090,7 @@ export class Lowerer {
    * committed refusals in "auto" and keeps them refusals in "static-only". */
   readonly tierMode: TierMode;
   readonly generalNodeStreams: boolean;
+  readonly dynNodeStreams: boolean;
   /** Functions finishModule placed in the dynamic tier (ir/tiers.ts). */
   dynamicFunctions: DynamicFunction[] = [];
   /** Their IR, for coverage's dynamic statement count. */
@@ -3182,6 +3220,7 @@ export class Lowerer {
     this.targetPlatform = mode.targetPlatform ?? process.platform;
     this.tierMode = mode.tierMode ?? "auto";
     this.generalNodeStreams = mode.generalNodeStreams ?? false;
+    this.dynNodeStreams = this.generalNodeStreams && (mode.dynNodeStreams ?? false);
     this.workerTargets = mode.workerTargets ?? [];
     this.workerTargets.forEach((sf, id) => {
       this.workerTargetIdByPath.set(tsgoPath(resolve(sf.fileName)), id);
@@ -3279,6 +3318,7 @@ export class Lowerer {
         this.externalTypeSpecifiersByFile.has(tsgoPath(resolve(sf.fileName))),
       typeMemo: this.typeMemo,
       ...(this.generalNodeStreams ? { generalNodeStreams: true } : {}),
+      ...(this.dynNodeStreams ? { dynNodeStreams: true } : {}),
       canMemoizeType: () =>
         this.typeParamBindings === null &&
         this.typeParamTsBindings === null &&
@@ -8563,8 +8603,12 @@ export class Lowerer {
    * validator ICEs on them as the usual backstop. */
   requireExactShape(node: ts.Node, actual: IrType, expected: IrType): void {
     if (typeEquals(actual, expected)) return;
-    if (!this.generalNodeStreams && needsGeneralNodeStreams(this, actual, expected))
+    if (!this.generalNodeStreams && needsGeneralNodeStreams(this, actual, expected)) {
       this.requiresGeneralNodeStreams = true;
+      // An untyped value into a stream slot also needs the checked-dynamic
+      // slots (dyn-streams.ts).
+      if (actual.kind === "dyn") this.requiresDynNodeStreams = true;
+    }
     // dyn mismatches first. A dyn ('unknown') value flowing into a typed
     // slot needs a CHECKED cast — the hint points at `as <type>`; tsc
     // usually rejects this before we do, so the fence is mostly defensive.
@@ -8956,6 +9000,7 @@ export class Lowerer {
           (id) => this.unions.get(id),
         );
       const handleOk = DYN_HANDLE_KINDS.has(expected.kind);
+      noteUntypedStreamFlow(this, expected);
       if (
         canDynCheckTo(
           expected,
diff --git a/packages/compiler/src/frontend/type-mapper.ts b/packages/compiler/src/frontend/type-mapper.ts
index 0bb31ad03c..096000b4db 100644
--- a/packages/compiler/src/frontend/type-mapper.ts
+++ b/packages/compiler/src/frontend/type-mapper.ts
@@ -1178,6 +1178,12 @@ export interface TypeMapperCtx {
    * from every stream producer share their slots. Off (the default), the
    * stdio handles keep those spellings. */
   generalNodeStreams?: boolean;
+  /** With generalNodeStreams, the program's checked-dynamic stream slots
+   * (lowerToIr decides it per program, dyn-streams.ts): NodeJS.ReadableStream
+   * and NodeJS.WritableStream map to the checked-dynamic representation, so
+   * a slot holds a native stream (a boxed runtime stream, identity kept) or
+   * a JavaScript stream object. Members dispatch per value at run time. */
+  dynNodeStreams?: boolean;
   /** Successful context-free mappings, keyed by the
    * TypeScript server's stable type id. A generated facade can reference the
    * same large record/function type thousands of times; remapping its full
@@ -3224,6 +3230,27 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
   // Under the program's general stream representation (generalNodeStreams)
   // NodeJS.WritableStream is the runtime Writable instead: the process
   // streams convert into it where they flow (process.stdioWritable).
+  // Checked-dynamic stream slots: an intersection over a NodeJS stream
+  // interface (`NodeJS.ReadableStream & { destroy?(): void }`) holds the
+  // same values as the interface.
+  if (
+    ctx.dynNodeStreams === true &&
+    (type.flags & ts.TypeFlags.Intersection) !== 0 &&
+    ts.constituentTypes(type).some((part) => {
+      const sym = part.getSymbol();
+      return (
+        (sym?.name === "ReadableStream" || sym?.name === "WritableStream") &&
+        sym.declarations.some(
+          (d) =>
+            ts.isInterfaceDeclaration(d) &&
+            ctx.isStdlibFile(d.getSourceFile()) &&
+            isDeclaredInAmbientNamespace(d, "NodeJS"),
+        )
+      );
+    })
+  ) {
+    return DYN;
+  }
   const nodeWritableInterface =
     psym?.name === "WritableStream" &&
     psym.declarations.some(
@@ -3233,7 +3260,7 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
         isDeclaredInAmbientNamespace(d, "NodeJS"),
     );
   if (nodeWritableInterface && ctx.generalNodeStreams === true) {
-    return { kind: "object", className: "%Writable" };
+    return ctx.dynNodeStreams === true ? DYN : { kind: "object", className: "%Writable" };
   }
   if (
     (psym?.name === "WriteStream" &&
@@ -3359,8 +3386,13 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
       ))
   ) {
     // The general representation: the runtime Readable, which child
-    // stdio reads produce through the child stream bridge.
-    if (ctx.generalNodeStreams === true) return { kind: "object", className: "%Readable" };
+    // stdio reads produce through the child stream bridge. With
+    // checked-dynamic stream slots, NodeJS.ReadableStream holds any
+    // stream value (dyn-streams.ts); stream.Readable keeps the class.
+    if (ctx.generalNodeStreams === true) {
+      if (ctx.dynNodeStreams === true && psym.name === "ReadableStream") return DYN;
+      return { kind: "object", className: "%Readable" };
+    }
     return { kind: "childStream" };
   }
   // dgram.Socket: the same ambiguous "Socket" name disambiguated by its
@@ -4435,7 +4467,12 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
         (ctx.generalNodeStreams === true &&
           arm.kind === "object" &&
           ((expected === "childStream" && arm.className === "%Readable") ||
-            (expected === "childWriter" && arm.className === "%Writable")));
+            (expected === "childWriter" && arm.className === "%Writable"))) ||
+        // Checked-dynamic stream slots: `NodeJS.ReadableStream | null`
+        // collapses to dyn (the reads still produce the bridged stream).
+        (ctx.dynNodeStreams === true &&
+          arm.kind === "dyn" &&
+          (expected === "childStream" || expected === "childWriter"));
       return (
         arms !== undefined &&
         arms.some(isExpected) &&
diff --git a/packages/compiler/src/index.ts b/packages/compiler/src/index.ts
index 7b1ef71093..9c02cdd8b1 100644
--- a/packages/compiler/src/index.ts
+++ b/packages/compiler/src/index.ts
@@ -545,6 +545,7 @@ async function compileExecutableNative(
     dc: features.dc,
     dynAsync: features.dynAsync,
     ...(features.dynTier ? { dynTier: true } : {}),
+    ...(features.streamDyn ? { streamDyn: true } : {}),
     events: features.events,
     emitter: features.emitter,
     symbol: features.symbol,
diff --git a/packages/compiler/src/ir/builtin-effects.ts b/packages/compiler/src/ir/builtin-effects.ts
index ee65a32a52..c80235ae13 100644
--- a/packages/compiler/src/ir/builtin-effects.ts
+++ b/packages/compiler/src/ir/builtin-effects.ts
@@ -77,6 +77,8 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "procStream.write",
   "child.stdioStream",
   "stream.unboxDyn",
+  "stream.pipeDyn",
+  "stream.unpipeDyn",
   "weakMap.new",
   "weakSet.new",
   "dyn.fromEntries",
diff --git a/packages/compiler/src/ir/builtin-signatures.ts b/packages/compiler/src/ir/builtin-signatures.ts
index 9a66b0d68b..c8ebf2eb92 100644
--- a/packages/compiler/src/ir/builtin-signatures.ts
+++ b/packages/compiler/src/ir/builtin-signatures.ts
@@ -2666,6 +2666,20 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
     argTypes: [DYN, STRING],
     result: { kind: "object", className: "%Duplex" },
   },
+  /** Checked-dynamic stream slots (dyn-streams.ts): the native stream a
+   * checked-dynamic value boxes (+1) — a runtime stream with the readable
+   * (1) or writable (2) side — or null for any other value. Never throws;
+   * the result union is `%Readable | null` or `%Writable | null`. */
+  "stream.dynNative": { argTypes: [DYN, F64], result: VOID },
+  /** readable.pipe(destination, end) into a checked-dynamic destination
+   * that is not a native stream: [source, the source boxed, destination,
+   * end]. Node's pipe listener protocol over the destination's members
+   * (scr_stream_dyn.c); throws what those members throw. */
+  "stream.pipeDyn": { argTypes: [null, DYN, DYN, BOOL], result: VOID },
+  /** readable.unpipe(destination) for a checked-dynamic destination that
+   * is not a native stream: [source, destination]. Unpipes the pipe
+   * stream.pipeDyn registered into it, if any (scr_stream_dyn.c). */
+  "stream.unpipeDyn": { argTypes: [null, DYN], result: VOID },
   "stream.stdioWritable": {
     argTypes: [PROCSTREAM_T],
     result: { kind: "object", className: "%Writable" },
diff --git a/packages/compiler/src/ir/runtime-features.ts b/packages/compiler/src/ir/runtime-features.ts
index b52afafcb4..0d680d7afe 100644
--- a/packages/compiler/src/ir/runtime-features.ts
+++ b/packages/compiler/src/ir/runtime-features.ts
@@ -168,6 +168,9 @@ const STREAM_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   "stream.destroyErr",
   "stream.errored",
   "stream.prop",
+  "stream.dynNative",
+  "stream.pipeDyn",
+  "stream.unpipeDyn",
   "stream.stdioWritable",
   "stream.unboxDyn",
   "stream.workerStdio",
@@ -204,6 +207,9 @@ const READABLE_DYN_ASYNC_PREFIXES: readonly string[] = ["readable.fromGen", "rea
 
 /** Entry points that link the dyn-invoke unit. */
 const DYN_INVOKE_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
+  "stream.dynNative",
+  "stream.pipeDyn",
+  "stream.unpipeDyn",
   "dyn.arrayProtoCall",
   "dyn.arrayPrototype",
   "dyn.builtinFunction",
@@ -553,6 +559,9 @@ const RUNTIME_FEATURE_RULES = {
   /** The program has dynamic-tier jsval operations (js.* builtins, the
    * jsval pass): links the scr_js_* units (two-tier design I3). */
   dynTier: rule({ prefixes: ["js."] }),
+  /** Checked-dynamic stream slots test values and pipe into JavaScript
+   * stream objects (scr_stream_dyn.c). */
+  streamDyn: rule({ fns: ["stream.dynNative", "stream.pipeDyn", "stream.unpipeDyn"] }),
   /** The stream unit emits through the emitter unit, also when the
    * program constructs no stream (a pipeline of functions). */
   emitter: rule({
@@ -731,6 +740,7 @@ function noRuntimeFeatures(): RuntimeFeatures {
     searchParams: false,
     stdioExtended: false,
     stream: false,
+    streamDyn: false,
     symbol: false,
     tls: false,
     tlsCa: false,
diff --git a/packages/compiler/src/ir/validate.ts b/packages/compiler/src/ir/validate.ts
index d6a4a09b4b..ee4ec7b3c8 100644
--- a/packages/compiler/src/ir/validate.ts
+++ b/packages/compiler/src/ir/validate.ts
@@ -5228,6 +5228,21 @@ function validateFunction(
       }
       return true;
     }
+    if (e.fn === "stream.dynNative") {
+      const def = e.type.kind === "union" ? unions.get(e.type.unionId) : undefined;
+      const side = e.args[1];
+      const className =
+        side?.kind === "numLit" ? (side.value === 1 ? "%Readable" : "%Writable") : null;
+      const ok =
+        def &&
+        className !== null &&
+        e.args[0]?.type.kind === "dyn" &&
+        def.arms.length === 2 &&
+        def.arms.some((a) => a.kind === "object" && a.className === className) &&
+        def.arms.some((a) => a.kind === "nullT");
+      if (!ok) err(`libCall ${e.fn} must return its side's stream class or null`, e.loc);
+      return true;
+    }
     if (e.fn === "child.stdioStream") {
       const def = e.type.kind === "union" ? unions.get(e.type.unionId) : undefined;
       const slot = e.args[1];
diff --git a/packages/compiler/src/native/sanitizer.test.ts b/packages/compiler/src/native/sanitizer.test.ts
index ea9690ce7f..35dcbddca2 100644
--- a/packages/compiler/src/native/sanitizer.test.ts
+++ b/packages/compiler/src/native/sanitizer.test.ts
@@ -77,6 +77,7 @@ test.skipIf(process.platform !== "linux" && process.platform !== "darwin")(
               eventTarget: false,
               wasm: false,
               dynTier: false,
+              streamDyn: false,
               warningReplace: false,
               stdioExtended: false,
               regex: true,
diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts.json
new file mode 100644
index 0000000000..85cf02e2e7
--- /dev/null
+++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts.json
@@ -0,0 +1,7 @@
+{
+ "order": [
+  "<repo>/tests/corpus/5819-node-stream-dyn-direct-pipe/sources.js",
+  "<repo>/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts"
+ ],
+ "diags": []
+}
diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6480-node-stream-dyn-readable/main.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6480-node-stream-dyn-readable/main.ts.json
new file mode 100644
index 0000000000..b3698b91c7
--- /dev/null
+++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6480-node-stream-dyn-readable/main.ts.json
@@ -0,0 +1,7 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6480-node-stream-dyn-readable/sources.js",
+  "<repo>/tests/corpus/6480-node-stream-dyn-readable/main.ts"
+ ],
+ "diags": []
+}
diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6481-node-stream-dyn-writable/main.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6481-node-stream-dyn-writable/main.ts.json
new file mode 100644
index 0000000000..45b8ebb8af
--- /dev/null
+++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6481-node-stream-dyn-writable/main.ts.json
@@ -0,0 +1,7 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6481-node-stream-dyn-writable/sources.js",
+  "<repo>/tests/corpus/6481-node-stream-dyn-writable/main.ts"
+ ],
+ "diags": []
+}
diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6482-node-stream-dyn-helpers/main.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6482-node-stream-dyn-helpers/main.ts.json
new file mode 100644
index 0000000000..b45958c333
--- /dev/null
+++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6482-node-stream-dyn-helpers/main.ts.json
@@ -0,0 +1,7 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6482-node-stream-dyn-helpers/sources.js",
+  "<repo>/tests/corpus/6482-node-stream-dyn-helpers/main.ts"
+ ],
+ "diags": []
+}
diff --git a/packages/runtime-pack-common/runtime-pack-matrix.mjs b/packages/runtime-pack-common/runtime-pack-matrix.mjs
index 28ccf0f593..a47ebc830b 100644
--- a/packages/runtime-pack-common/runtime-pack-matrix.mjs
+++ b/packages/runtime-pack-common/runtime-pack-matrix.mjs
@@ -99,6 +99,8 @@ export const OPTIONAL_RUNTIME_SOURCES = [
   ["scr_zlib_stream.c", all("zlib", "stream")],
   // The Readable operators await through the dyn-async unit.
   ["scr_stream_ops.c", all("stream", "dynAsync")],
+  // Checked-dynamic stream slots call JavaScript members through the dyn-invoke unit.
+  ["scr_stream_dyn.c", "streamDyn"],
   // fs.ReadStream / fs.WriteStream reference only the stream unit and base units.
   ["scr_fs_stream.c", "stream"],
   // FileHandle streams reference both the stream unit and FileHandle.
diff --git a/packages/runtime/src/scr_stream_dyn.c b/packages/runtime/src/scr_stream_dyn.c
new file mode 100644
index 0000000000..0859c4e07d
--- /dev/null
+++ b/packages/runtime/src/scr_stream_dyn.c
@@ -0,0 +1,389 @@
+/* Checked-dynamic stream slots (the compiler's dyn-streams.ts).
+ *
+ * A program whose NodeJS.ReadableStream / NodeJS.WritableStream slots also
+ * hold JavaScript stream objects compiles them as checked-dynamic values: a
+ * native stream rides its typed capsule, any other value is the JavaScript
+ * object itself. Member calls test the value here and take the native
+ * stream path or the value's own members. Linked only by such programs
+ * (the stream and dyn-invoke units together). */
+#include "scr_runtime.h"
+
+#include <stdlib.h>
+#include <string.h>
+
+static bool scr_stream_dyn_within(const ScrVt *vt, const ScrVt *cls) {
+  return vt == cls || (cls->pre <= vt->pre && vt->pre <= cls->post);
+}
+
+/* The native stream a checked-dynamic value boxes (+1), when it has the
+ * requested side (1 readable, 2 writable; Duplex, Transform and PassThrough
+ * have both), or NULL for any other value. Never throws. */
+ScrStream *scr_stream_dyn_native(const ScrDyn *value, double side) {
+  if (value->kind != SCR_DYN_TYPED_REF || value->v.typed_ref.type_key_len <= 7 ||
+      memcmp(value->v.typed_ref.type_key, "object:", 7) != 0)
+    return NULL;
+  ScrStream *stream = scr_dyn_typed_ref_unbox(value);
+  const ScrVt *vt = stream->vt;
+  bool duplex = scr_stream_dyn_within(vt, &scr_duplex_vt) ||
+                scr_stream_dyn_within(vt, &scr_transform_vt) ||
+                scr_stream_dyn_within(vt, &scr_passthrough_vt);
+  bool ok = duplex || (side == 1 ? scr_stream_dyn_within(vt, &scr_readable_vt)
+                                 : scr_stream_dyn_within(vt, &scr_writable_vt));
+  if (ok) return stream;
+  value->v.typed_ref.release(stream);
+  return NULL;
+}
+
+/* ── readable.pipe(destination) into a JavaScript stream object ─────────
+ * Node's Readable.prototype.pipe (lib/internal/streams/readable.js) over a
+ * native source and a destination that is not a native stream: the same
+ * listeners in the same order, with the destination's members called
+ * through the dyn-invoke unit. The context holds the source, its boxed
+ * value (what 'pipe' and 'unpipe' pass), the destination and the boxed
+ * listeners until cleanup; the live list breaks the listener cycles at
+ * exit without calling JavaScript. */
+
+enum {
+  SCR_DYN_PIPE_DATA,
+  SCR_DYN_PIPE_END,
+  SCR_DYN_PIPE_UNPIPE,
+  SCR_DYN_PIPE_DRAIN,
+  SCR_DYN_PIPE_ERROR,
+  SCR_DYN_PIPE_CLOSE,
+  SCR_DYN_PIPE_FINISH,
+  SCR_DYN_PIPE_ON_UNPIPE,
+  SCR_DYN_PIPE_COUNT
+};
+
+typedef struct ScrDynPipe {
+  size_t rc;
+  ScrStream *src;    /* +1 until dropped */
+  ScrDyn *src_box;   /* +1 until dropped */
+  ScrDyn *dest;      /* +1 until dropped */
+  ScrDyn *l[SCR_DYN_PIPE_COUNT]; /* +1 boxed listeners, or NULL */
+  bool end, cleaned, awaiting_drain, unpiped;
+  struct ScrDynPipe *live_prev, *live_next;
+} ScrDynPipe;
+
+static SCR_TL ScrDynPipe *scr_dyn_pipe_live = NULL;
+
+/* A string value for an event name (+1). */
+static ScrDyn *scr_dyn_pipe_str(const char *text, size_t len) {
+  ScrStr *s = scr_str_new(text, len);
+  ScrDyn *value = scr_dyn_new_str(s);
+  scr_str_release(s);
+  return value;
+}
+static SCR_TL bool scr_dyn_pipe_atexit_set = false;
+
+static void *scr_dyn_pipe_retain_v(void *p) {
+  ((ScrDynPipe *)p)->rc++;
+  return p;
+}
+
+static void scr_dyn_pipe_drop(ScrDynPipe *p) {
+  if (p->live_prev) p->live_prev->live_next = p->live_next;
+  else if (scr_dyn_pipe_live == p) scr_dyn_pipe_live = p->live_next;
+  if (p->live_next) p->live_next->live_prev = p->live_prev;
+  p->live_prev = p->live_next = NULL;
+  for (size_t i = 0; i < SCR_DYN_PIPE_COUNT; i++) {
+    ScrDyn *l = p->l[i];
+    p->l[i] = NULL;
+    if (l) scr_dyn_release(l);
+  }
+  if (p->src) {
+    ScrStream *src = p->src;
+    p->src = NULL;
+    scr_stream_release(src);
+  }
+  if (p->src_box) {
+    ScrDyn *box = p->src_box;
+    p->src_box = NULL;
+    scr_dyn_release(box);
+  }
+  if (p->dest) {
+    ScrDyn *dest = p->dest;
+    p->dest = NULL;
+    scr_dyn_release(dest);
+  }
+}
+
+static void scr_dyn_pipe_release_v(void *v) {
+  ScrDynPipe *p = v;
+  if (--p->rc > 0) return;
+  scr_dyn_pipe_drop(p);
+  free(p);
+}
+
+static void scr_dyn_pipe_atexit(void) {
+  while (scr_dyn_pipe_live) {
+    ScrDynPipe *p = scr_dyn_pipe_retain_v(scr_dyn_pipe_live);
+    scr_dyn_pipe_drop(p);
+    scr_dyn_pipe_release_v(p);
+  }
+}
+
+/* destination.method(args...), the result discarded; false on a throw. */
+static bool scr_dyn_pipe_call(ScrDynPipe *p, const char *method, ScrDyn *const *args, size_t argc) {
+  ScrDyn *result = scr_dyn_invoke(p->dest, method, args, argc, method);
+  if (result) scr_dyn_release(result);
+  return !scr_exc_pending();
+}
+
+static bool scr_dyn_pipe_dest_off(ScrDynPipe *p, const char *event, size_t which) {
+  if (!p->l[which]) return true;
+  ScrDyn *args[2] = {scr_dyn_pipe_str(event, strlen(event)), p->l[which]};
+  bool ok = scr_dyn_pipe_call(p, "removeListener", args, 2);
+  scr_dyn_release(args[0]);
+  return ok;
+}
+
+static void scr_dyn_pipe_src_off(ScrDynPipe *p, const char *event, size_t which) {
+  if (!p->l[which] || !p->src) return;
+  ScrStr *name = scr_str_new(event, strlen(event));
+  scr_emitter_release(scr_emitter_off_dyn((ScrEmitter *)p->src, name, p->l[which]));
+  scr_str_release(name);
+}
+
+static void scr_dyn_pipe_ondrain(ScrDynPipe *p) {
+  p->awaiting_drain = false;
+  if (!p->src) return;
+  ScrStr *data = scr_str_new("data", 4);
+  double listeners = scr_emitter_listener_count((ScrEmitter *)p->src, data);
+  scr_str_release(data);
+  if (listeners > 0) scr_stream_release(scr_stream_resume(p->src));
+}
+
+/* pipe's cleanup(): every listener off, then a pending drain resumes. */
+static void scr_dyn_pipe_cleanup(ScrDynPipe *p) {
+  if (p->cleaned) return;
+  if (!scr_dyn_pipe_dest_off(p, "close", SCR_DYN_PIPE_CLOSE) ||
+      !scr_dyn_pipe_dest_off(p, "finish", SCR_DYN_PIPE_FINISH) ||
+      !scr_dyn_pipe_dest_off(p, "drain", SCR_DYN_PIPE_DRAIN) ||
+      !scr_dyn_pipe_dest_off(p, "error", SCR_DYN_PIPE_ERROR) ||
+      !scr_dyn_pipe_dest_off(p, "unpipe", SCR_DYN_PIPE_ON_UNPIPE))
+    return;
+  scr_dyn_pipe_src_off(p, "end", SCR_DYN_PIPE_END);
+  scr_dyn_pipe_src_off(p, "end", SCR_DYN_PIPE_UNPIPE);
+  scr_dyn_pipe_src_off(p, "data", SCR_DYN_PIPE_DATA);
+  p->cleaned = true;
+  if (p->l[SCR_DYN_PIPE_DRAIN] && p->awaiting_drain) scr_dyn_pipe_ondrain(p);
+  ScrDynPipe *held = scr_dyn_pipe_retain_v(p);
+  scr_dyn_pipe_drop(held);
+  scr_dyn_pipe_release_v(held);
+}
+
+/* src.unpipe(dest): the source pauses (its only pipe goes) and the
+ * destination hears 'unpipe' with { hasUnpiped: false }. */
+static void scr_dyn_pipe_unpipe(ScrDynPipe *p) {
+  if (p->unpiped || !p->src) return;
+  p->unpiped = true;
+  scr_stream_release(scr_stream_pause(p->src));
+  ScrDyn *info = scr_dyn_new_obj();
+  scr_dyn_obj_set(info, "hasUnpiped", 10, scr_dyn_new_bool(false));
+  ScrDyn *args[3] = {scr_dyn_pipe_str("unpipe", 6), scr_dyn_retain(p->src_box), info};
+  (void)scr_dyn_pipe_call(p, "emit", args, 3);
+  for (size_t i = 0; i < 3; i++) scr_dyn_release(args[i]);
+}
+
+static ScrDyn *scr_dyn_pipe_listener(ScrDynPipe *p, int kind);
+
+/* pipe's pause(): the source stops until the destination drains. */
+static void scr_dyn_pipe_pause(ScrDynPipe *p) {
+  if (!p->cleaned) {
+    p->awaiting_drain = true;
+    scr_stream_release(scr_stream_pause(p->src));
+  }
+  if (!p->l[SCR_DYN_PIPE_DRAIN] && p->dest) {
+    p->l[SCR_DYN_PIPE_DRAIN] = scr_dyn_pipe_listener(p, SCR_DYN_PIPE_DRAIN);
+    ScrDyn *args[2] = {scr_dyn_pipe_str("drain", 5), p->l[SCR_DYN_PIPE_DRAIN]};
+    (void)scr_dyn_pipe_call(p, "on", args, 2);
+    scr_dyn_release(args[0]);
+  }
+}
+
+static ScrDyn *scr_dyn_pipe_thunk(ScrClosure *clo, ScrDyn *const *args, size_t argc) {
+  ScrDynPipe *p = scr_box_get_ref(clo->caps[0]); /* +1 */
+  int kind = (int)scr_box_get_f64(clo->caps[1]);
+  ScrDyn *first = argc ? args[0] : scr_dyn_undefined();
+  switch (kind) {
+  case SCR_DYN_PIPE_DATA: {
+    if (!p->dest) break;
+    ScrDyn *ret = scr_dyn_invoke(p->dest, "write", &first, 1, "write");
+    if (ret) {
+      if (ret->kind == SCR_DYN_BOOL && !scr_dyn_truthy(ret)) scr_dyn_pipe_pause(p);
+      scr_dyn_release(ret);
+    }
+    break;
+  }
+  case SCR_DYN_PIPE_END:
+    if (p->dest) (void)scr_dyn_pipe_call(p, "end", NULL, 0);
+    break;
+  case SCR_DYN_PIPE_UNPIPE:
+    scr_dyn_pipe_unpipe(p);
+    break;
+  case SCR_DYN_PIPE_DRAIN:
+    scr_dyn_pipe_ondrain(p);
+    break;
+  case SCR_DYN_PIPE_ERROR: {
+    if (!p->dest) break;
+    ScrDyn *dest = scr_dyn_retain(p->dest);
+    ScrDyn *self = scr_dyn_retain(p->l[SCR_DYN_PIPE_ERROR]);
+    scr_dyn_pipe_unpipe(p);
+    if (!scr_exc_pending()) {
+      ScrDyn *off[2] = {scr_dyn_pipe_str("error", 5), self};
+      ScrDyn *r = scr_dyn_invoke(dest, "removeListener", off, 2, "removeListener");
+      if (r) scr_dyn_release(r);
+      ScrDyn *count = scr_exc_pending() ? NULL : scr_dyn_invoke(dest, "listenerCount", off, 1, "listenerCount");
+      if (count && count->kind == SCR_DYN_NUM && !scr_dyn_truthy(count)) {
+        ScrDyn *emit[2] = {off[0], first};
+        r = scr_dyn_invoke(dest, "emit", emit, 2, "emit");
+        if (r) scr_dyn_release(r);
+      }
+      if (count) scr_dyn_release(count);
+      scr_dyn_release(off[0]);
+    }
+    scr_dyn_release(self);
+    scr_dyn_release(dest);
+    break;
+  }
+  case SCR_DYN_PIPE_CLOSE:
+    if (scr_dyn_pipe_dest_off(p, "finish", SCR_DYN_PIPE_FINISH)) scr_dyn_pipe_unpipe(p);
+    break;
+  case SCR_DYN_PIPE_FINISH:
+    if (scr_dyn_pipe_dest_off(p, "close", SCR_DYN_PIPE_CLOSE)) scr_dyn_pipe_unpipe(p);
+    break;
+  case SCR_DYN_PIPE_ON_UNPIPE: {
+    ScrDyn *info = argc > 1 ? args[1] : NULL;
+    if (p->src_box && scr_dyn_strict_eq(first, p->src_box) && info && info->kind == SCR_DYN_OBJ) {
+      ScrDyn *has = scr_dyn_obj_read(info, "hasUnpiped", 10);
+      if (!has || scr_exc_pending()) { /* a throwing getter */
+        if (has) scr_dyn_release(has);
+        break;
+      }
+      bool unpiped = has->kind == SCR_DYN_BOOL && !scr_dyn_truthy(has);
+      scr_dyn_release(has);
+      if (unpiped) {
+        scr_dyn_obj_set(info, "hasUnpiped", 10, scr_dyn_new_bool(true));
+        scr_dyn_pipe_cleanup(p);
+      }
+    }
+    break;
+  }
+  }
+  scr_dyn_pipe_release_v(p);
+  return scr_exc_pending() ? NULL : scr_dyn_retain(scr_dyn_undefined());
+}
+
+static ScrDyn *scr_dyn_pipe_listener(ScrDynPipe *p, int kind) {
+  ScrClosure *clo = scr_closure_new((void *)scr_dyn_pipe_thunk, 2);
+  clo->caps[0] = scr_box_new_obj(scr_dyn_pipe_retain_v, scr_dyn_pipe_release_v, NULL);
+  scr_box_set_ref(clo->caps[0], scr_dyn_pipe_retain_v(p));
+  clo->caps[1] = scr_box_new(SCR_BOX_F64);
+  scr_box_set_f64(clo->caps[1], (double)kind);
+  return scr_dyn_new_func(clo, scr_dyn_pipe_thunk, 1, "native pipe listener", NULL);
+}
+
+/* destination.on / once / prependListener(event, listener) */
+static bool scr_dyn_pipe_listen(ScrDynPipe *p, const char *method, const char *event, int kind) {
+  p->l[kind] = scr_dyn_pipe_listener(p, kind);
+  ScrDyn *args[2] = {scr_dyn_pipe_str(event, strlen(event)), p->l[kind]};
+  bool ok = scr_dyn_pipe_call(p, method, args, 2);
+  scr_dyn_release(args[0]);
+  return ok;
+}
+
+static void scr_dyn_pipe_src_on(ScrDynPipe *p, const char *event, int kind, bool once) {
+  p->l[kind] = scr_dyn_pipe_listener(p, kind);
+  ScrStr *name = scr_str_new(event, strlen(event));
+  scr_stream_release(scr_stream_on_dyn(p->src, name, p->l[kind], once, false));
+  scr_str_release(name);
+}
+
+/* readable.pipe(destination, { end }) for a destination that is not a
+ * native stream. src and src_box (the source's boxed value) and dest are
+ * borrowed. Throws what the destination's members throw. */
+void scr_stream_pipe_dyn(ScrStream *src, ScrDyn *src_box, ScrDyn *dest, bool end) {
+  ScrDynPipe *p = calloc(1, sizeof *p);
+  if (!p) abort();
+  p->rc = 1;
+  p->src = scr_stream_retain(src);
+  p->src_box = scr_dyn_retain(src_box);
+  p->dest = scr_dyn_retain(dest);
+  p->end = end;
+  p->live_next = scr_dyn_pipe_live;
+  if (scr_dyn_pipe_live) scr_dyn_pipe_live->live_prev = p;
+  scr_dyn_pipe_live = p;
+  if (!scr_dyn_pipe_atexit_set) {
+    scr_dyn_pipe_atexit_set = true;
+    scr_atexit(scr_dyn_pipe_atexit);
+  }
+  /* The source's end ends the destination (end: true) or unpipes it. */
+  scr_dyn_pipe_src_on(p, "end", end ? SCR_DYN_PIPE_END : SCR_DYN_PIPE_UNPIPE, true);
+  if (scr_exc_pending() || !scr_dyn_pipe_listen(p, "on", "unpipe", SCR_DYN_PIPE_ON_UNPIPE)) goto out;
+  scr_dyn_pipe_src_on(p, "data", SCR_DYN_PIPE_DATA, false);
+  if (scr_exc_pending()) goto out;
+  {
+    /* A throwing getter's exception propagates out of pipe(), as Node's
+     * prependListener helper reads the member the same way. */
+    ScrDyn *prepend = scr_dyn_obj_read(dest, "prependListener", 15);
+    if (!prepend || scr_exc_pending()) {
+      if (prepend) scr_dyn_release(prepend);
+      goto out;
+    }
+    bool has_prepend = prepend->kind == SCR_DYN_FUNC;
+    scr_dyn_release(prepend);
+    if (!scr_dyn_pipe_listen(p, has_prepend ? "prependListener" : "on", "error", SCR_DYN_PIPE_ERROR))
+      goto out;
+  }
+  if (!scr_dyn_pipe_listen(p, "once", "close", SCR_DYN_PIPE_CLOSE) ||
+      !scr_dyn_pipe_listen(p, "once", "finish", SCR_DYN_PIPE_FINISH))
+    goto out;
+  {
+    ScrDyn *args[2] = {scr_dyn_pipe_str("pipe", 4), scr_dyn_retain(src_box)};
+    bool ok = scr_dyn_pipe_call(p, "emit", args, 2);
+    scr_dyn_release(args[0]);
+    scr_dyn_release(args[1]);
+    if (!ok) goto out;
+  }
+  {
+    ScrDyn *need = scr_dyn_obj_read(dest, "writableNeedDrain", 17);
+    if (!need || scr_exc_pending()) {
+      if (need) scr_dyn_release(need);
+      goto out;
+    }
+    bool pause = need->kind == SCR_DYN_BOOL && scr_dyn_truthy(need);
+    scr_dyn_release(need);
+    if (pause) scr_dyn_pipe_pause(p);
+    else if (scr_stream_is_paused(src))
+      scr_stream_release(scr_stream_resume(src));
+  }
+out:
+  scr_dyn_pipe_release_v(p);
+}
+
+/* readable.unpipe(destination) for a destination that is not a native
+ * stream: the pipe scr_stream_pipe_dyn registered from src into it
+ * unpipes, as Node's unpipe does for a piped destination; nothing happens
+ * when there is none. An undefined destination (unpipe() without one)
+ * unpipes every JavaScript destination of src, oldest pipe first.
+ * Borrowed. */
+void scr_stream_unpipe_dyn(ScrStream *src, ScrDyn *dest) {
+  bool all = dest == NULL || dest->kind == SCR_DYN_UNDEF;
+  for (;;) {
+    /* The live list is newest first; the oldest matching pipe goes next.
+     * Unpiping runs JavaScript ('unpipe' listeners), so each round walks
+     * the list again. */
+    ScrDynPipe *match = NULL;
+    for (ScrDynPipe *p = scr_dyn_pipe_live; p; p = p->live_next) {
+      if (p->src != src || p->unpiped || p->cleaned || !p->dest) continue;
+      if (!all && !scr_dyn_strict_eq(p->dest, dest)) continue;
+      match = p;
+    }
+    if (!match) return;
+    ScrDynPipe *held = scr_dyn_pipe_retain_v(match);
+    scr_dyn_pipe_unpipe(held);
+    scr_dyn_pipe_release_v(held);
+    if (!all || scr_exc_pending()) return;
+  }
+}
diff --git a/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts b/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts
new file mode 100644
index 0000000000..a72328d3fa
--- /dev/null
+++ b/tests/corpus/5819-node-stream-dyn-direct-pipe/main.ts
@@ -0,0 +1,59 @@
+// A native Readable piped straight into a JavaScript sink, the only
+// untyped stream flow in the program; unpipe of a NodeJS.WritableStream
+// that holds a native Writable or a JavaScript sink; unpipe() of every
+// JavaScript sink; and JavaScript destinations whose prependListener or
+// writableNeedDrain getter throws.
+import { Readable, Writable } from "node:stream";
+import { plainSink, throwingSink } from "./sources.js";
+
+function detach(src: Readable, dst: NodeJS.WritableStream): void {
+  src.unpipe(dst);
+}
+
+async function main(): Promise<void> {
+  Readable.from(["a", "b"]).pipe(plainSink("direct"));
+  await new Promise<void>((resolve) => setTimeout(resolve, 10));
+
+  const parts: string[] = [];
+  const native = new Writable({
+    write(chunk: Buffer, _encoding, done) {
+      parts.push(chunk.toString());
+      done();
+    },
+  });
+  native.on("unpipe", () => console.log("native heard unpipe"));
+  const source = new Readable({ read() {} });
+  source.pipe(native);
+  detach(source, native);
+  source.push("dropped");
+  await new Promise<void>((resolve) => setTimeout(resolve, 10));
+  console.log("native got", JSON.stringify(parts.join("")), source.isPaused());
+
+  const sink = plainSink("detached");
+  const second = new Readable({ read() {} });
+  second.pipe(sink);
+  detach(second, sink);
+  detach(second, plainSink("never piped"));
+  second.push("late");
+  await new Promise<void>((resolve) => setTimeout(resolve, 10));
+  console.log("second paused", second.isPaused());
+
+  const third = new Readable({ read() {} });
+  third.pipe(plainSink("first of two"));
+  third.pipe(plainSink("second of two"));
+  third.unpipe();
+  third.push("after unpipe()");
+  await new Promise<void>((resolve) => setTimeout(resolve, 10));
+  console.log("third paused", third.isPaused(), third.listenerCount("data"));
+
+  for (const member of ["prependListener", "writableNeedDrain"]) {
+    try {
+      new Readable({ read() {} }).pipe(throwingSink(member));
+      console.log(member, "piped");
+    } catch (error) {
+      console.log("pipe threw", (error as Error).message);
+    }
+  }
+}
+
+void main();
diff --git a/tests/corpus/5819-node-stream-dyn-direct-pipe/sources.js b/tests/corpus/5819-node-stream-dyn-direct-pipe/sources.js
new file mode 100644
index 0000000000..a06d04899e
--- /dev/null
+++ b/tests/corpus/5819-node-stream-dyn-direct-pipe/sources.js
@@ -0,0 +1,45 @@
+// A plain JavaScript sink, and sinks whose prependListener or
+// writableNeedDrain getter throws.
+
+/** @returns {any} */
+export function plainSink(label) {
+  const parts = [];
+  const listeners = {};
+  const sink = {
+    write(chunk) {
+      parts.push(String(chunk));
+      return true;
+    },
+    end() {
+      console.log(label, "sink end", parts.join(""));
+    },
+    on(event, listener) {
+      (listeners[event] ??= []).push(listener);
+      return sink;
+    },
+    once(event, listener) {
+      return sink.on(event, listener);
+    },
+    removeListener(event, listener) {
+      listeners[event] = (listeners[event] ?? []).filter((each) => each !== listener);
+      return sink;
+    },
+    emit(event, ...args) {
+      if (event === "unpipe") console.log(label, "heard unpipe");
+      for (const listener of listeners[event] ?? []) listener(...args);
+      return true;
+    },
+  };
+  return sink;
+}
+
+/** @returns {any} */
+export function throwingSink(member) {
+  const sink = plainSink(member);
+  Object.defineProperty(sink, member, {
+    get() {
+      throw new Error(`${member} getter failed`);
+    },
+  });
+  return sink;
+}
diff --git a/tests/corpus/6480-node-stream-dyn-readable/main.ts b/tests/corpus/6480-node-stream-dyn-readable/main.ts
new file mode 100644
index 0000000000..21d7bf2b13
--- /dev/null
+++ b/tests/corpus/6480-node-stream-dyn-readable/main.ts
@@ -0,0 +1,72 @@
+// NodeJS.ReadableStream slots hold native streams and JavaScript stream
+// objects alike: a plain object implementing on/pipe, a constructor-function
+// stream, and a native Readable pass through a TypeScript function typed
+// NodeJS.ReadableStream. TypeScript consumes them with on("data") and
+// pipe() into a native Writable, JavaScript consumes the returned values
+// (the same objects), and a native Readable typed NodeJS.ReadableStream
+// pipes into JavaScript sinks through Node's pipe protocol.
+import { Readable, Writable } from "node:stream";
+import { jsCollect, plainSink, plainSource, same, sink, source } from "./sources.js";
+
+function pass(stream: NodeJS.ReadableStream): NodeJS.ReadableStream {
+  return stream;
+}
+
+async function collect(label: string, stream: NodeJS.ReadableStream): Promise<void> {
+  const parts: string[] = [];
+  await new Promise<void>((resolve) => {
+    stream.on("data", (chunk: Buffer | string) => parts.push(chunk.toString()));
+    stream.on("end", () => resolve());
+  });
+  console.log(label, "on data", parts.join(""));
+}
+
+async function pipeInto(label: string, stream: NodeJS.ReadableStream): Promise<void> {
+  const parts: string[] = [];
+  await new Promise<void>((resolve) => {
+    const native = new Writable({
+      write(chunk: Buffer, _encoding, done) {
+        parts.push(chunk.toString());
+        done();
+      },
+    });
+    native.on("finish", () => resolve());
+    stream.pipe(native);
+  });
+  console.log(label, "pipe", parts.join(""));
+}
+
+async function main(): Promise<void> {
+  await collect("plain", pass(plainSource(["p1", "p2"])));
+  await collect("constructor", pass(source(["s1", "s2"])));
+  await collect("native", pass(Readable.from(["n1", "n2"])));
+
+  await pipeInto("plain", pass(plainSource(["p3", "p4"])));
+  await pipeInto("constructor", pass(source(["s3", "s4"])));
+  await pipeInto("native", pass(Readable.from(["n3", "n4"])));
+
+  console.log("plain js", await jsCollect(pass(plainSource(["p5", "p6"]))));
+  console.log("constructor js", await jsCollect(pass(source(["s5", "s6"]))));
+  console.log("native js", await jsCollect(pass(Readable.from(["n5", "n6"]))));
+
+  const plain = plainSource(["p7"]);
+  const made = source(["s7"]);
+  const native = Readable.from(["n7"]);
+  console.log("identity", same(plain, pass(plain)), same(made, pass(made)), same(native, pass(native)));
+
+  await new Promise<void>((resolve) => {
+    pass(Readable.from(["n8", "n9"])).pipe(plainSink("native into plain"));
+    pass(plainSource(["p8"])).pipe(plainSink("plain into plain"));
+    setTimeout(resolve, 20);
+  });
+
+  await new Promise<void>((resolve) => {
+    const target = sink("native into constructor");
+    target.on("pipe", () => console.log("sink heard pipe"));
+    target.on("unpipe", () => console.log("sink heard unpipe"));
+    target.on("finish", () => setImmediate(resolve));
+    pass(Readable.from(["n10", "n11"])).pipe(target);
+  });
+}
+
+void main();
diff --git a/tests/corpus/6480-node-stream-dyn-readable/sources.js b/tests/corpus/6480-node-stream-dyn-readable/sources.js
new file mode 100644
index 0000000000..3b520c29f5
--- /dev/null
+++ b/tests/corpus/6480-node-stream-dyn-readable/sources.js
@@ -0,0 +1,138 @@
+// JavaScript stream objects that are not native streams: a plain object
+// implementing on/pipe, and a constructor function with its own listener
+// table. Sinks: a plain object with write/end, and a constructor function
+// that keeps listeners, so a native pipe() into it runs Node's protocol.
+
+/** @returns {any} */
+export function plainSource(chunks) {
+  const listeners = {};
+  const source = {
+    on(event, listener) {
+      (listeners[event] ??= []).push(listener);
+      return source;
+    },
+    emit(event, value) {
+      for (const listener of listeners[event] ?? []) listener(value);
+    },
+    pipe(dest) {
+      source.on("data", (chunk) => dest.write(chunk));
+      source.on("end", () => dest.end());
+      return dest;
+    },
+  };
+  setImmediate(() => {
+    for (const chunk of chunks) source.emit("data", chunk);
+    source.emit("end");
+  });
+  return source;
+}
+
+function Source(chunks) {
+  this.chunks = chunks;
+  this.listeners = {};
+  this.readable = true;
+  setImmediate(() => {
+    for (const chunk of this.chunks) this.emit("data", chunk);
+    this.emit("end");
+  });
+}
+Source.prototype.on = function (event, listener) {
+  (this.listeners[event] = this.listeners[event] || []).push(listener);
+  return this;
+};
+Source.prototype.emit = function (event, value) {
+  for (const listener of this.listeners[event] || []) listener(value);
+  return true;
+};
+Source.prototype.pipe = function (dest) {
+  this.on("data", (chunk) => dest.write(chunk));
+  this.on("end", () => dest.end());
+  return dest;
+};
+
+/** @returns {any} */
+export function source(chunks) {
+  return new Source(chunks);
+}
+
+/** @returns {any} */
+export function plainSink(label) {
+  const parts = [];
+  return {
+    write(chunk) {
+      parts.push(String(chunk));
+      return true;
+    },
+    end() {
+      console.log(label, "sink end", parts.join(""));
+    },
+    on() {
+      return this;
+    },
+    once() {
+      return this;
+    },
+    emit() {
+      return false;
+    },
+  };
+}
+
+function Sink(label) {
+  this.label = label;
+  this.parts = [];
+  this.listeners = [];
+  this.writable = true;
+}
+Sink.prototype.on = function (event, listener) {
+  this.listeners.push({ event, listener, once: false });
+  return this;
+};
+Sink.prototype.once = function (event, listener) {
+  this.listeners.push({ event, listener, once: true });
+  return this;
+};
+Sink.prototype.prependListener = function (event, listener) {
+  this.listeners.unshift({ event, listener, once: false });
+  return this;
+};
+Sink.prototype.removeListener = function (event, listener) {
+  this.listeners = this.listeners.filter((l) => l.event !== event || l.listener !== listener);
+  return this;
+};
+Sink.prototype.listenerCount = function (event) {
+  return this.listeners.filter((l) => l.event === event).length;
+};
+Sink.prototype.emit = function (event, ...args) {
+  const matching = this.listeners.filter((l) => l.event === event);
+  this.listeners = this.listeners.filter((l) => l.event !== event || !l.once);
+  for (const l of matching) l.listener(...args);
+  return matching.length > 0;
+};
+Sink.prototype.write = function (chunk) {
+  this.parts.push(String(chunk));
+  return true;
+};
+Sink.prototype.end = function () {
+  console.log(this.label, "ended with", this.parts.join("|"), "listening to", this.listeners.map((l) => l.event).join(","));
+  this.emit("finish");
+  console.log(this.label, "after finish listening to", this.listeners.map((l) => l.event).join(","));
+};
+
+/** @returns {any} */
+export function sink(label) {
+  return new Sink(label);
+}
+
+/** @returns {Promise<string>} */
+export function jsCollect(stream) {
+  return new Promise((resolve) => {
+    const parts = [];
+    stream.on("data", (chunk) => parts.push(String(chunk)));
+    stream.on("end", () => resolve(parts.join("")));
+  });
+}
+
+export function same(a, b) {
+  return a === b;
+}
diff --git a/tests/corpus/6481-node-stream-dyn-writable/main.ts b/tests/corpus/6481-node-stream-dyn-writable/main.ts
new file mode 100644
index 0000000000..9ac429b8f8
--- /dev/null
+++ b/tests/corpus/6481-node-stream-dyn-writable/main.ts
@@ -0,0 +1,67 @@
+// NodeJS.WritableStream slots hold a native Writable, a plain JavaScript
+// sink and process.stdout: TypeScript writes and ends them, and a copy
+// helper typed over both interfaces pipes native and JavaScript sources
+// into native and JavaScript destinations.
+import { Readable, Writable } from "node:stream";
+import { plainSink, plainSource } from "./sources.js";
+
+function emit(label: string, out: NodeJS.WritableStream): void {
+  const ok = out.write(`${label} one\n`);
+  out.write(`${label} two\n`);
+  console.log(label, "write answered", typeof ok);
+}
+
+function finish(out: NodeJS.WritableStream): void {
+  out.end();
+}
+
+function copy(src: NodeJS.ReadableStream, dst: NodeJS.WritableStream): NodeJS.WritableStream {
+  return src.pipe(dst);
+}
+
+function collector(parts: string[]): Writable {
+  return new Writable({
+    write(chunk: Buffer, _encoding, done) {
+      parts.push(chunk.toString());
+      done();
+    },
+  });
+}
+
+async function main(): Promise<void> {
+  const parts: string[] = [];
+  const native = collector(parts);
+  const finished = new Promise<void>((resolve) => native.on("finish", () => resolve()));
+  emit("native", native);
+  finish(native);
+  await finished;
+  console.log("native got", JSON.stringify(parts.join("")));
+
+  const sink = plainSink("plain");
+  emit("plain", sink);
+  finish(sink);
+
+  emit("stdout", process.stdout);
+
+  const target = plainSink("copy native into plain");
+  console.log("copy returns target", copy(Readable.from(["c1", "c2"]), target) === target);
+  await new Promise<void>((resolve) => setTimeout(resolve, 10));
+  copy(plainSource(["c3", "c4"]), plainSink("copy plain into plain"));
+  await new Promise<void>((resolve) => setTimeout(resolve, 10));
+  const copied: string[] = [];
+  const native2 = collector(copied);
+  await new Promise<void>((resolve) => {
+    native2.on("finish", () => resolve());
+    copy(plainSource(["c5", "c6"]), native2);
+  });
+  console.log("copy plain into native", copied.join(""));
+  const copiedNative: string[] = [];
+  const native3 = collector(copiedNative);
+  await new Promise<void>((resolve) => {
+    native3.on("finish", () => resolve());
+    copy(Readable.from(["c7", "c8"]), native3);
+  });
+  console.log("copy native into native", copiedNative.join(""));
+}
+
+void main();
diff --git a/tests/corpus/6481-node-stream-dyn-writable/sources.js b/tests/corpus/6481-node-stream-dyn-writable/sources.js
new file mode 100644
index 0000000000..9c84914556
--- /dev/null
+++ b/tests/corpus/6481-node-stream-dyn-writable/sources.js
@@ -0,0 +1,49 @@
+// A plain JavaScript source implementing on/pipe and a plain sink with
+// write/end.
+
+/** @returns {any} */
+export function plainSource(chunks) {
+  const listeners = {};
+  const source = {
+    on(event, listener) {
+      (listeners[event] ??= []).push(listener);
+      return source;
+    },
+    emit(event, value) {
+      for (const listener of listeners[event] ?? []) listener(value);
+    },
+    pipe(dest) {
+      source.on("data", (chunk) => dest.write(chunk));
+      source.on("end", () => dest.end());
+      return dest;
+    },
+  };
+  setImmediate(() => {
+    for (const chunk of chunks) source.emit("data", chunk);
+    source.emit("end");
+  });
+  return source;
+}
+
+/** @returns {any} */
+export function plainSink(label) {
+  const parts = [];
+  return {
+    write(chunk) {
+      parts.push(String(chunk));
+      return true;
+    },
+    end() {
+      console.log(label, "sink end", parts.join(""));
+    },
+    on() {
+      return this;
+    },
+    once() {
+      return this;
+    },
+    emit() {
+      return false;
+    },
+  };
+}
diff --git a/tests/corpus/6482-node-stream-dyn-helpers/main.ts b/tests/corpus/6482-node-stream-dyn-helpers/main.ts
new file mode 100644
index 0000000000..db1c686a9b
--- /dev/null
+++ b/tests/corpus/6482-node-stream-dyn-helpers/main.ts
@@ -0,0 +1,98 @@
+// The helper shapes real code writes around NodeJS.ReadableStream, fed by
+// native and JavaScript streams: an async function returning one of
+// several producers, a JavaScript consumer of its result, a part piped
+// into a PassThrough with { end: false } and unpiped, pause/resume/
+// setEncoding/readable, an optional destroy through an intersection type,
+// for await and pipeline over native values in the slot.
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+import { PassThrough, Readable, Writable } from "node:stream";
+import { pipeline } from "node:stream/promises";
+import * as zlib from "node:zlib";
+import { openPart, part, streamToText } from "./sources.js";
+
+async function open(kind: string, file: string): Promise<NodeJS.ReadableStream> {
+  if (kind === "file") return fs.createReadStream(file);
+  return openPart(["x", "y"]);
+}
+
+async function pipePart(item: NodeJS.ReadableStream, out: PassThrough): Promise<void> {
+  item.pipe(out, { end: false });
+  await new Promise<void>((resolve) => item.on("end", () => resolve()));
+  item.unpipe(out);
+}
+
+function pass(stream: NodeJS.ReadableStream): NodeJS.ReadableStream {
+  return stream;
+}
+
+function stop(stream: NodeJS.ReadableStream): void {
+  (stream as NodeJS.ReadableStream & { destroy?: () => void }).destroy?.();
+}
+
+async function iterate(label: string, stream: NodeJS.ReadableStream): Promise<void> {
+  let text = "";
+  for await (const chunk of stream) text += chunk.toString();
+  console.log(label, "iterated", text);
+}
+
+async function main(): Promise<void> {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), "scriptc-stream-dyn-"));
+  const file = path.join(dir, "a.txt");
+  fs.writeFileSync(file, "file text");
+  console.log("file", await streamToText(await open("file", file)));
+  console.log("part", await streamToText(await open("part", file)));
+
+  const out = new PassThrough();
+  const parts: string[] = [];
+  out.on("data", (chunk: Buffer) => parts.push(chunk.toString()));
+  await pipePart(await open("part", file), out);
+  await pipePart(await open("file", file), out);
+  const ended = new Promise<void>((resolve) => out.on("end", () => resolve()));
+  out.end();
+  await ended;
+  console.log("out", parts.join("|"));
+
+  const js = pass(part(["j1"]));
+  js.pause();
+  js.resume();
+  js.setEncoding("utf8");
+  console.log("js readable", js.readable);
+  await new Promise<void>((resolve) => js.on("end", () => resolve()));
+  const native = pass(Readable.from(["n1"]));
+  native.setEncoding("utf8");
+  console.log("native readable", native.readable);
+  const chunks: string[] = [];
+  await new Promise<void>((resolve) => {
+    native.on("data", (chunk: Buffer | string) => chunks.push(chunk.toString()));
+    native.on("end", () => resolve());
+  });
+  console.log("native chunks", chunks.join(""));
+
+  const closing = pass(Readable.from(["c1"]));
+  closing.on("close", () => console.log("native closed"));
+  stop(pass(part(["d1"])));
+  stop(closing);
+
+  await iterate("native", pass(Readable.from(["i1", "i2"])));
+  await iterate("file", pass(fs.createReadStream(file)));
+
+  const packed = path.join(dir, "a.gz");
+  fs.writeFileSync(packed, zlib.gzipSync(Buffer.from("packed text")));
+  const unpacked: string[] = [];
+  await pipeline(
+    pass(fs.createReadStream(packed)),
+    zlib.createGunzip(),
+    new Writable({
+      write(chunk: Buffer, _encoding, done) {
+        unpacked.push(chunk.toString());
+        done();
+      },
+    }),
+  );
+  console.log("pipeline", unpacked.join(""));
+  fs.rmSync(dir, { recursive: true });
+}
+
+void main();
diff --git a/tests/corpus/6482-node-stream-dyn-helpers/sources.js b/tests/corpus/6482-node-stream-dyn-helpers/sources.js
new file mode 100644
index 0000000000..c395b43799
--- /dev/null
+++ b/tests/corpus/6482-node-stream-dyn-helpers/sources.js
@@ -0,0 +1,71 @@
+// A constructor-function stream part with its own listener table, whose
+// pipe() honors { end: false } and whose unpipe() detaches it, plus a
+// JavaScript consumer that reads a stream to text.
+
+function Part(chunks) {
+  this.chunks = chunks;
+  this.listeners = {};
+  this.dest = null;
+  this.ending = true;
+  this.readable = true;
+  setImmediate(() => {
+    for (const chunk of this.chunks) {
+      if (this.dest) this.dest.write(chunk);
+      else this.emit("data", chunk);
+    }
+    if (this.dest && this.ending) this.dest.end();
+    this.emit("end");
+  });
+}
+Part.prototype.on = function (event, listener) {
+  (this.listeners[event] = this.listeners[event] || []).push(listener);
+  return this;
+};
+Part.prototype.emit = function (event, value) {
+  for (const listener of this.listeners[event] || []) listener(value);
+  return true;
+};
+Part.prototype.pipe = function (dest, options) {
+  this.dest = dest;
+  this.ending = !options || options.end !== false;
+  return dest;
+};
+Part.prototype.unpipe = function () {
+  console.log("part unpipe");
+  this.dest = null;
+  return this;
+};
+Part.prototype.pause = function () {
+  console.log("part pause");
+  return this;
+};
+Part.prototype.resume = function () {
+  console.log("part resume");
+  return this;
+};
+Part.prototype.setEncoding = function (encoding) {
+  console.log("part setEncoding", encoding);
+  return this;
+};
+Part.prototype.destroy = function () {
+  console.log("part destroy");
+};
+
+/** @returns {Promise<any>} */
+export async function openPart(chunks) {
+  return new Part(chunks);
+}
+
+/** @returns {any} */
+export function part(chunks) {
+  return new Part(chunks);
+}
+
+/** @returns {any} */
+export function streamToText(stream) {
+  return new Promise((resolve) => {
+    const parts = [];
+    stream.on("data", (chunk) => parts.push(String(chunk)));
+    stream.on("end", () => resolve(parts.join("")));
+  });
+}
diff --git a/tests/harness/corpus-costs/5819-node-stream-dyn-direct-pipe/main.ts.json b/tests/harness/corpus-costs/5819-node-stream-dyn-direct-pipe/main.ts.json
new file mode 100644
index 0000000000..12637a176c
--- /dev/null
+++ b/tests/harness/corpus-costs/5819-node-stream-dyn-direct-pipe/main.ts.json
@@ -0,0 +1 @@
+{ "plain": 4912, "san": 12058 }