Run JavaScript WritableStream, TransformStream and pipes natively (#2437)

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

@@ -34,8 +34,8 @@
   A `Duplex.from` body function may take the `{ signal }` argument, which is aborted with the error the Duplex is destroyed with.
   `stream.compose` accepts streams and async generator functions in any position, an async function as the last stage, and an array, a string, or a generator as the first stage; one stage is `Duplex.from` of it.
   A web `ReadableStream` is read as an async iterable by `Readable.from` and as the first stage of `compose` or `pipeline`, and through its reader by `Readable.fromWeb` (with `objectMode`, `highWaterMark`, `encoding`, and `signal` in an options object literal) and `Duplex.from`.
-  `new TransformStream()` without arguments builds the identity transform: its readable side is a web `ReadableStream`, its writable side supports `getWriter`, `abort`, `close`, and `locked`, and the writer supports `write`, `close`, `abort`, `releaseLock`, `ready`, `closed`, and `desiredSize`, with Node's backpressure (a write completes when the readable side reads its chunk).
-  A web stream in a later compose stage, `Readable.toWeb`, a `TransformStream` with a transformer or queuing strategies, the `WritableStream` constructor, `pipeTo`, and `pipeThrough` are unsupported.
+  In TypeScript, `new TransformStream()` without arguments builds the identity transform: its readable side is a web `ReadableStream`, its writable side supports `getWriter`, `abort`, `close`, and `locked`, and the writer supports `write`, `close`, `abort`, `releaseLock`, `ready`, `closed`, and `desiredSize`, with Node's backpressure (a write completes when the readable side reads its chunk).
+  A web stream in a later compose stage and `Readable.toWeb` are unsupported, and in TypeScript, so are a `TransformStream` with a transformer or queuing strategies, the `WritableStream` constructor, `pipeTo`, and `pipeThrough`.
   The `ERR_INVALID_ARG_VALUE` message for a stream that is no longer readable or writable names the stream's class instead of printing the stream.
   `readable.compose(stage, { signal })` is `stream.compose(readable, stage)` with the signal added; the options are an object literal whose `signal` is a variable or a property read.
   `pipeline` from `node:stream/promises` accepts a trailing `{ signal, end }` object, with or without function stages, and `finished` accepts `readable`, `writable`, `error`, `cleanup`, and `signal` options, with boolean literals for the flags.
@@ -60,8 +60,9 @@
   The `readline/promises` `Readline` class writes to the same outputs.
   `readline.emitKeypressEvents` emits `'keypress'` events on `process.stdin` or a readable stream, shared with terminal interfaces on that stream as in Node; a listener's first parameter must admit `undefined`, which it receives for escape sequences.
   Chained listener registration is unsupported.
-  `node:stream/web` exports the same `ReadableStream` class as the global; in TypeScript, its other classes are unsupported.
-  In JavaScript, `ReadableStream`, `ReadableStreamDefaultReader`, `ReadableStreamDefaultController`, `CountQueuingStrategy`, and `ByteLengthQueuingStrategy`, as globals and as `node:stream/web` exports, are constructor values with Node's prototypes, brand checks, and errors.
-  Their underlying sources (`start`, `pull`, `cancel`), queuing strategies with a `size` function, readers, `tee`, `ReadableStream.from`, and async iteration settle promises in Node's order.
+  `node:stream/web` exports the same classes as the globals; in TypeScript, only `ReadableStream` and the identity `TransformStream` have lowerings.
+  In JavaScript, `ReadableStream`, `WritableStream`, and `TransformStream` with their default readers, writers, and controllers, `CountQueuingStrategy`, `ByteLengthQueuingStrategy`, `TextEncoderStream`, and `TextDecoderStream`, as globals and as `node:stream/web` exports, are constructor values with Node's prototypes, brand checks, and errors.
+  Their underlying sources, sinks, and transformers, queuing strategies with a `size` function, readers and writers, `tee`, `pipeTo` and `pipeThrough` with their options and an abort signal, `ReadableStream.from`, and async iteration settle promises in Node's order, and the streams take own properties.
+  `TextDecoderStream` decodes UTF-8; its other encodings are unsupported.
   A JavaScript `ReadableStream` given as a `Response` or `Request` body, or as `fetch`'s body, is read through a stream of its own: `response.body` is that stream rather than the one passed in, and the stream passed in is locked from the start.
-  In JavaScript, a byte stream (`type: 'bytes'`), `pipeTo`, `pipeThrough`, `Object.getPrototypeOf` on these streams, readers, controllers, and strategies, their use as `WeakMap` and `WeakSet` keys, `ReadableStream.from` of a function that is not iterable, and the other `node:stream/web` classes are unsupported.
+  In JavaScript, a byte stream (`type: 'bytes'`), a BYOB reader, a class that extends one of these classes, `Object.getPrototypeOf` on these values, their use as `WeakMap` and `WeakSet` keys, and `ReadableStream.from` of a function that is not iterable are unsupported.

docs/src/generated/node-v24-compatibility.json

@@ -81289,9 +81289,9 @@
         "inherited": true
       },
       "static": {
-        "status": "not-implemented",
-        "detail": "No static compiler lowering is registered for this API yet.",
-        "verification": "registry-gap"
+        "status": "partial",
+        "detail": "Implemented for the documented static async-iteration subset.",
+        "verification": "test-backed"
       }
     },
     {

internal/compatibility/generated/node-v24-internal.json

@@ -111929,6 +111929,8 @@
           "tests/corpus/stream-web-readable.ts",
           "tests/corpus/stream-web-require.cjs",
           "tests/corpus/web-stream-readable-untyped.mjs",
+          "tests/corpus/web-stream-transform-pipes-untyped.mjs",
+          "tests/corpus/web-stream-writable-untyped.mjs",
           "tests/harness/fetch-conformance.test.ts"
         ]
       },
@@ -113011,8 +113013,11 @@
         "inherited": true
       },
       "static": {
-        "status": "not-implemented",
-        "evidence": "compiler-unmatched:webstreams"
+        "status": "partial",
+        "evidence": "compiler-feature:webstreams.WritableStream",
+        "tests": [
+          "tests/corpus/web-stream-writable-untyped.mjs"
+        ]
       },
       "anchorSource": "exact"
     },
@@ -113405,7 +113410,8 @@
         "status": "partial",
         "evidence": "compiler-feature:webstreams.TransformStream",
         "tests": [
-          "tests/corpus/7743-transform-stream-identity.ts"
+          "tests/corpus/7743-transform-stream-identity.ts",
+          "tests/corpus/web-stream-transform-pipes-untyped.mjs"
         ]
       },
       "anchorSource": "exact"

internal/compatibility/static-support.json

@@ -434,7 +434,8 @@
     { "chapter": "dns", "symbols": ["dns.lookup", "dns.lookupService", "dns.getServers", "dns.getDefaultResultOrder", "dns.setDefaultResultOrder"], "status": "partial", "evidence": ["tests/corpus/5648-dns-lookup-forms.ts", "tests/corpus/5649-dns-service-servers.ts", "tests/corpus/5650-dns-lookup-errors-untyped.cjs"] },
     { "chapter": "dns", "symbols": ["dnsPromises.lookup", "dnsPromises.lookupService", "dnsPromises.getDefaultResultOrder"], "status": "partial", "evidence": ["tests/corpus/dns-promises-lookup.ts", "tests/corpus/dns-promises-untyped.cjs"] },
     { "chapter": "webstreams", "symbols": ["ReadableStream"], "status": "partial", "evidence": ["tests/corpus/stream-web-readable.ts", "tests/corpus/web-stream-readable-untyped.mjs", "tests/corpus/stream-web-require.cjs", "tests/corpus/2862-web-readable-async-iteration.ts"] },
-    { "chapter": "webstreams", "symbols": ["TransformStream"], "status": "partial", "evidence": ["tests/corpus/7743-transform-stream-identity.ts"] },
+    { "chapter": "webstreams", "symbols": ["TransformStream"], "status": "partial", "evidence": ["tests/corpus/7743-transform-stream-identity.ts", "tests/corpus/web-stream-transform-pipes-untyped.mjs"] },
+    { "chapter": "webstreams", "symbols": ["WritableStream"], "status": "partial", "evidence": ["tests/corpus/web-stream-writable-untyped.mjs"] },
     { "chapter": "globals", "signatures": ["Class: `TransformStream`"], "status": "partial", "evidence": ["tests/corpus/7743-transform-stream-identity.ts"] }
   ],
   "dedicated": [

packages/compiler/src/frontend/lowering/builtins/web-streams-dyn.ts

@@ -21,6 +21,13 @@ export const WEB_STREAM_CLASSES = [
   "ReadableStreamDefaultController",
   "CountQueuingStrategy",
   "ByteLengthQueuingStrategy",
+  "WritableStream",
+  "WritableStreamDefaultWriter",
+  "WritableStreamDefaultController",
+  "TransformStream",
+  "TransformStreamDefaultController",
+  "TextEncoderStream",
+  "TextDecoderStream",
 ] as const;
 
 const CLASS_NAMES: readonly string[] = WEB_STREAM_CLASSES;
@@ -48,6 +55,33 @@ export function webStreamClassKindOf(lowerer: Lowerer, node: ts.Expression): num
   return index < 0 ? null : index;
 }
 
+/** Whether a JavaScript source of the program names one of the classes.
+ * The global object answers them by a computed name only then
+ * (global-values.ts), so a program without web streams does not link
+ * them for its other computed global reads. */
+export function programNamesWebStreamClass(lowerer: Lowerer): boolean {
+  if (lowerer.namesWebStreamClass !== undefined) return lowerer.namesWebStreamClass;
+  const names: ReadonlySet<string> = new Set(WEB_STREAM_CLASSES);
+  let found = false;
+  for (const file of lowerer.program.getSourceFiles()) {
+    if (file.isDeclarationFile || !isJsSourceFile(file)) continue;
+    ts.walkPreorder(file, (node) => {
+      if (
+        ts.isIdentifier(node) &&
+        names.has(node.text) &&
+        webStreamClassKindOf(lowerer, node) !== null
+      ) {
+        found = true;
+        return "stop";
+      }
+      return undefined;
+    });
+    if (found) break;
+  }
+  lowerer.namesWebStreamClass = found;
+  return found;
+}
+
 export function webStreamClassValue(kind: number, loc: SrcLoc): IrExpr {
   return { kind: "libCall", fn: "webstream.class", args: [numLit(kind, loc)], type: DYN, loc };
 }
@@ -130,6 +164,8 @@ const WEB_STREAM_INTERFACES: ReadonlySet<string> = new Set([
   "ReadableStreamReadValueResult",
   "ReadableStreamReadDoneResult",
   "UnderlyingSource",
+  "UnderlyingSink",
+  "Transformer",
   "QueuingStrategy",
   "QueuingStrategySize",
 ]);

packages/compiler/src/frontend/lowering/global-values.ts

@@ -32,6 +32,11 @@ import { builtinNamespaceValue } from "./builtin-namespaces.js";
 import { uriFunctionValue } from "./expressions/identifiers.js";
 import { navigatorValue } from "./builtins/navigator.js";
 import { primitiveCtorClosure } from "./lower-exprs.js";
+import {
+  WEB_STREAM_CLASSES,
+  programNamesWebStreamClass,
+  webStreamClassValue,
+} from "./builtins/web-streams-dyn.js";
 
 const LOOKUP_FN = "%global.builtinValue";
 
@@ -177,9 +182,14 @@ export function globalValuesInstall(
     const name: IrExpr = { kind: "varRef", localId: "name.0", type: STRING, loc };
     const body: IrStmt[] = [];
     for (const global of [...new Set(names)].sort()) {
+      const webStream = (WEB_STREAM_CLASSES as readonly string[]).indexOf(global);
       const value: IrExpr | null = REFUSED.has(global)
         ? { kind: "unitLit", unit: "null", type: { kind: "nullT" }, loc }
-        : jsGlobalValueByName(lowerer, global, loc);
+        : webStream >= 0
+          ? programNamesWebStreamClass(lowerer)
+            ? webStreamClassValue(webStream, loc)
+            : null
+          : jsGlobalValueByName(lowerer, global, loc);
       if (value === null) continue;
       body.push({
         kind: "if",

packages/compiler/src/frontend/lowering/lowerer.ts

@@ -3645,6 +3645,9 @@ export class Lowerer {
   /** global-values.ts's answer per JavaScript source: the receiver of its
    * first read of the global object by a computed name, or null. */
   readonly computedGlobalReads = new Map<ts.SourceFile, ts.Expression | null>();
+  /** web-streams-dyn.ts's answer: whether a JavaScript source names a web
+   * stream class; undefined until asked. */
+  namesWebStreamClass: boolean | undefined = undefined;
   /** The first lifted function of each name (liftedFnNamed), for the
    * liftedFns indexed so far. liftedFns only grows. */
   private readonly liftedFnsByName = new Map<string, IrFunction>();
@@ -4256,6 +4259,7 @@ export class Lowerer {
     "createRequireBindingTargets",
     "createRequireElementTargets",
     "computedGlobalReads",
+    "namesWebStreamClass",
     "timeValueDateDecisions",
     "timeValueDateWrites",
     "dynamicOverloadAnswers",

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

@@ -1044,8 +1044,9 @@ const RUNTIME_FEATURE_RULES = {
   mime: rule({ prefixes: ["mime."] }),
   /** Web streams in checked-dynamic code (scr_webstreams.c). */
   webStreams: rule({ prefixes: ["webstream."] }),
-  /** AbortController and AbortSignal (scr_abort.c, or the fetch unit). */
-  abortSignal: rule({ prefixes: ["abort."] }),
+  /** AbortController and AbortSignal (scr_abort.c, or the fetch unit).
+   * A WritableStream controller's signal is one (scr_webstreams.c). */
+  abortSignal: rule({ prefixes: ["abort.", "webstream."] }),
   /** util.debounce and util.throttle (scr_util_timing.c). */
   utilTiming: rule({ fns: ["util.debounce", "util.throttle"] }),
   /** events.once(emitter, name, { signal }) (scr_events_once_signal.c). */

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/web-stream-transform-pipes-untyped.mjs.json

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

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

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

packages/runtime/src/scr_async_dyn.c

@@ -1551,8 +1551,11 @@ static ScrDyn *scr_dyn_then_of(const ScrDyn *value, const ScrDyn *reader) {
   return NULL;
 }
 
-/* Resolves `target` with `value` (borrowed). */
-static void scr_dyn_resolve_value(ScrPromise *target, ScrDyn *value, const ScrDyn *reader) {
+/* Resolves `target` with `value` (borrowed). Inlined into both callers, so
+ * the code of scr_dyn_then_resolve_call does not depend on whether a
+ * program links scr_dyn_promise_resolve_with. */
+__attribute__((always_inline)) static inline void scr_dyn_resolve_value(ScrPromise *target, ScrDyn *value,
+                                                                const ScrDyn *reader) {
   if (value->kind == SCR_DYN_PROMISE) {
     scr_promise_resolve_dyn(target, scr_dyn_retain(value));
     return;
@@ -1669,6 +1672,10 @@ ScrPromise *scr_dyn_promise_resolve(const ScrDyn *value, const ScrDyn *reader) {
   return scr_dyn_promise_entry(NULL, value, reader);
 }
 
+void scr_dyn_promise_resolve_with(ScrPromise *target, ScrDyn *value) {
+  scr_dyn_resolve_value(target, value, NULL);
+}
+
 static void scr_dyn_settled_store(ScrArr *a, double i, ScrPromise *src) {
   ScrDyn *record = scr_dyn_new_obj();
   bool fulfilled = !scr_promise_is_rejected(src);

packages/runtime/src/scr_dyn.h

@@ -244,7 +244,14 @@ typedef enum {
   Y(WS_COUNT_STRATEGY) /* CountQueuingStrategy (scr_webstreams.c) */ \
   Y(WS_RCONTROLLER) /* ReadableStreamDefaultController (scr_webstreams.c) */ \
   Y(WS_READABLE) /* ReadableStream for checked-dynamic code (scr_webstreams.c) */ \
-  Y(WS_READER) /* ReadableStreamDefaultReader (scr_webstreams.c) */
+  Y(WS_READER) /* ReadableStreamDefaultReader (scr_webstreams.c) */ \
+  Y(WS_TCONTROLLER) /* TransformStreamDefaultController (scr_webstreams.c) */ \
+  Y(WS_TEXT_DECODER) /* TextDecoderStream (scr_webstreams.c) */ \
+  Y(WS_TEXT_ENCODER) /* TextEncoderStream (scr_webstreams.c) */ \
+  Y(WS_TRANSFORM) /* TransformStream for checked-dynamic code (scr_webstreams.c) */ \
+  Y(WS_WCONTROLLER) /* WritableStreamDefaultController (scr_webstreams.c) */ \
+  Y(WS_WRITABLE) /* WritableStream for checked-dynamic code (scr_webstreams.c) */ \
+  Y(WS_WRITER) /* WritableStreamDefaultWriter (scr_webstreams.c) */
 
 /* Values below the largest pinned one that no tag holds, each with the
  * reason. Never reuse one without removing it here. */
@@ -2058,6 +2065,12 @@ typedef struct ScrDynHandleClass {
   /* The prototype's methods and getters in definition order (the order
    * Object.keys reports); NULL = the methods, then the getters. */
   const char *const *order;
+  /* A call of one of the class's method values
+   * (`WritableStream.prototype.close.call(stream)`): the class's method,
+   * which an own property of the same name on the handle does not
+   * replace. NULL = ScrDynHandleOps.invoke. */
+  ScrDyn *(*method_invoke)(void *h, ScrDyn *self, const char *method, ScrDyn *const *args, size_t argc,
+                           const char *what);
 } ScrDynHandleClass;
 
 /* Installs a tag's ops with the class its handles belong to; both outlive
@@ -2084,6 +2097,11 @@ ScrDyn *scr_dyn_handle_class_prototype_of(const ScrDyn *handle);
  * receiver (+1); NULL when the chain has none or with the exception
  * pending. */
 ScrDyn *scr_dyn_handle_class_member(ScrDynHandleTag tag, void *h, const char *key, size_t len);
+/* A method value's call on a handle of its tag (`self`, borrowed): the
+ * class's method_invoke, or the ops' invoke for a tag without one. +1, or
+ * NULL with the exception pending. */
+ScrDyn *scr_dyn_handle_class_method_call(const ScrDynHandleOps *ops, ScrDyn *self, const char *method,
+                                         ScrDyn *const *args, size_t argc);
 
 /* The operations a Transform handle gives the stream unit (scr_stream.c):
  * the slot where the handle keeps its stream (+1) with the stream's
@@ -2160,6 +2178,11 @@ ScrPromise *scr_dyn_promise_any(const ScrDyn *entries, const ScrDyn *reader);
  * itself, a thenable adopts through its resolve job, a throwing `then`
  * read rejects, and any other value fulfills. +1. */
 ScrPromise *scr_dyn_promise_resolve(const ScrDyn *value, const ScrDyn *reader);
+/* Resolves `target` with a dyn value (borrowed), as a promise's resolve
+ * function does: a native promise adopts, a thenable adopts through its
+ * resolve job, a throwing `then` read rejects, and any other value
+ * fulfills. */
+void scr_dyn_promise_resolve_with(ScrPromise *target, ScrDyn *value);
 
 /* Native predicates for typed reference capsules and boxed errors. */
 bool scr_dyn_ref_typeof_is(const ScrDyn *d, const char *name);

packages/runtime/src/scr_dyn_handle_class.c

@@ -183,6 +183,14 @@ ScrDyn *scr_dyn_handle_class_member(ScrDynHandleTag tag, void *h, const char *ke
   return value;
 }
 
+ScrDyn *scr_dyn_handle_class_method_call(const ScrDynHandleOps *ops, ScrDyn *self, const char *method,
+                                         ScrDyn *const *args, size_t argc) {
+  const ScrDynHandleClass *klass = hc_classes[self->v.handle.tag];
+  if (klass && klass->method_invoke)
+    return klass->method_invoke(self->v.handle.ptr, self, method, args, argc, method);
+  return ops->invoke(self->v.handle.ptr, self, method, args, argc, method);
+}
+
 static void hc_define_method(ScrDyn *prototype, ScrDynHandleTag tag, const char *name) {
   ScrDyn *fn = scr_dyn_handle_class_method(tag, name);
   if (fn) hc_define(prototype, name, fn, true, true, true);

packages/runtime/src/scr_json.c

@@ -7004,7 +7004,14 @@ static ScrDyn *scr_dynh_method_call(ScrClosure *closure, ScrDyn *const *args, si
   ScrDyn *self = scr_dyn_this_get();
   ScrDyn *result;
   if (self->kind == SCR_DYN_HANDLE && self->v.handle.tag == tag) {
+#ifdef SCR_HANDLE_CLASSES
+    /* A class's method value reaches the class's method past the handle's
+     * own properties; this build is the one programs with such classes
+     * link. */
+    result = scr_dyn_handle_class_method_call(ops, self, method->name, args, argc);
+#else
     result = ops->invoke(self->v.handle.ptr, self, method->name, args, argc, method->name);
+#endif
   } else {
     const char *message = method->invalid_this ? method->invalid_this
                           : ops->invalid_this  ? ops->invalid_this

tests/corpus/web-stream-transform-pipes-untyped.mjs

@@ -0,0 +1,212 @@
+// Web TransformStream and pipes from untyped JavaScript: transformers with
+// start, transform, flush and cancel, backpressure between the sides,
+// controller errors and terminate, pipeTo and pipeThrough with their
+// options, abort signals, own properties on streams and method wrappers
+// over them, a transformer start that returns a thenable, and encoder and
+// decoder streams, in Node's promise order.
+import { TransformStream as WebTransformStream, TransformStreamDefaultController } from "node:stream/web";
+
+const log = [];
+const out = (...args) => log.push(args.map(String).join(" "));
+const caught = (fn) => {
+  try {
+    fn();
+    return "ok";
+  } catch (e) {
+    return `${e.name}:${e.code ?? ""}:${e.message}`;
+  }
+};
+const settled = (p) => p.then((v) => `ok:${v}`, (e) => `${e?.name}:${e?.code ?? ""}:${e?.message ?? e}`);
+const collect = async (readable) => {
+  const got = [];
+  for await (const chunk of readable) got.push(chunk);
+  return got.join(",");
+};
+
+out("classes", TransformStream === WebTransformStream, TransformStream.name, TransformStream.length, caught(() => new TransformStreamDefaultController()));
+out("prototype", Object.keys(TransformStream.prototype).join(","), Object.keys(TransformStreamDefaultController.prototype).join(","));
+
+{
+  const order = [];
+  const t = new TransformStream({
+    start(c) {
+      order.push(`start ${c.desiredSize}`);
+      c.enqueue("S");
+    },
+    transform(chunk, c) {
+      order.push(`transform ${chunk}`);
+      c.enqueue(chunk.toUpperCase());
+    },
+    flush(c) {
+      order.push("flush");
+      c.enqueue("F");
+    },
+  });
+  const writer = t.writable.getWriter();
+  writer.write("a");
+  writer.write("b");
+  writer.close();
+  out("transform", await collect(t.readable), order.join(" | "), await settled(writer.closed));
+}
+
+{
+  const t = new TransformStream({}, undefined, { highWaterMark: 1 });
+  const writer = t.writable.getWriter();
+  out("backpressure", writer.desiredSize, t.readable.locked);
+  const first = writer.write("x");
+  const second = writer.write("y");
+  out("queued", writer.desiredSize);
+  const reader = t.readable.getReader();
+  out("read", (await reader.read()).value, (await reader.read()).value, await settled(first), await settled(second));
+}
+
+{
+  const failure = new Error("transform-failure");
+  const t = new TransformStream({ transform() { throw failure; } });
+  const writer = t.writable.getWriter();
+  const reader = t.readable.getReader();
+  const read = settled(reader.read());
+  const write = settled(writer.write("z"));
+  out("transform-error", await write, await read, await settled(writer.closed));
+}
+
+{
+  /** @type {any} */
+  let controller;
+  const t = new TransformStream({ start(c) { controller = c; } });
+  controller.enqueue("before");
+  controller.terminate();
+  const reader = t.readable.getReader();
+  out("terminate", (await reader.read()).value, (await reader.read()).done, await settled(t.writable.getWriter().write("after")), caught(() => controller.enqueue("late")));
+}
+
+{
+  const cancels = [];
+  const t = new TransformStream({ cancel(reason) { cancels.push(`cancel ${reason}`); } });
+  out("cancel", await settled(t.readable.cancel("stop")), cancels.join(","), await settled(t.writable.getWriter().closed));
+}
+
+{
+  const source = ReadableStream.from(["p", "q", "r"]);
+  const got = [];
+  const sink = new WritableStream({ write(chunk) { got.push(chunk); } });
+  const piped = source.pipeThrough(new TransformStream({ transform(chunk, c) { c.enqueue(chunk + chunk); } })).pipeTo(sink);
+  out("pipe", await settled(piped), got.join(","), source.locked);
+}
+
+{
+  const source = new ReadableStream({ start(c) { c.enqueue(1); c.close(); } });
+  const sink = new WritableStream();
+  out("prevent-close", await settled(source.pipeTo(sink, { preventClose: true })), sink.locked, await settled(sink.getWriter().write(2)));
+}
+
+{
+  const failure = new Error("source-failure");
+  const source = new ReadableStream({ start(c) { c.error(failure); } });
+  const aborts = [];
+  const sink = new WritableStream({ abort(reason) { aborts.push(reason === failure); } });
+  out("source-error", await settled(source.pipeTo(sink)), aborts.join(","));
+}
+
+{
+  const controller = new AbortController();
+  const reason = new Error("signal");
+  const source = new ReadableStream({ pull() {}, cancel(e) { out("signal-cancel", e === reason); } }, { highWaterMark: 0 });
+  const sink = new WritableStream({ abort(e) { out("signal-abort", e === reason); } });
+  const piped = source.pipeTo(sink, { signal: controller.signal });
+  await Promise.resolve();
+  controller.abort(reason);
+  out("signal", await settled(piped), source.locked, sink.locked);
+}
+
+{
+  const source = ReadableStream.from(["own"]);
+  const sink = new WritableStream();
+  /** @type {any} */ (source).getReader = () => { throw new Error("public reader"); };
+  /** @type {any} */ (sink).tag = "kept";
+  await source.pipeTo(sink);
+  out("own-properties", /** @type {any} */ (sink).tag, typeof /** @type {any} */ (source).getReader, caught(() => { /** @type {any} */ (source).locked = true; }));
+}
+
+{
+  // An own property that wraps the saved prototype method reaches the
+  // class's method.
+  const calls = [];
+  const wrap = (target, name, method) => {
+    target[name] = function (...args) {
+      calls.push(name);
+      return method.apply(this, args);
+    };
+  };
+  const writable = new WritableStream({ write(chunk) { calls.push(`sink:${chunk}`); } });
+  wrap(writable, "getWriter", WritableStream.prototype.getWriter);
+  const writer = writable.getWriter();
+  wrap(writer, "write", WritableStreamDefaultWriter.prototype.write);
+  wrap(writer, "close", WritableStreamDefaultWriter.prototype.close);
+  await writer.write("a");
+  await writer.close();
+  const readable = ReadableStream.from(["b"]);
+  wrap(readable, "getReader", ReadableStream.prototype.getReader);
+  const reader = readable.getReader();
+  wrap(reader, "read", ReadableStreamDefaultReader.prototype.read);
+  const first = await reader.read();
+  const t = new TransformStream({
+    transform(chunk, controller) {
+      wrap(controller, "enqueue", TransformStreamDefaultController.prototype.enqueue);
+      controller.enqueue(`${chunk}!`);
+    },
+  });
+  t.writable.getWriter().write("c");
+  const transformed = await t.readable.getReader().read();
+  out("wrappers", calls.join(","), first.value, transformed.value, writable.locked);
+}
+
+{
+  // A transformer start that returns a thenable: both sides start when it
+  // settles, and its rejection errors the stream.
+  const order = [];
+  const later = { then(resolve) { order.push("then"); setTimeout(() => { order.push("resolve"); resolve(); }, 0); } };
+  const t = new TransformStream({ start() { return later; }, transform(chunk, c) { order.push(`transform:${chunk}`); c.enqueue(chunk); } });
+  Promise.resolve().then(() => order.push("tick1")).then(() => order.push("tick2")).then(() => order.push("tick3"));
+  const written = t.writable.getWriter().write("x").then(() => order.push("written"));
+  const read = await t.readable.getReader().read();
+  await written;
+  out("start-thenable", order.join(","), read.value);
+  const ticks = [];
+  const sync = new TransformStream({ start() { return { then(resolve) { ticks.push("then"); resolve(); } }; }, transform(chunk, c) { ticks.push("transform"); c.enqueue(chunk); } });
+  Promise.resolve().then(() => ticks.push("tick1")).then(() => ticks.push("tick2")).then(() => ticks.push("tick3")).then(() => ticks.push("tick4"));
+  sync.writable.getWriter().write("z").then(() => ticks.push("written"));
+  sync.readable.getReader().read().then(() => ticks.push("read"));
+  await new Promise((r) => setTimeout(r, 0));
+  out("start-thenable-ticks", ticks.join(","));
+  const failing = new TransformStream({ start() { return { then(_, reject) { reject(new Error("start failed")); } }; } });
+  out("start-thenable-reject", await settled(failing.writable.getWriter().write("y")), await settled(failing.readable.getReader().read()));
+  const throwing = new TransformStream({ start() { return { get then() { throw new Error("then getter"); } }; } });
+  out("start-then-getter", await settled(throwing.writable.getWriter().write("y")));
+  const plain = new TransformStream({ start() { return { then: 1 }; } });
+  const pending = plain.readable.getReader().read();
+  out("start-object", await settled(plain.writable.getWriter().write("p")), (await pending).value);
+}
+
+out(
+  "invalid",
+  await settled(ReadableStream.from([]).pipeTo(/** @type {any} */ ({}))),
+  caught(() => ReadableStream.from([]).pipeThrough(/** @type {any} */ ({}))),
+  caught(() => new TransformStream({ readableType: "bytes" })),
+  caught(() => new TransformStream({ transform: 1 })),
+);
+
+{
+  const bytes = [];
+  await ReadableStream.from(["hé", "\ud83d", "\ude00", "\ud800"])
+    .pipeThrough(new TextEncoderStream())
+    .pipeTo(new WritableStream({ write(chunk) { bytes.push(Array.from(chunk).join(".")); } }));
+  out("encoder", bytes.join(" | "), new TextEncoderStream().encoding, Object.prototype.toString.call(new TextEncoderStream()));
+  const text = [];
+  await ReadableStream.from([new Uint8Array([0xe2, 0x82]), new Uint8Array([0xac, 0x21]), new Uint8Array([0xf0])])
+    .pipeThrough(new TextDecoderStream())
+    .pipeTo(new WritableStream({ write(chunk) { text.push(chunk); } }));
+  const decoder = new TextDecoderStream("utf-8", { fatal: true });
+  out("decoder", text.join("|"), decoder.encoding, decoder.fatal, decoder.ignoreBOM, caught(() => new TextDecoderStream("nope")));
+}
+console.log(log.join("\n"));

tests/corpus/web-stream-writable-untyped.mjs

@@ -0,0 +1,144 @@
+// Web WritableStream from untyped JavaScript: the classes as constructor
+// values, prototypes and brand checks, writers with backpressure and their
+// ready and closed promises, queuing strategies, sink errors, close and
+// abort ordering, the controller's signal, and argument errors, in Node's
+// promise order.
+import { WritableStream as WebWritableStream, WritableStreamDefaultWriter, WritableStreamDefaultController } from "node:stream/web";
+
+const log = [];
+const out = (...args) => log.push(args.map(String).join(" "));
+const caught = (fn) => {
+  try {
+    fn();
+    return "ok";
+  } catch (e) {
+    return `${e.name}:${e.code ?? ""}:${e.message}`;
+  }
+};
+const settled = (p) => p.then((v) => `ok:${v}`, (e) => `${e?.name}:${e?.code ?? ""}:${e?.message ?? e}`);
+
+out("classes", typeof WritableStream, WritableStream === WebWritableStream, WritableStream.name, WritableStream.length, WritableStreamDefaultWriter.length);
+out("illegal", caught(() => new WritableStreamDefaultController()), caught(() => /** @type {any} */ (WritableStream)()));
+out("prototype", Object.keys(WritableStream.prototype).join(","), Object.keys(WritableStreamDefaultWriter.prototype).join(","), Object.keys(WritableStreamDefaultController.prototype).join(","));
+
+{
+  const order = [];
+  const chunks = [];
+  const stream = new WritableStream(
+    {
+      start() { order.push("start"); },
+      write(chunk, controller) {
+        order.push(`write ${chunk}`);
+        chunks.push(chunk);
+        return new Promise((resolve) => setTimeout(resolve, 1));
+      },
+      close() { order.push("close"); },
+    },
+    new CountQueuingStrategy({ highWaterMark: 2 }),
+  );
+  const writer = stream.getWriter();
+  out("writer", writer instanceof WritableStreamDefaultWriter, stream.locked, writer.desiredSize, caught(() => stream.getWriter()));
+  const writes = [writer.write("a"), writer.write("b"), writer.write("c")];
+  out("queued", writer.desiredSize, Object.prototype.toString.call(stream), Object.prototype.toString.call(writer));
+  let readySettled = false;
+  writer.ready.then(() => { readySettled = true; });
+  await Promise.resolve();
+  out("backpressure", readySettled, order.join(" | "));
+  for (const w of writes) await w;
+  await writer.ready;
+  out("drained", writer.desiredSize, readySettled, chunks.join(""));
+  await writer.close();
+  out("closed", await settled(writer.closed), writer.desiredSize, order.join(" | "), await settled(writer.write("late")));
+  writer.releaseLock();
+  out("released", stream.locked, await settled(writer.closed), caught(() => writer.desiredSize));
+}
+
+{
+  const sizes = [];
+  const stream = new WritableStream(
+    { write() {} },
+    { highWaterMark: 10, size(chunk) { sizes.push(chunk.length); return chunk.length; } },
+  );
+  const writer = stream.getWriter();
+  const good = writer.write("abcd");
+  out("sizes", writer.desiredSize, sizes.join(","));
+  const bad = writer.write({ length: -1 });
+  out("bad-size", await settled(good), await settled(bad), await settled(writer.closed), writer.desiredSize);
+}
+
+{
+  const failure = new Error("sink-write");
+  const stream = new WritableStream({ write() { throw failure; } });
+  const writer = stream.getWriter();
+  const first = writer.write("x");
+  const second = writer.write("y");
+  out("write-error", await settled(first), await settled(second), await settled(writer.closed), await settled(writer.ready));
+}
+
+{
+  const log2 = [];
+  /** @type {any} */
+  let controller;
+  const reason = new Error("stop");
+  const stream = new WritableStream({
+    start(c) {
+      controller = c;
+      c.signal.addEventListener("abort", () => log2.push(`signal ${c.signal.reason === reason}`));
+    },
+    write(chunk) { log2.push(`write ${chunk}`); },
+    abort(e) { log2.push(`abort ${e === reason}`); },
+  });
+  const writer = stream.getWriter();
+  await writer.write("one");
+  const abort = writer.abort(reason);
+  log2.push("aborting");
+  out("abort", await settled(abort), log2.join(" | "), controller.signal.aborted, await settled(writer.closed), writer.desiredSize);
+}
+
+{
+  /** @type {any} */
+  let controller;
+  const stream = new WritableStream({ start(c) { controller = c; } });
+  const writer = stream.getWriter();
+  controller.error(new Error("controller-error"));
+  out("controller-error", await settled(writer.write("z")), await settled(writer.closed), writer.desiredSize, await settled(stream.abort()));
+}
+
+{
+  const stream = new WritableStream();
+  const close = stream.close();
+  out("direct-close", await settled(stream.close()), await settled(close), await settled(stream.abort()));
+}
+
+out(
+  "brands",
+  caught(() => WritableStream.prototype.getWriter.call({})),
+  await settled(WritableStream.prototype.abort.call({})),
+  await settled(/** @type {any} */ (Object.getOwnPropertyDescriptor(WritableStreamDefaultWriter.prototype, "ready")).get.call({})),
+  caught(() => /** @type {any} */ (Object.getOwnPropertyDescriptor(WritableStreamDefaultWriter.prototype, "desiredSize")).get.call({})),
+  caught(() => WritableStreamDefaultController.prototype.error.call({})),
+);
+out(
+  "invalid",
+  caught(() => new WritableStream(null)),
+  caught(() => new WritableStream({ type: "bytes" })),
+  caught(() => new WritableStream({ write: 1 })),
+  caught(() => new WritableStream({ start: 1 })),
+  caught(() => new WritableStream({}, { highWaterMark: -1 })),
+  caught(() => new WritableStreamDefaultWriter(/** @type {any} */ ({}))),
+);
+{
+  const classes = { WritableStream, WritableStreamDefaultWriter, TransformStream };
+  const isInstance = (value, C) => value instanceof C;
+  const stream = new WritableStream();
+  const transform = new TransformStream();
+  out(
+    "by-value",
+    isInstance(stream, classes.WritableStream),
+    isInstance(stream.getWriter(), classes.WritableStreamDefaultWriter),
+    isInstance(transform, classes.TransformStream),
+    isInstance(transform.writable, classes.WritableStream),
+    isInstance(stream, classes.TransformStream),
+  );
+}
+console.log(log.join("\n"));

tests/harness/corpus-costs/web-stream-transform-pipes-untyped.mjs.json

@@ -0,0 +1 @@
+{ "plain": 4300, "san": 55000 }

tests/harness/corpus-costs/web-stream-writable-untyped.mjs.json

@@ -0,0 +1 @@
+{ "plain": 3900, "san": 6700 }

tests/harness/npm.test.ts

@@ -57,6 +57,7 @@ const KNOWN: Record<string, Known> = {
     divergence:
       "Timeout handles are numbers (typeof) and lose their ref flag once fired or cleared",
   },
+  // require() inside a function.
   "island-web-plumbing": { run: "SC1090" },
   "lazy-traps": { build: "SC2013" },
   "misc-shims": { run: "SC1090" },
@@ -87,15 +88,6 @@ const KNOWN: Record<string, Known> = {
   "node24-url-mutation": { run: "SC1090" },
   "node24-util-adapters": { run: "Object.create with this value is not supported yet" },
   "node24-util-types": { run: "SC1090" },
-  "node24-web-stream-encoding": { run: "SC2020" },
-  "node24-web-stream-errors": { run: "SC1090" },
-  "node24-web-stream-lifecycle": { run: "SC2020" },
-  "node24-web-stream-pipe-abort": { run: "SC2020" },
-  "node24-web-stream-pipe-errors": { run: "SC2020" },
-  "node24-web-stream-pipes": { run: "SC2020" },
-  "node24-web-stream-queues": { run: "SC2020" },
-  "node24-web-stream-tee": { run: "SC1090" },
-  "node24-web-stream-transforms": { run: "SC1090" },
   "package-imports": { build: "SC2013" },
   "promise-bridge": { build: "SC1090" },
   "self-name": { build: "SC2013" },
@@ -103,6 +95,7 @@ const KNOWN: Record<string, Known> = {
   "util-shims": { run: "SC2020" },
   // The package redeclares the var it exported: Node copies the first value.
   "var-export-redeclared": { run: "SC1090" },
+  // `class Upper extends TransformStream`: extending a builtin class from JavaScript.
   "web-streams": { run: "SC1090" },
   "workspace-copied": { build: "SC2013" },
   "workspace-unclean": { build: "SC2013" },