Address multiple TS-streams defects / clarifications

src/per_isolate/webstreams/AGENTS.md

@@ -30,9 +30,10 @@ private-brand dispatch, no `instanceof`) apply here — see
   lists (`queue.ts` and `native.ts` headers).
 - The native source contract (marker symbol, standard pull/cancel hooks,
   byobRequest discrimination, once-per-pull delivery, per-pull abort
-  signal for cancellation, under-delivery = fused
-  `{done: true, value: partial}` EOF, tee hook, `expectedLength`
-  exact-total byte contract) is specified in the `native.ts` header.
+  signal for cancellation, under-delivery = EOF signal delivering the
+  partial as `{done: false, value: partial}` with the next read
+  observing EOF, tee hook, `expectedLength` exact-total byte contract)
+  is specified in the `native.ts` header.
   The C++ implementation (`ReadableStreamNativeSource` in
   `src/workerd/api/js-readable-stream.{h,c++}`) MUST conform to it; JS
   mocks in tests exercise the conduit independently. Key addition:

src/per_isolate/webstreams/compression.ts

@@ -108,17 +108,17 @@ function isActualObject(value: unknown): boolean {
 }
 
 // True for BufferSource chunks the codec accepts: ArrayBuffers and views,
-// excluding anything SharedArrayBuffer-backed (per Web IDL, [AllowShared] is
-// not granted here; WPT pins the rejection). Captured getters are used for
-// the view's buffer — prototype accessors are user-patchable.
+// INCLUDING anything SharedArrayBuffer-backed — the shared bytes are
+// copied out by snapshotChunk, matching the identity streams and the C++
+// implementation (DECIDED 2026-08-28; the strict [AllowShared]-less
+// BufferSource reading was considered and overridden for internal
+// consistency, which is why the WPT bad-chunks files are disabled).
 function isValidChunk(chunk: unknown): boolean {
-  if (isArrayBuffer(chunk)) return true;
-  if (isSharedArrayBuffer(chunk)) return false;
-  if (!isArrayBufferView(chunk)) return false;
-  const buffer = isDataView(chunk)
-    ? DataViewPrototypeGetBuffer(chunk)
-    : TypedArrayPrototypeGetBuffer(chunk);
-  return !isSharedArrayBuffer(buffer);
+  return (
+    isArrayBuffer(chunk) ||
+    isSharedArrayBuffer(chunk) ||
+    isArrayBufferView(chunk)
+  );
 }
 
 // Validates a chunk and copies its CURRENT bytes. Runs synchronously
@@ -141,6 +141,15 @@ function snapshotChunk(chunk: unknown): Uint8Array {
     buffer = chunk as ArrayBuffer;
     byteOffset = 0;
     byteLength = ArrayBufferPrototypeByteLengthGet(chunk) as number;
+  } else if (isSharedArrayBuffer(chunk)) {
+    // A raw SharedArrayBuffer: measure through a fresh whole-buffer view
+    // (SharedArrayBuffer.prototype.byteLength is not among the primordial
+    // captures; the view's length is read through the captured getter).
+    buffer = chunk as unknown as ArrayBuffer;
+    byteOffset = 0;
+    byteLength = TypedArrayPrototypeGetByteLength(
+      new Uint8Array(chunk as unknown as ArrayBuffer)
+    ) as number;
   } else if (isDataView(chunk)) {
     buffer = DataViewPrototypeGetBuffer(chunk) as ArrayBuffer;
     byteOffset = DataViewPrototypeGetByteOffset(chunk) as number;
@@ -251,12 +260,12 @@ function createCodecPair(
           );
         }
         if (!entry.ok) {
-          // An invalid chunk errors BOTH sides, matching the legacy
-          // implementation (any write failure errored the whole pair) —
-          // without this the readable side would hang on its pending
-          // pull.
-          failBoth(entry.error);
-          throw entry.error;
+          // An invalid chunk rejects ITS OWN write only — the non-fatal
+          // write-rejection channel keeps the pair usable and later
+          // queued writes still deliver (the per-write invalid-chunk
+          // contract shared with the identity streams and the C++
+          // implementation). Codec failures below remain FATAL.
+          throw writableInternals.nonFatalWriteRejection(entry.error);
         }
         // EAGER: the codec consumes the snapshot; a codec error throws
         // HERE, rejecting the write — the spec's transform-time error

src/per_isolate/webstreams/identity.ts

@@ -410,13 +410,13 @@ class IdentityTransformStream {
         );
       }
       const entry = ArrayPrototypeShift(snapshots) as SnapshotEntry;
-      // A recorded validation error surfaces here, at its FIFO turn: this
-      // write rejects and the stream errors, but everything written before
-      // it has already been delivered. Queued writes are discarded by the
-      // erroring writable without sink steps; their snapshots go with them.
+      // A recorded validation error surfaces here, at its FIFO turn, as a
+      // NON-FATAL write rejection: this write's promise rejects while the
+      // stream stays usable and queued writes behind it still deliver —
+      // the per-write invalid-chunk contract shared with the C++ internal
+      // controllers.
       if (!entry.ok) {
-        snapshots.length = 0;
-        throw entry.error;
+        throw writableInternals.nonFatalWriteRejection(entry.error);
       }
       const copied = entry.copied;
       if (copied === undefined) return; // zero-length no-op

src/per_isolate/webstreams/native.ts

@@ -49,17 +49,16 @@
 //     accumulation toward `atLeast` happens inside the native source;
 //     the conduit never re-pulls an unsatisfied read. Responding with
 //     fewer bytes than `atLeast` IMPLICITLY SIGNALS CLOSURE: the partial
-//     fill commits fused as { done: true, value: partialView }, the
-//     stream closes, and every subsequent read returns EOF. This result
-//     is deliberately INDISTINGUISHABLE from the standard's one fused
-//     corner — a pending min read committed in the closed state via
-//     close() + respond(0) (RespondInClosedState →
-//     CommitPullIntoDescriptor, which fulfills the read-into request
-//     with the partial view and done: true). The signaling MECHANISM
-//     differs (a queued JS source is pulled again and must close()
-//     explicitly; for a native source the under-delivery itself is the
-//     signal), but consumers observe the same result shapes on both
-//     backends.
+//     fill commits as { done: false, value: partialView }, the stream
+//     closes, and every subsequent read returns EOF (BYOB reads get a
+//     zero-length view over their transferred buffer). This is the
+//     decided C++-parity readAtLeast tail shape — the below-minimum
+//     tail is never fused with the done flag — and it matches the
+//     queued backend's deferred end-of-data commit, so consumers
+//     observe the same tail on both backends. (Only an explicit
+//     same-turn respond(0)-after-close on the QUEUED backend produces
+//     the spec's fused { done: true, value: partial } fold; the native
+//     conduit has no respond(0) — close() is the EOF verb.)
 //   - Close is otherwise a fused close-commit: controller.close() may
 //     follow a respond() in the same pull turn — deliberately divergent
 //     from the queued backend's drain-then-sentinel model.
@@ -979,16 +978,17 @@ class NativePullConduit implements ByteStreamConsumerType {
     }
     // MIN-READ CONTRACT: the respond is the source's COMPLETE answer for
     // this read. Delivering fewer bytes than the requested minimum
-    // (atLeast) signals end-of-stream: the partial fill commits FUSED
-    // with done: true, the stream closes, and every subsequent read
-    // returns EOF. The conduit never re-pulls an unsatisfied read — any
+    // (atLeast) signals end-of-stream: the partial fill commits
+    // { done: false, value: partialView }, the stream closes, and every
+    // subsequent read returns EOF — the decided C++-parity readAtLeast
+    // tail shape. The conduit never re-pulls an unsatisfied read — any
     // accumulation toward the minimum happens inside the native source
     // (tryRead-style). State transitions before promise resolutions.
     //
     // EXPECTED-LENGTH: under-delivery is a close signal, so the
     // exact-total contract applies — landing short of expectedLength is
     // underflow and errors the stream instead of closing it. (Landing
-    // exactly on it is a legitimate fused close.)
+    // exactly on it is a legitimate close-with-delivery.)
     if (
       this.#expectedLength !== undefined &&
       this.#bytesDelivered < this.#expectedLength
@@ -1005,7 +1005,7 @@ class NativePullConduit implements ByteStreamConsumerType {
       return;
     }
     this.#status = 'closed';
-    desc.resolve({ done: true, value });
+    desc.resolve({ done: false, value });
     this.#settleRemainingAsEof();
     this.#hooks.closeStream();
   }
@@ -1075,11 +1075,11 @@ class NativePullConduit implements ByteStreamConsumerType {
     this.#byobRequestCache = null;
 
     // DEFENSIVE, currently unreachable: under the min-read contract every
-    // respond commits (satisfied → done: false; under-delivered → fused
-    // EOF), so a partially-filled descriptor cannot persist to close().
-    // Retained because the contract describes close() fusing with a
-    // partial fill, should a future revision reintroduce partial
-    // responds.
+    // respond commits (satisfied → done: false; under-delivered → the
+    // done: false tail followed by EOF), so a partially-filled descriptor
+    // cannot persist to close(). Retained should a future revision
+    // reintroduce partial responds; the shape mirrors the under-delivery
+    // tail (partial done: false, then #settleRemainingAsEof).
     const head = this.#requests[0];
     if (
       head !== undefined &&
@@ -1101,7 +1101,7 @@ class NativePullConduit implements ByteStreamConsumerType {
       }
       ArrayPrototypeSplice(this.#requests, 0, 1);
       desc.resolve({
-        done: true,
+        done: false,
         value: new desc.viewCtor(
           desc.buffer,
           desc.byteOffset,