Lower ReadableStream tee, pipeTo and pipeThrough in TypeScript (#2513)

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

@@ -35,7 +35,9 @@
   `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`.
   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`.
+  In TypeScript, `tee`, `pipeTo`, and `pipeThrough`, the `WritableStream` constructor with an underlying sink, and a `TransformStream` with a transformer or queuing strategies compile: a program that uses one of them builds its `ReadableStream`, `WritableStream`, and `TransformStream` values with the same classes as JavaScript, and a `Response` or `Request` body has those members too.
+  A TypeScript program that uses only `getReader`, `cancel`, `locked`, and the identity `TransformStream` keeps the smaller native streams.
+  A web stream in a later compose stage and `Readable.toWeb` are unsupported.
   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,7 +62,7 @@
   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 classes as the globals; in TypeScript, only `ReadableStream` and the identity `TransformStream` have lowerings.
+  `node:stream/web` exports the same classes as the globals; in TypeScript, `ReadableStream`, `WritableStream`, and `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.

packages/compiler/ambient/scriptc-node-fallback.d.ts

@@ -1115,9 +1115,9 @@ declare var ReadableStream: {
   from(iterable: string): ReadableStream<string>;
   readonly prototype: ReadableStream<unknown>;
 };
-/* The writable side of a TransformStream and its writer. The
- * WritableStream constructor has no native lowering; writable streams come
- * from `new TransformStream()`. */
+/* The writable side of a TransformStream and its writer. A program that
+ * constructs a WritableStream, or a TransformStream with a transformer,
+ * builds its streams with the web stream classes (scr_webstreams.c). */
 interface WritableStreamDefaultWriter<W = unknown> {
   readonly closed: Promise<void>;
   readonly desiredSize: number | null;
@@ -1133,14 +1133,25 @@ interface WritableStream<W = unknown> {
   close(): Promise<void>;
   getWriter(): WritableStreamDefaultWriter<W>;
 }
-/* The identity transform: chunks written to `writable` are read from
- * `readable`, and a write completes when its chunk is read. */
+/* Without a transformer, the identity transform: chunks written to
+ * `writable` are read from `readable`, and a write completes when its chunk
+ * is read. */
 interface TransformStream<I = unknown, O = unknown> {
   readonly readable: ReadableStream<O>;
   readonly writable: WritableStream<I>;
 }
+interface Transformer<I = unknown, O = unknown> {
+  start?(controller: TransformStreamDefaultController<O>): unknown;
+  transform?(chunk: I, controller: TransformStreamDefaultController<O>): unknown;
+  flush?(controller: TransformStreamDefaultController<O>): unknown;
+  cancel?(reason?: unknown): unknown;
+}
 declare var TransformStream: {
-  new <I = unknown, O = I>(): TransformStream<I, O>;
+  new <I = unknown, O = I>(
+    transformer?: Transformer<I, O>,
+    writableStrategy?: QueuingStrategy<I>,
+    readableStrategy?: QueuingStrategy<O>,
+  ): TransformStream<I, O>;
   readonly prototype: TransformStream<unknown, unknown>;
 };
 /* The other web stream interfaces. JavaScript sources construct and read
@@ -1158,8 +1169,14 @@ interface WritableStreamDefaultController {
   readonly signal: AbortSignal;
   error(reason?: unknown): void;
 }
+interface UnderlyingSink<W = unknown> {
+  start?(controller: WritableStreamDefaultController): unknown;
+  write?(chunk: W, controller: WritableStreamDefaultController): unknown;
+  close?(): unknown;
+  abort?(reason?: unknown): unknown;
+}
 declare var WritableStream: {
-  new <W = unknown>(sink?: object, strategy?: QueuingStrategy<W>): WritableStream<W>;
+  new <W = unknown>(sink?: UnderlyingSink<W>, strategy?: QueuingStrategy<W>): WritableStream<W>;
   readonly prototype: WritableStream<unknown>;
 };
 declare var WritableStreamDefaultWriter: {

packages/compiler/src/compat/fetch-profile.ts

@@ -155,6 +155,12 @@ const typedInterfaceUnsupported = "typed source has no compiler bridge for this
 const metadataExclusion =
   "WebIDL brand metadata is observable reflection, not an executable compatibility operation";
 
+/** ReadableStream methods a typed program reaches through the web stream
+ * classes (scr_webstreams.c): on a stream it constructs with them, and on
+ * the native stream through its web view (scr_fetch.c, sf_stream_web_view).
+ * A program that calls one links the classes. */
+export const READABLE_STREAM_WEB_VIEW_CALLS = ["pipeThrough", "pipeTo", "tee"] as const;
+
 function buildFetchCompatProfile() {
   return {
     schemaVersion: 1,
@@ -233,7 +239,7 @@ function buildFetchCompatProfile() {
       ],
       responseCalls: ["json", "text", "bytes", "clone"],
       readableStreamReads: ["locked"],
-      readableStreamCalls: ["cancel", "getReader"],
+      readableStreamCalls: ["cancel", "getReader", ...READABLE_STREAM_WEB_VIEW_CALLS],
     },
     operations: [
       {
@@ -447,6 +453,13 @@ function buildFetchCompatProfile() {
         facets: ["webidl-conversion", "identity", "state-machine", "error-shape"],
         evidence: [generated("stream-traces"), fixture("static-stream")],
       },
+      ...READABLE_STREAM_WEB_VIEW_CALLS.map((member): FetchCompatOperation => ({
+        id: `stdlib.readable-stream.${member}`,
+        name: `ReadableStream.${member}`,
+        kind: "method",
+        facets: ["identity", "promise-settlement", "state-machine", "error-shape"],
+        evidence: [fixture("static-web-streams")],
+      })),
       {
         id: "stdlib.readable-stream-default-reader.closed",
         name: "ReadableStreamDefaultReader.closed",
@@ -757,25 +770,15 @@ function buildFetchCompatProfile() {
           "getReader",
           "prototype",
         ),
-        ...["pipeThrough", "values"].map((member) =>
-          unsupportedEntry(
-            `stdlib.readable-stream.${member}`,
-            "ReadableStream",
-            member,
-            "prototype",
-            member === "values"
-              ? "direct for-await consumption compiles statically; storing or driving the first-class iterator handle is unsupported"
-              : "the wider Web Streams graph is outside the native readable-stream slice",
-          ),
+        ...READABLE_STREAM_WEB_VIEW_CALLS.map((member) =>
+          staticEntry(`stdlib.readable-stream.${member}`, "ReadableStream", member, "prototype"),
         ),
-        ...["pipeTo", "tee"].map((member) =>
-          unsupportedEntry(
-            `stdlib.readable-stream.${member}`,
-            "ReadableStream",
-            member,
-            "prototype",
-            "this operation is outside the native readable-stream implementation",
-          ),
+        unsupportedEntry(
+          "stdlib.readable-stream.values",
+          "ReadableStream",
+          "values",
+          "prototype",
+          "direct for-await consumption compiles statically; storing or driving the first-class iterator handle is unsupported",
         ),
         unsupportedEntry(
           "stdlib.readable-stream.symbol.asyncIterator",
@@ -954,7 +957,8 @@ export interface StaticFetchMembers {
   readableStreamReads: ReadonlySet<string>;
   readableStreamCalls: ReadonlySet<string>;
   /** ReadableStream prototype members the native stream does not
-   * implement, symbol-keyed ones as `[Symbol.name]`. */
+   * implement itself, symbol-keyed ones as `[Symbol.name]`: the unsupported
+   * ones, and the calls it answers through its web view. */
   readableStreamUnsupported: ReadonlySet<string>;
 }
 
@@ -979,8 +983,8 @@ export function staticFetchMembers(): StaticFetchMembers {
     ),
     readableStreamReads: new Set(profile.members.readableStreamReads),
     readableStreamCalls: new Set(profile.members.readableStreamCalls),
-    readableStreamUnsupported: new Set(
-      profile.inventory.entries
+    readableStreamUnsupported: new Set([
+      ...profile.inventory.entries
         .filter(
           (entry) =>
             entry.owner === "ReadableStream" &&
@@ -988,7 +992,8 @@ export function staticFetchMembers(): StaticFetchMembers {
             entry.status === "unsupported",
         )
         .map((entry) => entry.member),
-    ),
+      ...READABLE_STREAM_WEB_VIEW_CALLS,
+    ]),
   };
   return staticFetchMemberSets;
 }

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

@@ -1,4 +1,7 @@
-import { staticFetchMembers } from "../../../compat/fetch-profile.js";
+import {
+  READABLE_STREAM_WEB_VIEW_CALLS,
+  staticFetchMembers,
+} from "../../../compat/fetch-profile.js";
 import { numLit } from "../../../ir/build.js";
 import { everyStmt } from "../../../ir/traverse.js";
 import {
@@ -16,6 +19,7 @@ import { constituentTypes } from "../../ts7/checker.js";
 import { ENTRY_NAME, type Lowerer } from "../lowerer.js";
 import { stdlibGlobalNameOf } from "../surfaces.js";
 import { streamWebImportOf } from "./module-bindings.js";
+import { lowerDynObjectLiteral } from "../expressions/object-literals.js";
 
 /* The web stream classes in JavaScript sources: the globals and the
  * node:stream/web exports, which are the same constructors. Each reference
@@ -92,6 +96,60 @@ export function programNamesWebStreamClass(lowerer: Lowerer): boolean {
   return found;
 }
 
+/** The classes a TypeScript source constructs through the web stream
+ * classes once it uses an operation the native streams lack. */
+const TYPED_CLASSES: ReadonlySet<string> = new Set([
+  "ReadableStream",
+  "WritableStream",
+  "TransformStream",
+]);
+const WEB_VIEW_CALLS: ReadonlySet<string> = new Set(READABLE_STREAM_WEB_VIEW_CALLS);
+
+/** Whether a TypeScript source of the program uses a web stream operation
+ * the native streams lack: a ReadableStream tee(), pipeTo() or
+ * pipeThrough() call, a WritableStream constructor, or a TransformStream
+ * constructor with arguments (a transformer or queuing strategies). Such a
+ * program builds its ReadableStream, WritableStream and TransformStream
+ * values with the web stream classes, as JavaScript does; a program that
+ * uses only getReader(), cancel() and the identity transform keeps the
+ * native streams. */
+export function programUsesTypedWebStreams(lowerer: Lowerer): boolean {
+  if (lowerer.typedWebStreams !== undefined) return lowerer.typedWebStreams;
+  let found = false;
+  for (const file of lowerer.program.getSourceFiles()) {
+    if (file.isDeclarationFile || isJsSourceFile(file)) continue;
+    ts.walkPreorder(file, (node) => {
+      const member = staticMemberName(node);
+      if (member !== null && WEB_VIEW_CALLS.has(member) && isReadableStreamMember(lowerer, node))
+        found = true;
+      else if (ts.isNewExpression(node)) {
+        const name = stdlibGlobalNameOf(lowerer, unparenthesized(node.expression));
+        if (
+          name === "WritableStream" ||
+          (name === "TransformStream" && (node.arguments?.nodes.length ?? 0) > 0)
+        )
+          found = true;
+      }
+      return found ? "stop" : undefined;
+    });
+    if (found) break;
+  }
+  lowerer.typedWebStreams = found;
+  return found;
+}
+
+/** The class a TypeScript `new` expression constructs through the web
+ * stream classes (programUsesTypedWebStreams), as its index in
+ * WEB_STREAM_CLASSES; null otherwise. */
+function typedWebStreamClassKindOf(lowerer: Lowerer, node: ts.Expression): number | null {
+  const expr = unparenthesized(node);
+  if (!ts.isIdentifier(expr) || isJsSourceFile(ts.getSourceFileOfNode(expr))) return null;
+  const name = streamWebImportOf(lowerer, expr) ?? stdlibGlobalNameOf(lowerer, expr);
+  if (name === null || !TYPED_CLASSES.has(name) || !programUsesTypedWebStreams(lowerer))
+    return null;
+  return CLASS_NAMES.indexOf(name);
+}
+
 export function webStreamClassValue(kind: number, loc: SrcLoc): IrExpr {
   return { kind: "libCall", fn: "webstream.class", args: [numLit(kind, loc)], type: DYN, loc };
 }
@@ -102,13 +160,24 @@ export function lowerWebStreamClassRead(lowerer: Lowerer, node: ts.Expression):
   return kind === null ? null : webStreamClassValue(kind, locOf(node));
 }
 
-/** `new X(...args)` on one of the classes in JavaScript. */
+/** `new X(...args)` on one of the classes in JavaScript, and on
+ * ReadableStream, WritableStream and TransformStream in a TypeScript
+ * program that uses the web stream classes (programUsesTypedWebStreams).
+ * A TypeScript source's object literal argument (a source, sink,
+ * transformer or strategy) becomes a plain object with its methods. */
 export function lowerWebStreamNew(lowerer: Lowerer, expr: ts.NewExpression): IrExpr | null {
-  const kind = webStreamClassKindOf(lowerer, expr.expression);
+  const typed = webStreamClassKindOf(lowerer, expr.expression) === null;
+  const kind = typed
+    ? typedWebStreamClassKindOf(lowerer, expr.expression)
+    : webStreamClassKindOf(lowerer, expr.expression);
   if (kind === null) return null;
   const nodes = expr.arguments?.nodes ?? [];
   if (nodes.some(ts.isSpreadElement)) return null;
   const loc = locOf(expr);
+  const argument = (arg: ts.Expression): IrExpr =>
+    typed && ts.isObjectLiteralExpression(arg)
+      ? lowerDynObjectLiteral(lowerer, arg)
+      : lowerer.lowerExprExpecting(arg, DYN);
   return {
     kind: "libCall",
     fn: "dyn.construct",
@@ -116,7 +185,7 @@ export function lowerWebStreamNew(lowerer: Lowerer, expr: ts.NewExpression): IrE
       webStreamClassValue(kind, loc),
       {
         kind: "dynArrLit",
-        elems: nodes.map((arg) => lowerer.lowerExprExpecting(arg, DYN)),
+        elems: nodes.map(argument),
         type: DYN,
         loc,
       },
@@ -244,10 +313,29 @@ function isReadableStreamType(lowerer: Lowerer, type: ts.Type): boolean {
   );
 }
 
-/** Whether a member access resolves to a member of the stdlib
- * ReadableStream interface. */
-function isReadableStreamMember(lowerer: Lowerer, access: ts.PropertyAccessExpression): boolean {
-  const member = lowerer.checker.getSymbolAtLocation(access.name);
+type MemberAccess = ts.PropertyAccessExpression | ts.ElementAccessExpression;
+
+/** The member a property access names, or an element access with a string
+ * literal key (`body.tee` and `body["tee"]`); null for any other node. */
+function staticMemberName(node: ts.Node): string | null {
+  if (ts.isPropertyAccessExpression(node)) return node.name.text;
+  if (ts.isElementAccessExpression(node)) {
+    const key = node.argumentExpression;
+    if (ts.isStringLiteral(key) || ts.isNoSubstitutionTemplateLiteral(key)) return key.text;
+  }
+  return null;
+}
+
+/** Whether a member access (staticMemberName) resolves to a member of the
+ * stdlib ReadableStream interface. */
+function isReadableStreamMember(lowerer: Lowerer, access: ts.Node): boolean {
+  const key = ts.isPropertyAccessExpression(access)
+    ? access.name
+    : ts.isElementAccessExpression(access)
+      ? access.argumentExpression
+      : null;
+  if (key === null) return false;
+  const member = lowerer.checker.getSymbolAtLocation(key);
   if (member === undefined || !lowerer.isStdlibSymbol(member)) return false;
   let owner: ts.Node | undefined = member.declarations[0]?.parent;
   while (owner !== undefined && !ts.isInterfaceDeclaration(owner)) owner = owner.parent;
@@ -285,6 +373,7 @@ function programLinksFetch(functions: IrFunction[]): boolean {
  * count in a program that links the fetch unit, since the value can be a
  * fetch body. */
 function programUsesWebStreamOnlyMembers(lowerer: Lowerer, functions: IrFunction[]): boolean {
+  if (programUsesTypedWebStreams(lowerer)) return true;
   const members = staticFetchMembers().readableStreamUnsupported;
   const iterates = members.has("[Symbol.asyncIterator]");
   let untyped = false;
@@ -293,9 +382,10 @@ function programUsesWebStreamOnlyMembers(lowerer: Lowerer, functions: IrFunction
     let found = false;
     ts.walkPreorder(file, (node) => {
       let receiver: ts.Type | null = null;
-      if (ts.isPropertyAccessExpression(node) && members.has(node.name.text)) {
+      const member = staticMemberName(node);
+      if (member !== null && members.has(member)) {
         if (isReadableStreamMember(lowerer, node)) found = true;
-        else receiver = lowerer.typeOf(node.expression);
+        else receiver = lowerer.typeOf((node as MemberAccess).expression);
       } else if (iterates && ts.isForOfStatement(node) && node.awaitModifier !== undefined) {
         receiver = lowerer.typeOf(node.expression);
         if (isReadableStreamType(lowerer, receiver)) found = true;
@@ -310,7 +400,9 @@ function programUsesWebStreamOnlyMembers(lowerer: Lowerer, functions: IrFunction
 }
 
 /** Links the web stream classes into a program whose JavaScript uses a
- * ReadableStream member the fetch unit's own stream lacks: a fetch body's
+ * ReadableStream member the fetch unit's own stream lacks, or whose
+ * TypeScript uses the web stream classes (programUsesTypedWebStreams): a
+ * fetch body's or a native stream's
  * handle stands for a web ReadableStream once such a member is used
  * (scr_fetch.c, sf_stream_web_view). %main installs the classes first. A
  * program that names one of the classes links them already. */

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

@@ -1425,7 +1425,7 @@ export function fenceStaticReadableStreamMember(
   lowerer.noLowering(
     `ReadableStream.${member} in a static build`,
     access,
-    "the native static ReadableStream surface is locked plus cancel() and getReader(); values()/[Symbol.asyncIterator]() compile when consumed directly by for await, while stored iterator handles remain unsupported",
+    "ReadableStream's locked compiles as a read, and getReader(), cancel(), tee(), pipeTo() and pipeThrough() as direct calls; values()/[Symbol.asyncIterator]() compile when consumed directly by for await, while method values and stored iterator handles remain unsupported",
     sym,
   );
 }
@@ -1965,8 +1965,9 @@ export function lowerStaticReadableStreamNew(
 
 /** `new TransformStream()`: the identity transform, a native handle whose
  * readable side is a native ReadableStream and whose writable side is the
- * runtime's WritableStream. A transformer or queuing strategies have no
- * native lowering yet and keep the SC2020 refusal. */
+ * runtime's WritableStream. A TransformStream with a transformer or
+ * queuing strategies is built with the web stream classes instead
+ * (web-streams-dyn.ts, lowerWebStreamNew). */
 export function lowerStaticTransformStreamNew(
   lowerer: Lowerer,
   expr: ts.NewExpression,

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

@@ -3859,6 +3859,9 @@ export class Lowerer {
   /** web-streams-dyn.ts's answer: whether a JavaScript source names a web
    * stream class; undefined until asked. */
   namesWebStreamClass: boolean | undefined = undefined;
+  /** web-streams-dyn.ts's answer: whether a TypeScript source uses a web
+   * stream operation the native streams lack; undefined until asked. */
+  typedWebStreams: 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>();
@@ -4493,6 +4496,7 @@ export class Lowerer {
     "sloppyThisGlobalReads",
     "globalLookupNames",
     "namesWebStreamClass",
+    "typedWebStreams",
     "timeValueDateDecisions",
     "timeValueDateWrites",
     "dynamicOverloadAnswers",

packages/compiler/src/frontend/type-mapper.ts

@@ -220,8 +220,10 @@ export const FETCH_AMBIENT_TYPES = [
   "ReadableStreamReadValueResult",
   "ReadableStreamReadDoneResult",
   "TransformStream",
+  "TransformStreamDefaultController",
   "WritableStream",
   "WritableStreamDefaultWriter",
+  "WritableStreamDefaultController",
   "Blob",
   "File",
   "BlobPropertyBag",

packages/compiler/surface-manifest.json

@@ -5039,17 +5039,15 @@
       "id": "stdlib.readable-stream.pipeThrough",
       "kind": "stdlib",
       "name": "ReadableStream.pipeThrough",
-      "status": "unsupported",
-      "code": "SC2020",
-      "note": "Node 24.22.0 / Undici 7.29.1; the wider Web Streams graph is outside the native readable-stream slice"
+      "status": "static",
+      "note": "Node 24.22.0 / Undici 7.29.1; facets: identity, promise-settlement, state-machine, error-shape; differential evidence: fixture:static-web-streams"
     },
     {
       "id": "stdlib.readable-stream.pipeTo",
       "kind": "stdlib",
       "name": "ReadableStream.pipeTo",
-      "status": "unsupported",
-      "code": "SC2020",
-      "note": "Node 24.22.0 / Undici 7.29.1; this operation is outside the native readable-stream implementation"
+      "status": "static",
+      "note": "Node 24.22.0 / Undici 7.29.1; facets: identity, promise-settlement, state-machine, error-shape; differential evidence: fixture:static-web-streams"
     },
     {
       "id": "stdlib.readable-stream.symbol.asyncIterator",
@@ -5063,9 +5061,8 @@
       "id": "stdlib.readable-stream.tee",
       "kind": "stdlib",
       "name": "ReadableStream.tee",
-      "status": "unsupported",
-      "code": "SC2020",
-      "note": "Node 24.22.0 / Undici 7.29.1; this operation is outside the native readable-stream implementation"
+      "status": "static",
+      "note": "Node 24.22.0 / Undici 7.29.1; facets: identity, promise-settlement, state-machine, error-shape; differential evidence: fixture:static-web-streams"
     },
     {
       "id": "stdlib.readable-stream.values",

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/web-streams-typed-element-tee.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/web-streams-typed-element-tee.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/web-streams-typed-pipes.ts.json

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

tests/corpus/web-streams-typed-element-tee.ts

@@ -0,0 +1,23 @@
+// ReadableStream tee() reached through an element access with a string key
+// is the program's only web stream operation: the program still links the
+// web stream classes, so a Response body splits into two branches.
+
+async function text(stream: ReadableStream<Uint8Array>): Promise<string> {
+  const reader = stream.getReader();
+  const decoder = new TextDecoder();
+  let out = "";
+  for (;;) {
+    const result = await reader.read();
+    if (result.done) break;
+    out += decoder.decode(result.value, { stream: true });
+  }
+  return out + decoder.decode();
+}
+
+async function main(): Promise<void> {
+  const body = new Response("split me").body!;
+  const [left, right] = body["tee"]();
+  console.log("element tee", await text(left), "|", await text(right), body.locked);
+}
+
+void main();

tests/corpus/web-streams-typed-pipes.ts

@@ -0,0 +1,158 @@
+// ReadableStream tee(), pipeTo() and pipeThrough() in TypeScript, with
+// WritableStream sinks and TransformStream transformers: chunk order, close
+// and flush, cancel through a tee branch, the pipe options, Node's
+// TypeErrors for locked streams, error propagation through a pipe, and a
+// Response body piped through a transform.
+
+const describe = (e: unknown): string => `${(e as Error).name}: ${(e as Error).message}`;
+
+function source(chunks: string[]): ReadableStream<string> {
+  return new ReadableStream<string>({
+    start(controller) {
+      for (const chunk of chunks) controller.enqueue(chunk);
+      controller.close();
+    },
+  });
+}
+
+async function drain<T>(stream: ReadableStream<T>): Promise<T[]> {
+  const out: T[] = [];
+  const reader = stream.getReader();
+  for (;;) {
+    const result = await reader.read();
+    if (result.done) break;
+    out.push(result.value);
+  }
+  return out;
+}
+
+async function tees(): Promise<void> {
+  const [left, right] = source(["a", "b", "c"]).tee();
+  console.log("tee", (await drain(left)).join(""), (await drain(right)).join(""));
+
+  const log: string[] = [];
+  const watched = new ReadableStream<number>({
+    start(controller) {
+      controller.enqueue(1);
+      controller.enqueue(2);
+    },
+    cancel(reason) {
+      log.push(`source cancel ${String(reason)}`);
+    },
+  });
+  const [a, b] = watched.tee();
+  // The source is cancelled once both branches are, with both reasons.
+  const leftDone = a.cancel("left done").then(() => log.push("left cancelled"));
+  log.push("left requested");
+  await Promise.all([leftDone, b.cancel("right done").then(() => log.push("right cancelled"))]);
+  console.log("tee cancel", log.join(" | "), watched.locked);
+
+  const locked = source(["x"]);
+  locked.getReader();
+  try {
+    locked.tee();
+  } catch (e) {
+    console.log("tee locked", describe(e));
+  }
+}
+
+async function pipes(): Promise<void> {
+  const written: string[] = [];
+  const sink = new WritableStream<string>({
+    write(chunk) {
+      written.push(chunk);
+    },
+    close() {
+      written.push("<close>");
+    },
+  });
+  await source(["one", "two"]).pipeTo(sink);
+  console.log("pipeTo", written.join(","), sink.locked);
+
+  const kept: string[] = [];
+  const open = new WritableStream<string>({
+    write(chunk) {
+      kept.push(chunk);
+    },
+    close() {
+      kept.push("<close>");
+    },
+  });
+  await source(["p"]).pipeTo(open, { preventClose: true });
+  const writer = open.getWriter();
+  await writer.write("after");
+  await writer.close();
+  console.log("preventClose", kept.join(","));
+
+  const failing = new ReadableStream<string>({
+    start(controller) {
+      controller.enqueue("ok");
+      controller.error(new RangeError("source failed"));
+    },
+  });
+  const aborted: string[] = [];
+  try {
+    await failing.pipeTo(
+      new WritableStream<string>({
+        write(chunk) {
+          aborted.push(chunk);
+        },
+        abort(reason) {
+          aborted.push(`abort ${describe(reason)}`);
+        },
+      }),
+    );
+  } catch (e) {
+    console.log("pipeTo error", describe(e), aborted.join(" | "));
+  }
+
+  const busy = source(["q"]);
+  busy.getReader();
+  try {
+    await busy.pipeTo(new WritableStream<string>());
+  } catch (e) {
+    console.log("pipeTo locked", describe(e));
+  }
+}
+
+async function transforms(): Promise<void> {
+  const upper = new TransformStream<string, string>({
+    transform(chunk, controller) {
+      controller.enqueue(chunk.toUpperCase());
+    },
+    flush(controller) {
+      controller.enqueue("!");
+    },
+  });
+  console.log("pipeThrough", (await drain(source(["ab", "cd"]).pipeThrough(upper))).join(""));
+
+  const lengths = new TransformStream<string, number>({
+    transform(chunk, controller) {
+      controller.enqueue(chunk.length);
+    },
+  });
+  console.log("lengths", (await drain(source(["a", "bbb", "cc"]).pipeThrough(lengths))).join(" "));
+
+  const identity = new TransformStream<string, string>();
+  console.log("identity", (await drain(source(["i", "j"]).pipeThrough(identity))).join(""));
+
+  const body = new Response("fetch body").body!;
+  const decoded = body.pipeThrough(
+    new TransformStream<Uint8Array, string>({
+      transform(chunk, controller) {
+        controller.enqueue(new TextDecoder().decode(chunk));
+      },
+    }),
+  );
+  console.log("response body", (await drain(decoded)).join(""));
+
+  const [first, second] = new Response("split").body!.tee();
+  const decoder = new TextDecoder();
+  const text = async (s: ReadableStream<Uint8Array>): Promise<string> =>
+    (await drain(s)).map((chunk) => decoder.decode(chunk)).join("");
+  console.log("response tee", await text(first), await text(second));
+}
+
+await tees();
+await pipes();
+await transforms();

tests/fixtures/fetch/static-web-streams/main.mts

@@ -0,0 +1,96 @@
+// ReadableStream tee(), pipeTo() and pipeThrough() in a static build: a
+// fetched body (whole and chunked) tee'd, piped into a WritableStream and
+// through a TransformStream, constructed streams through a transformer,
+// and Node's TypeErrors for locked streams and errors through a pipe.
+const base = process.argv[2]!;
+const describe = (e: unknown): string => (e instanceof Error ? `${e.name}: ${e.message}` : String(e));
+const decoder = new TextDecoder();
+
+async function collect(stream: ReadableStream<Uint8Array>): Promise<string> {
+  const reader = stream.getReader();
+  let text = "";
+  for (;;) {
+    const result = await reader.read();
+    if (result.done) break;
+    text += decoder.decode(result.value, { stream: true });
+  }
+  return text + decoder.decode();
+}
+
+const fetched = await fetch(`${base}/text`);
+const [left, right] = fetched.body!.tee();
+console.log("tee", await collect(left), await collect(right), fetched.bodyUsed);
+
+const chunked = await fetch(`${base}/chunked`);
+const parts: string[] = [];
+await chunked.body!.pipeTo(
+  new WritableStream<Uint8Array>({
+    write(chunk) {
+      parts.push(decoder.decode(chunk, { stream: true }));
+    },
+    close() {
+      parts.push("<close>");
+    },
+  }),
+);
+console.log("pipeTo", parts.join("").length > 0, parts[parts.length - 1], chunked.bodyUsed);
+
+const upper = await fetch(`${base}/text`);
+const shouted = upper.body!.pipeThrough(
+  new TransformStream<Uint8Array, Uint8Array>({
+    transform(chunk, controller) {
+      controller.enqueue(new TextEncoder().encode(decoder.decode(chunk).toUpperCase()));
+    },
+  }),
+);
+console.log("pipeThrough", await collect(shouted));
+
+const words = new ReadableStream<string>({
+  start(controller) {
+    controller.enqueue("alpha");
+    controller.enqueue("beta");
+    controller.close();
+  },
+});
+const counted = words.pipeThrough(
+  new TransformStream<string, number>({
+    transform(chunk, controller) {
+      controller.enqueue(chunk.length);
+    },
+    flush(controller) {
+      controller.enqueue(-1);
+    },
+  }),
+);
+const counts: number[] = [];
+const reader = counted.getReader();
+for (;;) {
+  const result = await reader.read();
+  if (result.done) break;
+  counts.push(result.value);
+}
+console.log("constructed", counts.join(" "));
+
+const locked = await fetch(`${base}/text`);
+locked.body!.getReader();
+try {
+  locked.body!.tee();
+} catch (e) {
+  console.log("tee locked", describe(e));
+}
+try {
+  await locked.body!.pipeTo(new WritableStream<Uint8Array>());
+} catch (e) {
+  console.log("pipeTo locked", describe(e));
+}
+
+const failing = new ReadableStream<string>({
+  start(controller) {
+    controller.error(new TypeError("source failed"));
+  },
+});
+try {
+  await failing.pipeTo(new WritableStream<string>());
+} catch (e) {
+  console.log("pipeTo error", describe(e));
+}

tests/harness/__snapshots__/node-types-fenced.txt

@@ -142,41 +142,23 @@ fenced.ts:80:3 - error SC2020: 'RequestInit option 'integrity' in a static build
 
   hint: the compiler does not preserve this RequestInit member's conversion or transport behavior
 
-fenced.ts:84:3 - error SC2020: 'ReadableStream.tee' is typed by @types/node but has no scriptc lowering yet
-
-  83 |   const body = (await fetch(url)).body!;
-  84 |   body.tee();
-     |   ^~~~~~~~
-  85 |   const tee = body.tee;
-
-  hint: this operation is outside the native readable-stream implementation
-
-fenced.ts:85:15 - error SC2020: 'ReadableStream.tee' is typed by @types/node but has no scriptc lowering yet
+fenced.ts:85:15 - error SC2020: 'ReadableStream.tee in a static build' is typed by @types/node but has no scriptc lowering yet
 
   84 |   body.tee();
   85 |   const tee = body.tee;
      |               ^~~~~~~~
   86 |   body["tee"]();
 
-  hint: this operation is outside the native readable-stream implementation
+  hint: ReadableStream's locked compiles as a read, and getReader(), cancel(), tee(), pipeTo() and pipeThrough() as direct calls; values()/[Symbol.asyncIterator]() compile when consumed directly by for await, while method values and stored iterator handles remain unsupported
 
-fenced.ts:86:3 - error SC2020: 'ReadableStream.tee' is typed by @types/node but has no scriptc lowering yet
-
-  85 |   const tee = body.tee;
-  86 |   body["tee"]();
-     |   ^~~~~~~~~~~
-  87 |   const bracketTee = body["tee"];
-
-  hint: this operation is outside the native readable-stream implementation
-
-fenced.ts:87:22 - error SC2020: 'ReadableStream.tee' is typed by @types/node but has no scriptc lowering yet
+fenced.ts:87:22 - error SC2020: 'ReadableStream.tee in a static build' is typed by @types/node but has no scriptc lowering yet
 
   86 |   body["tee"]();
   87 |   const bracketTee = body["tee"];
      |                      ^~~~~~~~~~~
   88 | }
 
-  hint: this operation is outside the native readable-stream implementation
+  hint: ReadableStream's locked compiles as a read, and getReader(), cancel(), tee(), pipeTo() and pipeThrough() as direct calls; values()/[Symbol.asyncIterator]() compile when consumed directly by for await, while method values and stored iterator handles remain unsupported
 
 fenced.ts:93:23 - error SC2020: 'util.callbackify into a callback whose error parameter excludes null' is part of the standard library types but has no scriptc lowering yet
 

tests/harness/corpus-costs/web-streams-typed-element-tee.ts.json

@@ -0,0 +1 @@
+{ "plain": 3000, "san": 6000 }

tests/harness/corpus-costs/web-streams-typed-pipes.ts.json

@@ -0,0 +1 @@
+{ "plain": 3000, "san": 6000 }

tests/harness/fetch.test.ts

@@ -165,6 +165,7 @@ describe(`static fetch differential${sanitize ? " (sanitized)" : ""}`, () => {
     "static-dispatch-throw",
     "static-response-json",
     "static-clone",
+    "static-web-streams",
   ] as const;
   test.for(
     staticCases.flatMap((name) => (["llvm"] as const).map((backend) => [name, backend] as const)),

tests/harness/surface-manifest.test.ts

@@ -450,7 +450,7 @@ test("unimplemented Response and ReadableStream calls stay fenced", () => {
       "  void value.blob(); void value.formData();\n" +
       "}\n" +
       "function stream(value: ReadableStream<Uint8Array>): void {\n" +
-      "  void value.tee(); void value.pipeTo(null as never);\n" +
+      "  void value.values(); void value[Symbol.asyncIterator]();\n" +
       "}\n" +
       "void response; void stream;\n",
   });