Give JavaScript fetch bodies the members of a web ReadableStream (#2459)

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

@@ -65,4 +65,7 @@
   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 `Response` or `Request` body has the members of a `ReadableStream`, including `tee`, `pipeTo`, `pipeThrough`, `values`, and `for await`, and its lock and `bodyUsed` follow them.
+  After one of those members is used, a body given to a new `Response` or `Request` is locked from the start, as a JavaScript `ReadableStream` is.
+  In a program that names none of the web stream classes, those members are available when the program's JavaScript names one of them or iterates a body with `for await`; reached only through a computed key, they are not.
   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.

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

@@ -1274,7 +1274,7 @@ export function emitLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlValue {
     case "webstream":
       // A stream's start, pull and cancel steps run as promise jobs, so
       // a program with the classes runs its event loop.
-      if (e.fn === "webstream.class") host.usesTimers = true;
+      if (e.fn === "webstream.class" || e.fn === "webstream.install") host.usesTimers = true;
       return host.emitGenericLibCall(e);
     case "test":
       return emitTestLibCall(host, e);

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

@@ -120,6 +120,7 @@ export const LIB_FN_SYMS: Record<string, string> = {
   "util.throttle": "scr_util_throttle",
   "webstream.class": "scr_webstream_class",
   "webstream.instanceOf": "scr_webstream_instanceof",
+  "webstream.install": "scr_webstream_install",
   "mime.typeNew": "scr_mime_type_new",
   "mime.typeParse": "scr_mime_type_parse",
   "mime.paramsNew": "scr_mime_params_new",

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

@@ -938,6 +938,9 @@ export interface StaticFetchMembers {
   headersCalls: ReadonlySet<string>;
   readableStreamReads: ReadonlySet<string>;
   readableStreamCalls: ReadonlySet<string>;
+  /** ReadableStream prototype members the native stream does not
+   * implement, symbol-keyed ones as `[Symbol.name]`. */
+  readableStreamUnsupported: ReadonlySet<string>;
 }
 
 let staticFetchMemberSets: StaticFetchMembers | undefined;
@@ -961,6 +964,16 @@ export function staticFetchMembers(): StaticFetchMembers {
     ),
     readableStreamReads: new Set(profile.members.readableStreamReads),
     readableStreamCalls: new Set(profile.members.readableStreamCalls),
+    readableStreamUnsupported: new Set(
+      profile.inventory.entries
+        .filter(
+          (entry) =>
+            entry.owner === "ReadableStream" &&
+            (entry.placement === "prototype" || entry.placement === "prototype-symbol") &&
+            entry.status === "unsupported",
+        )
+        .map((entry) => entry.member),
+    ),
   };
   return staticFetchMemberSets;
 }

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

@@ -1,9 +1,19 @@
+import { staticFetchMembers } from "../../../compat/fetch-profile.js";
 import { numLit } from "../../../ir/build.js";
-import { BOOL, DYN, STRING, type IrExpr, type SrcLoc } from "../../../ir/ir.js";
+import { everyStmt } from "../../../ir/traverse.js";
+import {
+  BOOL,
+  DYN,
+  STRING,
+  VOID,
+  type IrExpr,
+  type IrFunction,
+  type SrcLoc,
+} from "../../../ir/ir.js";
 import * as ts from "../../ts7/adapter.js";
 import { isJsSourceFile, locOf } from "../../program.js";
 import { constituentTypes } from "../../ts7/checker.js";
-import type { Lowerer } from "../lowerer.js";
+import { ENTRY_NAME, type Lowerer } from "../lowerer.js";
 import { stdlibGlobalNameOf } from "../surfaces.js";
 import { streamWebImportOf } from "./module-bindings.js";
 
@@ -221,3 +231,98 @@ export function isWebStreamValueInJavaScript(lowerer: Lowerer, node: ts.Expressi
     return true;
   return typeInvolvesWebStream(lowerer, lowerer.typeOf(node));
 }
+
+/** Whether a checker type is the stdlib ReadableStream, or a union with it. */
+function isReadableStreamType(lowerer: Lowerer, type: ts.Type): boolean {
+  if ((type.flags & ts.TypeFlags.Union) !== 0)
+    return constituentTypes(type).some((t) => isReadableStreamType(lowerer, t));
+  const symbol = type.getSymbol();
+  return (
+    symbol !== undefined &&
+    ts.symbolName(symbol) === "ReadableStream" &&
+    lowerer.isStdlibSymbol(symbol)
+  );
+}
+
+/** 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);
+  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;
+  return owner !== undefined && owner.name.text === "ReadableStream";
+}
+
+/** Whether a checker type is any or unknown, which a fetch body can be. */
+function isUntypedType(type: ts.Type): boolean {
+  return (type.flags & (ts.TypeFlags.Any | ts.TypeFlags.Unknown)) !== 0;
+}
+
+/** Whether the program links the fetch unit (a fetch. entry point), whose
+ * Response and Request bodies can reach JavaScript as untyped values. */
+function programLinksFetch(functions: IrFunction[]): boolean {
+  let found = false;
+  const visitor = {
+    expr: (e: IrExpr): boolean => {
+      if (e.kind === "libCall" && e.fn.startsWith("fetch.")) found = true;
+      return !found;
+    },
+    stmt: (): boolean => !found,
+  };
+  for (const fn of functions) {
+    for (const s of fn.body) everyStmt(s, visitor);
+    if (found) return true;
+  }
+  return false;
+}
+
+/** Whether a JavaScript source of the program uses a ReadableStream member
+ * that the fetch unit's own stream does not implement (the fetch profile's
+ * readableStreamUnsupported): a member access the checker resolves to the
+ * interface, or `for await` over a ReadableStream value. The same member
+ * names on an untyped receiver, or `for await` over an untyped value,
+ * count in a program that links the fetch unit, since the value can be a
+ * fetch body. */
+function programUsesWebStreamOnlyMembers(lowerer: Lowerer, functions: IrFunction[]): boolean {
+  const members = staticFetchMembers().readableStreamUnsupported;
+  const iterates = members.has("[Symbol.asyncIterator]");
+  let untyped = false;
+  for (const file of lowerer.program.getSourceFiles()) {
+    if (file.isDeclarationFile || !isJsSourceFile(file)) continue;
+    let found = false;
+    ts.walkPreorder(file, (node) => {
+      let receiver: ts.Type | null = null;
+      if (ts.isPropertyAccessExpression(node) && members.has(node.name.text)) {
+        if (isReadableStreamMember(lowerer, node)) found = true;
+        else receiver = lowerer.typeOf(node.expression);
+      } else if (iterates && ts.isForOfStatement(node) && node.awaitModifier !== undefined) {
+        receiver = lowerer.typeOf(node.expression);
+        if (isReadableStreamType(lowerer, receiver)) found = true;
+      }
+      if (found) return "stop";
+      if (receiver !== null && isUntypedType(receiver)) untyped = true;
+      return undefined;
+    });
+    if (found) return true;
+  }
+  return untyped && programLinksFetch(functions);
+}
+
+/** 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
+ * 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. */
+export function installFetchBodyWebStreams(lowerer: Lowerer, functions: IrFunction[]): void {
+  if (programNamesWebStreamClass(lowerer) || !programUsesWebStreamOnlyMembers(lowerer, functions))
+    return;
+  const main = functions.find((fn) => fn.name === ENTRY_NAME);
+  if (!main) return;
+  const loc = main.loc;
+  main.body.unshift({
+    kind: "exprStmt",
+    expr: { kind: "libCall", fn: "webstream.install", args: [], type: VOID, loc },
+    loc,
+  });
+}

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

@@ -118,7 +118,11 @@ import { fenceWebCryptoMember, lowerWebCryptoCall } from "./builtins/webcrypto.j
 import { valueStreamOf } from "./builtins/hash-stream.js";
 import { lowerNetAddressCall, lowerNetAddressNew } from "./builtins/net-address.js";
 import { lowerMimeCall, lowerMimeNew } from "./builtins/mime.js";
-import { lowerWebStreamNew, lowerWebStreamStaticCall } from "./builtins/web-streams-dyn.js";
+import {
+  installFetchBodyWebStreams,
+  lowerWebStreamNew,
+  lowerWebStreamStaticCall,
+} from "./builtins/web-streams-dyn.js";
 import { dynUndefinedExpr, nodeThrowExpr, numLit, varRef } from "../../ir/build.js";
 import { iteratorSlotAdapter } from "./iterator-adapters.js";
 import { iterableSlotValue } from "./iterable-values.js";
@@ -9459,6 +9463,7 @@ export class Lowerer {
     guardFrozenWrites(this, functions);
     applyHandleProperties(this, functions);
     installBufferValue(functions);
+    installFetchBodyWebStreams(this, functions);
 
     // Globals typed by a class that never REGISTERED (a JS class whose
     // collection fenced — Symbol-keyed fields, an unsupported base): the

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

@@ -1211,6 +1211,7 @@ function buildLibFnSignatures() {
     "util.throttle": { argTypes: [DYN, DYN, DYN, DYN], result: DYN },
     "webstream.class": { argTypes: [F64], result: DYN },
     "webstream.instanceOf": { argTypes: [DYN, F64], result: BOOL },
+    "webstream.install": { argTypes: [], result: VOID },
     "mime.typeNew": { argTypes: [DYN, DYN], result: DYN },
     "mime.typeParse": { argTypes: [DYN], result: DYN },
     "mime.paramsNew": { argTypes: [], result: DYN },

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

@@ -83,6 +83,7 @@ const DYN_ASYNC_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   // Web streams settle their promises with dyn values.
   "webstream.class",
   "webstream.instanceOf",
+  "webstream.install",
   "dyn.promiseAll",
   "dyn.promiseAllSettled",
   "dyn.promiseAny",
@@ -390,6 +391,7 @@ const SYMBOL_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
   // Web streams answer async iteration through Symbol.asyncIterator.
   "webstream.class",
   "webstream.instanceOf",
+  "webstream.install",
   // EventEmitter's kShapeMode and kCapture properties
   ...BUILTIN_CLASS_LIB_FNS,
   "dyn.usingDispose",
@@ -954,6 +956,7 @@ const RUNTIME_FEATURE_RULES = {
       // scr_webstreams.c calls the iterators ReadableStream.from reads.
       "webstream.class",
       "webstream.instanceOf",
+      "webstream.install",
     ],
     // Constant-key reads of Object.prototype member names answer the
     // inherited member through scr_dyn_invoke.c.
@@ -1076,6 +1079,7 @@ const RUNTIME_FEATURE_RULES = {
       "global.native",
       // ERR_ARG_NOT_ITERABLE renders the value as util.format's %s does.
       "webstream.class",
+      "webstream.install",
       "util.styleText",
       "crypto.cipherNew",
       "webcrypto.encrypt",

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-helpers.mjs.json

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

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-linked.mjs.json

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

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-network.mjs.json

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

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-web-stream-unlinked.mjs.json

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

packages/runtime/src/scr_dyn.h

@@ -2084,6 +2084,16 @@ ScrDyn *scr_dyn_handle_class_prototype(ScrDynHandleTag tag);
 /* Handles of `tag` (another unit's native representation of the same
  * interface) are instances of the class of `klass_tag`. */
 void scr_dyn_handle_class_alias(ScrDynHandleTag tag, ScrDynHandleTag klass_tag);
+/* The instance of the class a handle of an alias tag stands for, which
+ * the class's methods and getters run on (+1, or NULL with the exception
+ * pending). Without one, they throw their ERR_INVALID_THIS TypeError on
+ * the alias's handles. */
+void scr_dyn_handle_class_alias_view(ScrDynHandleTag tag, ScrDyn *(*view)(ScrDyn *handle));
+/* The receiver a method or getter of the class of `klass_tag` runs on for
+ * `self`: true with *out set to self, or to the instance an alias handle
+ * stands for (+1, or NULL with the exception pending); false when self is
+ * neither. */
+bool scr_dyn_handle_class_receiver(ScrDynHandleTag klass_tag, ScrDyn *self, ScrDyn **out);
 /* The class tag a handle is an instance of, or SCR_DYNH_CAPACITY for a
  * handle of no class. */
 ScrDynHandleTag scr_dyn_handle_class_tag_of(const ScrDyn *handle);

packages/runtime/src/scr_dyn_handle_class.c

@@ -20,6 +20,8 @@ static SCR_TL const ScrDynHandleClass *hc_classes[SCR_DYNH_CAPACITY];
 /* The class tag of another tag's handles + 1 (scr_dyn_handle_class_alias);
  * 0 = none. */
 static SCR_TL uint16_t hc_aliases[SCR_DYNH_CAPACITY];
+/* scr_dyn_handle_class_alias_view, per alias tag. */
+static SCR_TL ScrDyn *(*hc_alias_views[SCR_DYNH_CAPACITY])(ScrDyn *handle);
 static SCR_TL ScrDyn *hc_ctors[SCR_DYNH_CAPACITY];
 static SCR_TL ScrDyn *hc_prototypes[SCR_DYNH_CAPACITY];
 static SCR_TL bool hc_teardown_registered;
@@ -73,9 +75,13 @@ static ScrDyn *hc_getter(ScrClosure *closure, ScrDyn *const *args, size_t argc)
   const ScrDynHandleOps *ops = hc_ops(tag);
   ScrDyn *self = scr_dyn_this_get();
   ScrDyn *result = NULL;
-  if (self->kind == SCR_DYN_HANDLE && self->v.handle.tag == tag) {
-    result = ops->get(self->v.handle.ptr, accessor->name, strlen(accessor->name));
-    if (!result && !scr_exc_pending()) result = scr_dyn_retain(scr_dyn_undefined());
+  ScrDyn *receiver = NULL;
+  if (scr_dyn_handle_class_receiver(tag, self, &receiver)) {
+    if (receiver) {
+      result = ops->get(receiver->v.handle.ptr, accessor->name, strlen(accessor->name));
+      if (!result && !scr_exc_pending()) result = scr_dyn_retain(scr_dyn_undefined());
+      scr_dyn_release(receiver);
+    }
   } else {
     hc_invalid_this(ops);
     if (accessor->rejects && ops->rejected) result = ops->rejected();
@@ -245,6 +251,23 @@ void scr_dyn_handle_class_alias(ScrDynHandleTag tag, ScrDynHandleTag klass_tag)
   hc_aliases[tag] = (uint16_t)(klass_tag + 1);
 }
 
+void scr_dyn_handle_class_alias_view(ScrDynHandleTag tag, ScrDyn *(*view)(ScrDyn *handle)) {
+  hc_alias_views[tag] = view;
+}
+
+bool scr_dyn_handle_class_receiver(ScrDynHandleTag klass_tag, ScrDyn *self, ScrDyn **out) {
+  *out = NULL;
+  if (self->kind != SCR_DYN_HANDLE) return false;
+  ScrDynHandleTag tag = self->v.handle.tag;
+  if (tag == klass_tag) {
+    *out = scr_dyn_retain(self);
+    return true;
+  }
+  if (hc_aliases[tag] != (uint16_t)(klass_tag + 1) || !hc_alias_views[tag]) return false;
+  *out = hc_alias_views[tag](self);
+  return true;
+}
+
 ScrDynHandleTag scr_dyn_handle_class_tag_of(const ScrDyn *handle) {
   ScrDynHandleTag tag = handle->v.handle.tag;
   if (hc_classes[tag]) return tag;

packages/runtime/src/scr_fetch.c

@@ -164,6 +164,20 @@ struct SfStream {
   /* Consuming the body takes Uint8Array chunks only: set on a body from a
    * web ReadableStream and on the tee branches of such a body. */
   bool u8_chunks;
+  /* The web ReadableStream this stream's checked-dynamic handles stand
+   * for once a member the fetch stream lacks is used (+1, or NULL; see
+   * sf_stream_web_view). It reads this stream; from then on every member
+   * and every use of the body go through it. */
+  ScrDyn *web_view;
+  /* A reader of web_view (+1) that keeps it locked while this stream was
+   * locked when web_view was made: by a reader of its own, released
+   * later, or by reading the body. */
+  ScrDyn *web_view_lock;
+  /* web_view's reader of this stream, taken at its first pull (owned
+   * through `reader`). */
+  SfReader *view_reader;
+  /* web_view's source holds a reference to this stream. */
+  bool view_source_live;
 #endif
   ScrArr *from_array;
   ScrBytes *from_bytes;
@@ -917,6 +931,18 @@ static void sf_stream_drop_source_callbacks(
 static void sf_stream_release(SfStream *s) {
   if (!s) return;
   if (--s->rc > 0) {
+#ifdef SCR_HANDLE_CLASSES
+    /* Only the web view's source still holds the stream: the view no
+     * longer needs to be found from it, and the view, which reads it,
+     * keeps it while anything else holds the view. */
+    if (s->rc == 1 && s->view_source_live && s->web_view) {
+      ScrDyn *view = s->web_view, *lock = s->web_view_lock;
+      s->web_view = s->web_view_lock = NULL;
+      scr_dyn_release(lock);
+      scr_dyn_release(view); /* may release s */
+      return;
+    }
+#endif
     sf_response_discard_if_unobserved(s);
     return;
   }
@@ -933,6 +959,8 @@ static void sf_stream_release(SfStream *s) {
   scr_dyn_release(s->from_dyn);
 #ifdef SCR_HANDLE_CLASSES
   scr_dyn_release(s->web_reader);
+  scr_dyn_release(s->web_view_lock);
+  scr_dyn_release(s->web_view);
 #endif
   scr_arr_release(s->from_array);
   scr_bytes_release(s->from_bytes);
@@ -1629,6 +1657,97 @@ static ScrDyn *sf_web_body(const ScrDyn *body) {
 static bool sf_is_web_body(const ScrDyn *body) {
   return body && body->kind == SCR_DYN_HANDLE && body->v.handle.tag == SCR_DYNH_WS_READABLE;
 }
+
+static SfReader *sf_stream_get_reader(SfStream *s);
+static bool sf_reader_read_request(SfReader *r, ScrPromise *p, const ScrFetchReadSink *sink, void *sink_ctx);
+static ScrPromise *sf_stream_cancel(SfStream *s, ScrDyn *reason, bool through_reader);
+
+/* The web view's source (scr_webstream_readable_over): it reads the stream
+ * through a reader of its own, taken at the first pull. */
+static bool sf_view_read_into(void *ctx, const ScrFetchReadSink *sink, void *sink_ctx) {
+  SfStream *s = ctx;
+  if (!s->view_reader) {
+    SfReader *r = sf_stream_get_reader(s);
+    if (!r) return false;
+    s->view_reader = r; /* the stream owns it as its reader */
+    sf_reader_release(r);
+  }
+  /* A chunk the request takes at once can close the view, whose source
+   * then lets the stream go while it still drains. */
+  sf_stream_retain(s);
+  bool ok = sf_reader_read_request(s->view_reader, NULL, sink, sink_ctx);
+  sf_stream_release(s);
+  return ok;
+}
+
+static ScrPromise *sf_view_cancel(void *ctx, ScrDyn *reason) {
+  SfStream *s = ctx;
+  return sf_stream_cancel(s, reason, s->view_reader != NULL && s->reader == s->view_reader);
+}
+
+static void sf_view_release(void *ctx) {
+  SfStream *s = ctx;
+  s->view_source_live = false;
+  sf_stream_release(s);
+}
+
+static const ScrWebStreamSource sf_view_source = {sf_view_read_into, sf_view_cancel, sf_view_release};
+
+/* The web ReadableStream the stream's checked-dynamic handles stand for
+ * (borrowed), made on first use. Node's Response.body is one; the fetch
+ * stream answers getReader, cancel and locked itself until a member it
+ * lacks is used, and a lock it holds then keeps the view locked. */
+static ScrDyn *sf_stream_web_view(SfStream *s) {
+  if (s->web_view) return s->web_view;
+  s->web_view = scr_webstream_readable_over(&sf_view_source, sf_stream_retain(s));
+  s->view_source_live = true;
+  if (s->reader || s->internal_lock) {
+    s->web_view_lock = scr_dyn_invoke(s->web_view, "getReader", NULL, 0, "ReadableStream.getReader");
+    if (!s->web_view_lock) return NULL;
+  }
+  return s->web_view;
+}
+
+/* Whether `key` names one of the fetch stream's own methods (its ops'
+ * method table). */
+static bool sf_stream_own_method(const char *key, size_t len) {
+  ScrDyn probe;
+  memset(&probe, 0, sizeof probe);
+  probe.kind = SCR_DYN_HANDLE;
+  probe.v.handle.tag = SCR_DYNH_WEB_STREAM;
+  for (const ScrDynHandleMethod *m = scr_dyn_handle_ops_of(&probe)->methods; m && m->name; m++)
+    if (sf_name(key, len, m->name)) return true;
+  return false;
+}
+
+/* scr_dyn_handle_class_alias_view for the stream's handles (+1). */
+static ScrDyn *sf_stream_view_of(ScrDyn *handle) {
+  ScrDyn *view = sf_stream_web_view((SfStream *)handle->v.handle.ptr);
+  return view ? scr_dyn_retain(view) : NULL;
+}
+
+/* A body whose stream has a web view is read through the view, which
+ * locks it as reading a body locks Node's stream: a native stream that
+ * reads the view (+1), or NULL with the TypeError pending. Otherwise the
+ * stream itself (+1). */
+static SfStream *sf_stream_body_source(SfStream *s) {
+  if (!s->web_view) return sf_stream_retain(s);
+  ScrDyn *reading = sf_web_body(s->web_view);
+  if (!reading) return NULL;
+  SfStream *out = sf_stream_retain((SfStream *)reading->v.handle.ptr);
+  scr_dyn_release(reading);
+  return out;
+}
+#endif
+
+/* Whether a body's web view is locked or disturbed, which makes the body
+ * unusable as one, as its own lock and disturbed state do. */
+#ifdef SCR_HANDLE_CLASSES
+static bool sf_view_unusable(const SfStream *s) {
+  return s->web_view && scr_web_stream_bridge->unusable(s->web_view);
+}
+#else
+#define sf_view_unusable(s) false
 #endif
 
 static void sf_stream_pull_from_source(SfStream *s) {
@@ -1882,10 +2001,22 @@ static void sf_reader_release_lock(SfReader *r) {
 
   sf_reader_reject_message_all(r, "Invalid state: Releasing reader");
 
+#ifdef SCR_HANDLE_CLASSES
+  /* The web view was made while this reader held the lock. */
+  ScrDyn *view_lock = r != s->view_reader ? s->web_view_lock : NULL;
+  if (view_lock) s->web_view_lock = NULL;
+#endif
   size_t handle_refs = r->handle_refs;
   s->reader = NULL;
   r->stream = NULL;
   sf_reader_release(r); /* stream's ownership */
+#ifdef SCR_HANDLE_CLASSES
+  if (view_lock) {
+    ScrDyn *released = scr_dyn_invoke(view_lock, "releaseLock", NULL, 0, "ReadableStreamDefaultReader.releaseLock");
+    if (released) scr_dyn_release(released);
+    scr_dyn_release(view_lock);
+  }
+#endif
   while (handle_refs-- > 0) sf_stream_release(s);
 }
 
@@ -1999,8 +2130,21 @@ static ScrPromise *sf_stream_cancel(SfStream *s, ScrDyn *reason,
 }
 
 static ScrPromise *sf_stream_collect(SfStream *s, int mode) {
+#ifdef SCR_HANDLE_CLASSES
+  if (s->web_view && !(s->disturbed || s->reader || s->internal_lock || sf_view_unusable(s))) {
+    SfStream *source = sf_stream_body_source(s);
+    if (!source) {
+      ScrPromise *p = scr_promise_new();
+      scr_promise_reject_pending(p);
+      return p;
+    }
+    ScrPromise *p = sf_stream_collect(source, mode);
+    sf_stream_release(source);
+    return p;
+  }
+#endif
   ScrPromise *p = scr_promise_new();
-  if (s->disturbed || s->reader || s->internal_lock) {
+  if (s->disturbed || s->reader || s->internal_lock || sf_view_unusable(s)) {
     sf_type_error("Body is unusable: Body has already been read");
     scr_promise_reject_pending(p);
     return p;
@@ -2456,6 +2600,14 @@ static ScrDyn *sf_stream_invoke(void *ptr, ScrDyn *self, const char *method,
                                 ScrDyn *const *args, size_t argc,
                                 const char *what) {
   SfStream *s = ptr;
+#ifdef SCR_HANDLE_CLASSES
+  /* Any other member, and every member once the web view exists, is the
+   * view's. */
+  if (s->web_view || !sf_stream_own_method(method, strlen(method))) {
+    ScrDyn *view = sf_stream_web_view(s);
+    return view ? scr_dyn_invoke(view, method, args, argc, what) : NULL;
+  }
+#endif
   if (strcmp(method, "getReader") == 0) {
     ScrDyn *options = argc ? args[0] : NULL;
     ScrDyn *materialized = NULL;
@@ -2494,12 +2646,35 @@ static ScrDyn *sf_stream_invoke(void *ptr, ScrDyn *self, const char *method,
 
 static ScrDyn *sf_stream_get(void *ptr, const char *key, size_t len) {
   SfStream *s = ptr;
+#ifdef SCR_HANDLE_CLASSES
+  /* The stream's own members are its own until the web view exists; any
+   * other member makes the view and is the view's. */
+  if (s->web_view || !(sf_name(key, len, "locked") || sf_stream_own_method(key, len))) {
+    ScrDyn *view = sf_stream_web_view(s);
+    if (!view) return NULL;
+    ScrStr *name = scr_str_new(key, len);
+    ScrDyn *value = scr_dyn_handle_key_get(view, name);
+    scr_str_release(name);
+    return value;
+  }
+#endif
   if (sf_name(key, len, "locked")) {
     return scr_dyn_new_bool(s->reader != NULL || s->internal_lock);
   }
   return NULL;
 }
 
+#ifdef SCR_HANDLE_CLASSES
+/* Symbol.asyncIterator is the web view's. */
+static ScrDyn *sf_stream_symbol_get(void *ptr, const ScrDyn *key) {
+  if (!scr_dyn_is_async_iterator_symbol(key)) return NULL;
+  ScrDyn *view = sf_stream_web_view(ptr);
+  if (!view) return NULL;
+  const ScrDynHandleOps *ops = scr_dyn_handle_ops_of(view);
+  return ops->symbol_get ? ops->symbol_get(view->v.handle.ptr, key) : NULL;
+}
+#endif
+
 static ScrDyn *sf_reader_invoke(void *ptr, ScrDyn *self, const char *method,
                                 ScrDyn *const *args, size_t argc,
                                 const char *what) {
@@ -4098,10 +4273,30 @@ static ScrPromise *sf_response_collect(SfResponse *r, int mode) {
  * guard. NULL with a TypeError pending when the body is unusable. */
 static SfResponse *sf_response_clone(SfResponse *r, const char *unusable) {
   if (!r->null_body &&
-      (r->body->disturbed || r->body->reader || r->body->internal_lock)) {
+      (r->body->disturbed || r->body->reader || r->body->internal_lock || sf_view_unusable(r->body))) {
     sf_type_error(unusable);
     return NULL;
   }
+#ifdef SCR_HANDLE_CLASSES
+  /* A body with a web view tees through it; each side reads one branch. */
+  SfStream *viewed[2] = {NULL, NULL};
+  if (!r->null_body && r->body->web_view) {
+    ScrDyn *branches = scr_dyn_invoke(r->body->web_view, "tee", NULL, 0, "ReadableStream.tee");
+    if (!branches) return NULL;
+    for (int i = 0; i < 2; i++) {
+      ScrDyn *reading = sf_web_body(branches->v.arr.items[i]);
+      if (!reading) {
+        sf_stream_release(viewed[0]);
+        scr_dyn_release(branches);
+        return NULL;
+      }
+      viewed[i] = sf_stream_retain((SfStream *)reading->v.handle.ptr);
+      viewed[i]->u8_chunks = r->body->u8_chunks;
+      scr_dyn_release(reading);
+    }
+    scr_dyn_release(branches);
+  }
+#endif
   SfResponse *c = calloc(1, sizeof *c);
   if (!c) sf_oom();
   c->rc = 1;
@@ -4110,7 +4305,16 @@ static SfResponse *sf_response_clone(SfResponse *r, const char *unusable) {
     sf_stream_close(c->body);
   } else {
     SfStream *branches[2];
+#ifdef SCR_HANDLE_CLASSES
+    if (viewed[0]) {
+      branches[0] = viewed[0];
+      branches[1] = viewed[1];
+    } else {
+      sf_stream_tee(r->body, branches);
+    }
+#else
     sf_stream_tee(r->body, branches);
+#endif
     sf_stream_release(r->body);
     r->body = branches[0];
     c->body = branches[1];
@@ -4177,6 +4381,12 @@ static ScrDyn *sf_response_get(void *ptr, const char *key, size_t len) {
       sf_name(key, len, "statusText"))
     sf_response_materialize(r);
   if (sf_name(key, len, "bodyUsed")) {
+#ifdef SCR_HANDLE_CLASSES
+    /* A read, cancel or pipe started on the web view disturbs it at once,
+     * before its pull reaches this stream. */
+    if (!r->null_body && r->body->web_view && scr_web_stream_bridge->disturbed(r->body->web_view))
+      return scr_dyn_new_bool(true);
+#endif
     return scr_dyn_new_bool(!r->null_body && r->body->disturbed);
   }
   if (sf_name(key, len, "body")) {
@@ -6153,13 +6363,28 @@ static bool sf_start_hop(SfTransfer *t) {
       stream = sf_stream_of(body);
     }
     if (!stream || stream->reader || stream->internal_lock ||
-        stream->disturbed) {
+        stream->disturbed || sf_view_unusable(stream)) {
       scr_dyn_release(seeded);
       if (t->client) scr_http_client_destroy(t->client);
       sf_reject(t,
                 "Response body object should not be disturbed or locked");
       return false;
     }
+#ifdef SCR_HANDLE_CLASSES
+    /* A stream with a web view uploads through it; the transfer holds the
+     * stream that reads the view. */
+    SfStream *through = NULL;
+    if (stream->web_view) {
+      through = sf_stream_body_source(stream);
+      if (!through) {
+        scr_dyn_release(seeded);
+        if (t->client) scr_http_client_destroy(t->client);
+        sf_reject(t, "Response body object should not be disturbed or locked");
+        return false;
+      }
+      stream = through;
+    }
+#endif
     stream->disturbed = true;
     stream->internal_lock = true;
     stream->request_owner = t;
@@ -6169,6 +6394,9 @@ static bool sf_start_hop(SfTransfer *t) {
         !stream->error && !t->done && t->client) {
       sf_stream_pull(stream);
     }
+#ifdef SCR_HANDLE_CLASSES
+    sf_stream_release(through);
+#endif
     scr_dyn_release(seeded);
   } else {
     scr_http_client_end(client);
@@ -6429,12 +6657,17 @@ static ScrDyn *sf_response_construct(ScrDyn *body, ScrDyn *init,
   } else if (body->kind == SCR_DYN_HANDLE &&
              body->v.handle.tag == SCR_DYNH_WEB_STREAM) {
     stream = (SfStream *)body->v.handle.ptr;
-    if (stream->reader || stream->internal_lock || stream->disturbed) {
+    if (stream->reader || stream->internal_lock || stream->disturbed || sf_view_unusable(stream)) {
       stream = NULL; /* borrowed until the successful retain below */
       sf_type_error("Response body object should not be disturbed or locked");
       goto fail;
     }
+#ifdef SCR_HANDLE_CLASSES
+    stream = sf_stream_body_source(stream);
+    if (!stream) goto fail;
+#else
     sf_stream_retain(stream);
+#endif
 #ifdef SCR_HANDLE_CLASSES
   } else if (sf_is_web_body(body)) {
     ScrDyn *web = sf_web_body(body);
@@ -6676,7 +6909,8 @@ ScrDyn *scr_fetch_request_new(ScrDyn *input, ScrDyn *init) {
   ScrDyn *body = sf_response_init_get(init, "body", 4);
   bool supplied = body && body->kind != SCR_DYN_UNDEF && body->kind != SCR_DYN_NULL;
   if (!supplied && source && !source->payload->null_body) {
-    if (source->payload->body->disturbed || source->payload->body->reader || source->payload->body->internal_lock) {
+    if (source->payload->body->disturbed || source->payload->body->reader || source->payload->body->internal_lock ||
+        sf_view_unusable(source->payload->body)) {
       scr_dyn_release(body); scr_str_release(method_text); scr_url_release(url);
       sf_type_error("Cannot construct a Request with a Request object that has already been used."); return NULL;
     }
@@ -7195,6 +7429,9 @@ static const ScrDynHandleOps sf_abort_controller_ops = {
 static const ScrDynHandleOps sf_stream_ops = {
     "ReadableStream", &sf_stream_retain_v, &sf_stream_release_v,
     &sf_stream_invoke, &sf_stream_get, &sf_no_set, NULL,
+#ifdef SCR_HANDLE_CLASSES
+    .symbol_get = &sf_stream_symbol_get,
+#endif
     .methods = sf_stream_methods,
     .invalid_this = "Value of \"this\" must be of type ReadableStream",
     .rejected = &sf_rejected};
@@ -7273,6 +7510,9 @@ void scr_fetch_install(void) {
   scr_dyn_handle_install(SCR_DYNH_ABORT_CONTROLLER,
                          &sf_abort_controller_ops);
   scr_dyn_handle_install(SCR_DYNH_WEB_STREAM, &sf_stream_ops);
+#ifdef SCR_HANDLE_CLASSES
+  scr_dyn_handle_class_alias_view(SCR_DYNH_WEB_STREAM, &sf_stream_view_of);
+#endif
   scr_dyn_handle_install(SCR_DYNH_WEB_READER, &sf_reader_ops);
   scr_dyn_handle_install(SCR_DYNH_WEB_CONTROLLER, &sf_controller_ops);
   scr_dyn_handle_install(SCR_DYNH_FETCH_REQUEST, &sf_request_ops);

packages/runtime/src/scr_generator.h

@@ -231,8 +231,29 @@ typedef struct ScrWebStreamBridge {
   /* Whether a ReadableStream is locked or disturbed, which makes it
    * unusable as a body. */
   bool (*unusable)(const ScrDyn *stream);
+  /* Whether a ReadableStream is disturbed: a read, a cancel or a pipe
+   * has started on it. */
+  bool (*disturbed)(const ScrDyn *stream);
 } ScrWebStreamBridge;
 extern SCR_TL const ScrWebStreamBridge *scr_web_stream_bridge;
+/* The read side of a stream another unit implements, which
+ * scr_webstream_readable_over reads: read_into queues one read request on
+ * it (false with the exception pending), cancel cancels it (+1 promise),
+ * and release drops the context. */
+typedef struct ScrWebStreamSource {
+  bool (*read_into)(void *ctx, const ScrFetchReadSink *sink, void *sink_ctx);
+  ScrPromise *(*cancel)(void *ctx, ScrDyn *reason);
+  void (*release)(void *ctx);
+} ScrWebStreamSource;
+/* A checked-dynamic ReadableStream (+1 handle) with a zero high-water mark
+ * that reads one chunk of `source` per pull (scr_webstreams.c). ctx's
+ * reference moves in; the stream releases it when it goes away. */
+ScrDyn *scr_webstream_readable_over(const ScrWebStreamSource *source, void *ctx);
+/* Installs the web stream classes (webstream.install): a program whose
+ * JavaScript uses a ReadableStream member the fetch unit's streams lack
+ * calls it at startup, which links the classes that fetch bodies then
+ * stand for. */
+void scr_webstream_install(void);
 /* The web stream class constructors for checked-dynamic code, by kind
  * (scr_webstreams.c: ReadableStream, ReadableStreamDefaultReader,
  * ReadableStreamDefaultController, CountQueuingStrategy,

packages/runtime/src/scr_json.c

@@ -7015,16 +7015,20 @@ static ScrDyn *scr_dynh_method_call(ScrClosure *closure, ScrDyn *const *args, si
   const ScrDynHandleOps *ops = scr_dyn_handle_ops(tag);
   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
+  /* A class's method value reaches the class's method past the handle's
+   * own properties, and runs on the instance a handle of an alias tag
+   * stands for; this build is the one programs with such classes link. */
+  ScrDyn *receiver = NULL;
+  if (scr_dyn_handle_class_receiver(tag, self, &receiver)) {
+    result = receiver ? scr_dyn_handle_class_method_call(ops, receiver, method->name, args, argc) : NULL;
+    scr_dyn_release(receiver);
   } else {
+#else
+  if (self->kind == SCR_DYN_HANDLE && self->v.handle.tag == tag) {
+    result = ops->invoke(self->v.handle.ptr, self, method->name, args, argc, method->name);
+  } else {
+#endif
     const char *message = method->invalid_this ? method->invalid_this
                           : ops->invalid_this  ? ops->invalid_this
                                                : "Illegal invocation";

packages/runtime/src/scr_webstreams.c

@@ -1691,6 +1691,96 @@ static ScrDyn *ws_readable_from_thunk(ScrClosure *closure, ScrDyn *const *args,
   return ws_readable_from(argc > 0 ? args[0] : scr_dyn_undefined());
 }
 
+/* ── streams over another unit's reads (scr_webstream_readable_over) ─ */
+
+typedef struct WsOver {
+  WsObj obj;
+  WsReadable *stream; /* +1 */
+  const ScrWebStreamSource *source;
+  void *ctx; /* released through source->release */
+} WsOver;
+
+static void ws_over_trace(void *obj, ScrTraceVisit visit, void *ctx) { ws_visit(((WsOver *)obj)->stream, visit, ctx); }
+
+static void ws_over_drop(void *obj, bool gc) {
+  WsOver *o = obj;
+  ws_drop_obj(o->stream, gc);
+  o->stream = NULL;
+  /* The context is the other unit's, outside the collector. */
+  void *ctx = o->ctx;
+  o->ctx = NULL;
+  if (ctx) o->source->release(ctx);
+}
+
+static const WsObjType ws_over_type = {ws_over_trace, ws_over_drop};
+
+/* The stream's controller while it still holds this source. */
+static WsRController *ws_over_controller(WsOver *o) {
+  WsRController *c = o->stream ? o->stream->controller : NULL;
+  return c && !c->algorithms_cleared ? c : NULL;
+}
+
+static void ws_over_chunk(void *ctx, ScrDyn *value) {
+  WsOver *o = ctx;
+  WsRController *c = ws_over_controller(o);
+  if (c) {
+    if (ws_rcontroller_enqueue(c, value)) {
+      ws_pull_fulfilled(c, NULL);
+    } else {
+      ScrDyn *error = ws_take_pending();
+      ws_rcontroller_error(c, error);
+      scr_dyn_release(error);
+    }
+  }
+  ws_release(o);
+}
+
+static void ws_over_close(void *ctx) {
+  WsOver *o = ctx;
+  WsRController *c = ws_over_controller(o);
+  if (c) ws_rcontroller_close(c);
+  ws_release(o);
+}
+
+static void ws_over_error(void *ctx, ScrDyn *reason) {
+  WsOver *o = ctx;
+  WsRController *c = ws_over_controller(o);
+  if (c) ws_rcontroller_error(c, reason ? reason : scr_dyn_undefined());
+  ws_release(o);
+}
+
+static const ScrFetchReadSink ws_over_sink = {ws_over_chunk, ws_over_close, ws_over_error};
+
+static ScrDyn *ws_over_pull(void *ctx, WsRController *c) {
+  (void)c;
+  WsOver *o = ctx;
+  if (!o->source->read_into(o->ctx, &ws_over_sink, ws_retain(o))) {
+    ws_release(o);
+    return NULL;
+  }
+  return WS_PARKED;
+}
+
+static ScrPromise *ws_over_cancel(void *ctx, WsRController *c, ScrDyn *reason) {
+  (void)c;
+  WsOver *o = ctx;
+  return o->source->cancel(o->ctx, reason);
+}
+
+static const WsSourceOps ws_over_source_ops = {ws_over_pull, ws_over_cancel};
+
+ScrDyn *scr_webstream_readable_over(const ScrWebStreamSource *source, void *ctx) {
+  ws_install();
+  WsOver *o = ws_obj_new(sizeof *o, &ws_over_type);
+  o->source = source;
+  o->ctx = ctx;
+  WsReadable *s = ws_readable_create_native(&ws_over_source_ops, o, 0, 0);
+  o->stream = ws_readable_retain(s);
+  ScrDyn *box = ws_stream_box(s);
+  ws_readable_release(s);
+  return box;
+}
+
 /* ── async iteration (ReadableStream.prototype.values) ────────────── */
 
 typedef struct WsIter {
@@ -2047,7 +2137,12 @@ static bool ws_readable_unusable(const ScrDyn *stream) {
   return !s || ws_readable_locked(s) || s->disturbed;
 }
 
-static const ScrWebStreamBridge ws_fetch_bridge = {ws_reader_read_into, ws_readable_unusable};
+static bool ws_readable_disturbed(const ScrDyn *stream) {
+  WsReadable *s = ws_stream_of(stream);
+  return s && s->disturbed;
+}
+
+static const ScrWebStreamBridge ws_fetch_bridge = {ws_reader_read_into, ws_readable_unusable, ws_readable_disturbed};
 
 /* ── WritableStream (lib/internal/webstreams/writablestream.js) ──── */
 
@@ -5184,6 +5279,8 @@ static ScrDynHandleTag ws_class_tag(double kind) {
   return tags[index];
 }
 
+void scr_webstream_install(void) { ws_install(); }
+
 /* The web stream class `kind` names as a constructor value (+1): the
  * compiler's webstream.class. */
 ScrDyn *scr_webstream_class(double kind) {

tests/corpus/fetch-body-web-stream-helpers.mjs

@@ -0,0 +1,15 @@
+// Fetch bodies reached only through untyped helpers, in a program that
+// names none of the web stream classes: tee and for await in functions
+// whose parameters have no type.
+const decoder = new TextDecoder();
+const split = (body) => body.tee();
+const drain = async (stream) => {
+  let out = "";
+  for await (const chunk of stream) out += decoder.decode(chunk);
+  return out;
+};
+const response = new Response("through helpers");
+const [left, right] = split(response.body);
+console.log("helpers", await drain(left), await drain(right), response.bodyUsed);
+const request = new Request("http://example.com/", { method: "POST", body: "request helper", duplex: "half" });
+console.log("request", await drain(request.body), request.bodyUsed);

tests/corpus/fetch-body-web-stream-linked.mjs

@@ -0,0 +1,116 @@
+// Fetch Response and Request bodies as web ReadableStreams in a program
+// that also names the classes: tee, pipeTo, pipeThrough, values and
+// for await on a body, the body's lock and bodyUsed across them, and the
+// body's other uses (text, clone, a new Response) after one.
+const decoder = new TextDecoder();
+const text = (chunks) => chunks.map((c) => decoder.decode(c)).join("");
+const settled = (p) => p.then((v) => `ok:${v}`, (e) => `${e.name}:${e.message}`);
+const caught = (fn) => {
+  try {
+    return `ok:${fn()}`;
+  } catch (e) {
+    return `${e.name}:${e.message}`;
+  }
+};
+// Response.clone runs on checked-dynamic handles; the clone goes through
+// an untyped parameter.
+const cloneOf = (value) => value.clone();
+const readAll = async (stream) => {
+  const reader = stream.getReader();
+  const chunks = [];
+  for (;;) {
+    const { done, value } = await reader.read();
+    if (done) return text(chunks);
+    chunks.push(value);
+  }
+};
+
+{
+  const response = new Response("tee body");
+  const body = response.body;
+  console.log("members", typeof body.tee, typeof body.pipeTo, typeof body.pipeThrough, typeof body.values, "tee" in body, body instanceof ReadableStream);
+  console.log("before", body.locked, response.bodyUsed);
+  const [a, b] = body.tee();
+  console.log("teed", body.locked, response.bodyUsed, a instanceof ReadableStream, response.body === body);
+  console.log("branches", await readAll(a), await readAll(b), response.bodyUsed);
+  console.log("used", await settled(response.text()), caught(() => body.tee()));
+}
+
+{
+  const chunks = [];
+  const response = new Response("piped");
+  await response.body.pipeTo(new WritableStream({ write(c) { chunks.push(c); } }));
+  console.log("pipeTo", text(chunks), response.bodyUsed, response.body.locked);
+}
+
+{
+  const upper = new TransformStream({ transform(chunk, c) { c.enqueue(decoder.decode(chunk).toUpperCase()); } });
+  const out = new Response("through").body.pipeThrough(upper);
+  const reader = out.getReader();
+  console.log("pipeThrough", (await reader.read()).value, (await reader.read()).done);
+}
+
+{
+  const request = new Request("http://example.com/", { method: "POST", body: "iterate me", duplex: "half" });
+  let joined = "";
+  for await (const chunk of request.body) joined += decoder.decode(chunk);
+  console.log("for await", joined, request.bodyUsed, request.body.locked);
+  const values = [];
+  for await (const chunk of new Response("values").body.values()) values.push(decoder.decode(chunk));
+  console.log("values", values.join("|"));
+  const early = new Response("early");
+  for await (const chunk of early.body) {
+    console.log("break", chunk.length);
+    break;
+  }
+  console.log("after break", early.bodyUsed, early.body.locked, await settled(early.text()));
+}
+
+{
+  const response = new Response("read first");
+  const reader = response.body.getReader();
+  console.log("reader first", caught(() => { response.body.tee(); return "teed"; }), response.body.locked);
+  reader.releaseLock();
+  const [a] = response.body.tee();
+  console.log("after release", response.body.locked, await readAll(a));
+}
+
+{
+  const response = new Response("member value");
+  const tee = response.body.tee;
+  console.log("method value", tee === ReadableStream.prototype.tee, Array.isArray(tee.call(response.body)), response.body.locked);
+  const viewed = new Response("viewed");
+  void viewed.body.values;
+  console.log("viewed text", viewed.body.locked, viewed.bodyUsed, await viewed.text(), viewed.bodyUsed, viewed.body.locked);
+  const cloned = new Response("cloned");
+  void cloned.body.tee;
+  const copy = cloneOf(cloned);
+  console.log("viewed clone", await copy.text(), await cloned.text());
+  const source = new Response("rewrapped");
+  void source.body.pipeTo;
+  const rewrapped = new Response(source.body);
+  console.log("rewrapped", source.bodyUsed, await rewrapped.text(), source.bodyUsed);
+}
+
+{
+  // Starting a read, a cancel, a pipe or an iteration on the body's stream
+  // updates bodyUsed when it starts, as in Node.
+  const read = new Response("read now");
+  void read.body.values;
+  const pending = read.body.getReader().read();
+  console.log("bodyUsed read", read.bodyUsed);
+  await pending;
+  const canceled = new Response("cancel now");
+  void canceled.body.tee;
+  const cancel = canceled.body.cancel("why");
+  console.log("bodyUsed cancel", canceled.bodyUsed);
+  await cancel;
+  const piped = new Response("pipe now");
+  const pipe = piped.body.pipeTo(new WritableStream());
+  console.log("bodyUsed pipe", piped.bodyUsed, piped.body.locked);
+  await pipe;
+  const iterated = new Response("iterate now");
+  const step = iterated.body.values().next();
+  console.log("bodyUsed next", iterated.bodyUsed);
+  await step;
+}

tests/corpus/fetch-body-web-stream-network.mjs

@@ -0,0 +1,26 @@
+// fetch() response bodies from a local server as web ReadableStreams: for
+// await, tee, text() after a member read, and a break out of the
+// iteration, with the body's bodyUsed after them.
+import http from "node:http";
+const decoder = new TextDecoder();
+const server = http.createServer((req, res) => {
+  res.write("chunk-one;");
+  setTimeout(() => { res.end("chunk-two"); }, 5);
+});
+await new Promise((r) => server.listen(0, "127.0.0.1", r));
+const url = `http://127.0.0.1:${server.address().port}/`;
+const a = await fetch(url);
+let s = "";
+for await (const c of a.body) s += decoder.decode(c);
+console.log("for await", s, a.bodyUsed);
+const b = await fetch(url);
+const [x, y] = b.body.tee();
+const readAll = async (st) => { let t = ""; for await (const c of st) t += decoder.decode(c); return t; };
+console.log("tee", await readAll(x), await readAll(y));
+const c = await fetch(url);
+void c.body.values;
+console.log("text", await c.text());
+const d = await fetch(url);
+for await (const chunk of d.body) { console.log("first", decoder.decode(chunk)); break; }
+console.log("after break", d.bodyUsed);
+server.close();

tests/corpus/fetch-body-web-stream-unlinked.mjs

@@ -0,0 +1,25 @@
+// Fetch bodies as web ReadableStreams in a program that names none of the
+// web stream classes: tee, values and for await on Response and Request
+// bodies, and the body's lock and bodyUsed after them.
+const decoder = new TextDecoder();
+const settled = (p) => p.then((v) => `ok:${v}`, (e) => `${e.name}:${e.message}`);
+const drain = async (stream) => {
+  let out = "";
+  for await (const chunk of stream) out += decoder.decode(chunk);
+  return out;
+};
+
+const response = new Response("two ways");
+const [left, right] = response.body.tee();
+console.log("tee", await drain(left), await drain(right), response.bodyUsed, response.body.locked);
+console.log("used", await settled(response.text()));
+
+const request = new Request("http://example.com/", { method: "PUT", body: "request body", duplex: "half" });
+console.log("for await", await drain(request.body), request.bodyUsed);
+
+const parts = [];
+for await (const chunk of new Response("values()").body.values({ preventCancel: true })) parts.push(chunk.length);
+console.log("values", parts.join(","));
+
+const plain = new Response("plain");
+console.log("plain", plain.body.locked, await plain.text(), plain.bodyUsed);

tests/harness/corpus-costs/fetch-body-web-stream-helpers.mjs.json

@@ -0,0 +1 @@
+{ "plain": 2600, "san": 48000 }

tests/harness/corpus-costs/fetch-body-web-stream-linked.mjs.json

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

tests/harness/corpus-costs/fetch-body-web-stream-network.mjs.json

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

tests/harness/corpus-costs/fetch-body-web-stream-unlinked.mjs.json

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