diff --git a/packages/core/src/body/multipart-body.test.ts b/packages/core/src/body/multipart-body.test.ts
index 5b03d39..d616008 100644
--- a/packages/core/src/body/multipart-body.test.ts
+++ b/packages/core/src/body/multipart-body.test.ts
@@ -1,8 +1,9 @@
 // SPDX-License-Identifier: MIT
 // packages/core/src/body/multipart-body.test.ts
 // Exercises: BODY-2 (composite replayability, unknown-length collapse), HTTP-51 (shared framing routine,
-// boundary generation/validation, header quoting, and a part media type that cannot break the framing),
-// HTTP-26 (a media type is header-safe), RECOV-12 (a close failure never masks the primary failure)
+// boundary generation/validation, header quoting, a boundary parameter rendered so an RFC 9110 parser
+// can read it, and a part media type that cannot break the framing), HTTP-26 (a media type is
+// header-safe), RECOV-12 (a close failure never masks the primary failure)
 import {describe, expect, test} from 'bun:test';
 import fc from 'fast-check';
 import {MediaTypeParseError} from '../http/errors.js';
@@ -66,9 +67,11 @@ function collectingSink(): {
   };
 }
 
-async function drain(body: {
+// `Uint8Array<ArrayBuffer>`, not the bare alias: `BodyInit` excludes a view over a `SharedArrayBuffer`,
+// so the default `ArrayBufferLike` parameter is not assignable to the platform `Response` below.
+async function drainBytes(body: {
   writeTo: (sink: WritableStream<Uint8Array>) => Promise<void>;
-}): Promise<string> {
+}): Promise<Uint8Array<ArrayBuffer>> {
   const chunks: Uint8Array[] = [];
   await body.writeTo(new WritableStream({write: c => void chunks.push(c)}));
   const total = chunks.reduce((s, c) => s + c.length, 0);
@@ -78,7 +81,13 @@ async function drain(body: {
     out.set(c, offset);
     offset += c.length;
   }
-  return new TextDecoder().decode(out);
+  return out;
+}
+
+async function drain(body: {
+  writeTo: (sink: WritableStream<Uint8Array>) => Promise<void>;
+}): Promise<string> {
+  return new TextDecoder().decode(await drainBytes(body));
 }
 
 describe('MultipartBody replayability and length (BODY-2)', () => {
@@ -186,6 +195,47 @@ describe('MultipartBody boundary generation and validation (HTTP-51)', () => {
   });
 });
 
+describe('the rendered boundary parameter is a parseable one (HTTP-51)', () => {
+  // RFC 2046 `bchars` and RFC 9110 `tchar` are different sets: ' ', ',', ':', '=', '?', '/', '(' and
+  // ')' are legal in a boundary and illegal bare in a header parameter value. `validateBoundary` admits
+  // the first grammar and the renderer owes the second, so a boundary the constructor accepts must come
+  // back out quoted rather than bare.
+  test('a boundary that is not a bare token is quoted', () => {
+    expect(
+      multipartBody([{name: 'a', body: stringBody('x')}], 'a,b').mediaType,
+    ).toBe('multipart/form-data; boundary="a,b"');
+  });
+
+  test('a boundary that IS a bare token is left unquoted', () => {
+    expect(
+      multipartBody([{name: 'a', body: stringBody('x')}], 'plain-1').mediaType,
+    ).toBe('multipart/form-data; boundary=plain-1');
+    // The generated default stays byte-identical: it is drawn from ALPHA/DIGIT only.
+    expect(
+      multipartBody([{name: 'a', body: stringBody('x')}]).mediaType,
+    ).toMatch(/^multipart\/form-data; boundary=dexpace-[A-Za-z0-9]{32}$/);
+  });
+
+  test.each([['a,b'], ['bound ary'], ['a:b'], ['a=b'], ['a?b'], ['(a)/b']])(
+    'the header a peer receives round-trips through a real parameter parser: %p',
+    async boundary => {
+      // The runtime's own multipart parser, standing in for the peer. Bun's happens to tolerate the
+      // unquoted form, so this is the regression guard and NOT the reproducer: Node's (undici's)
+      // rejects the whole body with `TypeError: Failed to parse body as FormData`, which is why the
+      // same case is also in `tests/node-conformance/body-lifecycle.test.mjs`. Two independent
+      // parsers disagreeing about our own Content-Type is exactly what that tree is for.
+      const body = multipartBody(
+        [{name: 'field', body: stringBody('value')}],
+        boundary,
+      );
+      const response = new globalThis.Response(await drainBytes(body), {
+        headers: {'content-type': body.mediaType},
+      });
+      expect((await response.formData()).get('field')).toBe('value');
+    },
+  );
+});
+
 describe('MultipartBodyBuilder (HTTP-2, HTTP-3)', () => {
   test('static newBuilder and instance newBuilder pre-populates parts and boundary', async () => {
     const original = MultipartBody.newBuilder()
diff --git a/packages/core/src/body/multipart-body.ts b/packages/core/src/body/multipart-body.ts
index 6df3020..e1affd3 100644
--- a/packages/core/src/body/multipart-body.ts
+++ b/packages/core/src/body/multipart-body.ts
@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: MIT
 // packages/core/src/body/multipart-body.ts
 import type {Builder} from '../http/builder.js';
+import {MediaType} from '../http/media-type.js';
 import {EndOfStreamError} from '../io/errors.js';
 import {invariant} from '../invariant.js';
 import type {Body} from './body.js';
@@ -77,6 +78,31 @@ function renderPartHeader(part: MultipartPart, boundary: string): Uint8Array {
   return new TextEncoder().encode(header);
 }
 
+/**
+ * HTTP-51: the `Content-Type` a peer actually parses.
+ *
+ * RFC 2046 `bchars` and RFC 9110 `tchar` are different sets. `BOUNDARY_PATTERN` above admits ' ', ',',
+ * ':', '=', '?', '/', '(' and ')', none of which is a `tchar`, so interpolating the boundary bare
+ * produces a parameter value that stops at the first offending byte -- `boundary=a,b` reads as
+ * `boundary=a` plus a junk parameter, and the peer then never finds a delimiter. Node's own FormData
+ * parser rejects such a body outright with `TypeError: Failed to parse body as FormData`.
+ *
+ * Rendered through {@link MediaType} rather than a second quoting routine here: it is the module that
+ * owns HTTP-25's token-or-quoted-string decision, and `parse(render(x)) === x` is its guarantee. A
+ * boundary that IS a bare token still renders bare, so the generated default is byte-identical to what
+ * this class emitted before.
+ *
+ * Narrowing `validateBoundary` to `tchar` instead was rejected: HTTP-51 asks that a boundary VIOLATING
+ * the RFC 2046 grammar be refused, not that a conforming one be. The defect is in the rendering.
+ */
+function renderMediaType(boundary: string): string {
+  return MediaType.of(
+    'multipart',
+    'form-data',
+    new Map([['boundary', boundary]]),
+  ).render();
+}
+
 function trailerBytes(boundary: string): Uint8Array {
   return new TextEncoder().encode(`--${boundary}--\r\n`);
 }
@@ -144,7 +170,10 @@ function nonClosingSink(
 export class MultipartBody implements Body {
   /** Discriminates this variant within the {@link Body} union. */
   readonly kind = 'multipart' as const;
-  /** `multipart/form-data` carrying the boundary this instance frames its parts with. */
+  /**
+   * `multipart/form-data` carrying the boundary this instance frames its parts with, with the
+   * `boundary` parameter quoted whenever it is not a bare RFC 9110 token (HTTP-51).
+   */
   readonly mediaType: string;
   /** The total framed byte count, or -1 when any part's own length is unknown (BODY-2). */
   readonly contentLength: number;
@@ -157,7 +186,7 @@ export class MultipartBody implements Body {
     if (boundary !== undefined) validateBoundary(boundary);
     this.#boundary = boundary ?? generateBoundary();
     this.#parts = [...parts];
-    this.mediaType = `multipart/form-data; boundary=${this.#boundary}`;
+    this.mediaType = renderMediaType(this.#boundary);
     this.replayable = this.#parts.every(part => part.body.replayable);
     this.contentLength = computeContentLength(this.#parts, this.#boundary);
     invariant(
diff --git a/packages/core/src/body/response-body-logging.test.ts b/packages/core/src/body/response-body-logging.test.ts
index d118273..9fe6b83 100644
--- a/packages/core/src/body/response-body-logging.test.ts
+++ b/packages/core/src/body/response-body-logging.test.ts
@@ -4,11 +4,16 @@
 // reads), BODY-24 (exceeds-cap: prefix+tail once, second read fails), BODY-26 (drain failure cached,
 // partial bytes retained, error() does not drain), BODY-27 (close-once shared guard), BODY-28 (captured
 // buffer survives close), BODY-29 (reported length), BODY-32 (negative cap rejected), BODY-25 (a
-// zero-length delegate chunk is a stream-contract violation, never end-of-stream)
+// zero-length delegate chunk is a stream-contract violation, never end-of-stream), BODY-27/BODY-28 again
+// (close() ends the drain rather than poisoning the wrapper: snapshot still serves the captured prefix,
+// read() reports IO-42's state error, error() reports only a genuine drain failure)
 import {describe, expect, test} from 'bun:test';
 import fc from 'fast-check';
 import {InvariantViolation} from '../invariant.js';
-import {SourceContractViolationError} from '../io/errors.js';
+import {
+  ClosedResourceError,
+  SourceContractViolationError,
+} from '../io/errors.js';
 import {withResponseLogging} from './response-body-logging.js';
 
 function readableOf(...chunks: number[][]): ReadableStream<Uint8Array> {
@@ -282,6 +287,87 @@ describe('the tail path enforces the same chunk contract (BODY-25)', () => {
   });
 });
 
+describe('the tap is inert after close(), not poisoned by it (BODY-27, BODY-28)', () => {
+  // closeDelegate releases the reader. Every entry point that used to start a drain unconditionally then
+  // read from a detached reader, so `snapshot()` cached a raw `TypeError: Invalid state: The reader is
+  // not attached to a stream` as the wrapper's failure -- and `error()` reported that forever, over a
+  // capture that never failed. BODY-28 says the captured bytes survive close; they cannot survive it
+  // behind a fabricated error.
+
+  test('close-then-snapshot returns the captured prefix and starts no drain', async () => {
+    const logged = withResponseLogging(readableOf([1, 2], [3, 4]), 2);
+    await logged.read(); // exceeds-cap: the prefix is captured, the delegate stays live
+    await logged.close();
+
+    expect([...logged.snapshot()]).toEqual([1, 2]);
+    await new Promise(resolve => setTimeout(resolve, 0)); // a drain started here would settle by now
+    expect(logged.error()).toBeNull();
+  });
+
+  test('close-before-any-read leaves snapshot empty and error() null', async () => {
+    const logged = withResponseLogging(readableOf([1, 2, 3]), 100);
+    await logged.close();
+
+    expect([...logged.snapshot()]).toEqual([]);
+    await new Promise(resolve => setTimeout(resolve, 0));
+    expect(logged.error()).toBeNull();
+  });
+
+  test('close-then-read rejects with ClosedResourceError, not a detached-reader TypeError', async () => {
+    const logged = withResponseLogging(readableOf([1, 2, 3]), 100);
+    await logged.close();
+
+    const error = await rejection(logged.read());
+    expect(error).toBeInstanceOf(ClosedResourceError);
+    expect(error.message).toBe('LoggedResponseBody is closed');
+  });
+
+  test('close-then-read in the exceeds-cap regime rejects too -- there is no live tail left', async () => {
+    // The drain stopped at the cap and nobody took the tail, so the captured prefix is NOT the whole
+    // body. Serving it would hand the consumer a silently truncated response; the delegate that held
+    // the rest is gone.
+    const logged = withResponseLogging(readableOf([1, 2], [3, 4]), 2);
+    logged.snapshot(); // starts the drain without taking the tail
+    await new Promise(resolve => setTimeout(resolve, 0));
+    expect([...logged.snapshot()]).toEqual([1, 2]);
+    await logged.close();
+
+    expect(await rejection(logged.read())).toBeInstanceOf(ClosedResourceError);
+  });
+
+  test('close-then-error reports only a genuine drain failure (BODY-26)', async () => {
+    const boom = new Error('upstream reset');
+    const failing = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(Uint8Array.from([1, 2]));
+      },
+      pull(controller) {
+        controller.error(boom);
+      },
+    });
+    const logged = withResponseLogging(failing, 100);
+    expect(await rejection(logged.read())).toBe(boom);
+    // `cancel()` on an errored stream rejects with that stream's own stored error, which the
+    // 'a non-TypeError from cancel() propagates' case above already pins. Not what this test is about.
+    await logged.close().catch(() => undefined);
+
+    // The real failure is not displaced by a close-induced one, and snapshot still shows the partial
+    // capture BODY-26 asked to be retained.
+    expect(logged.error()).toBe(boom);
+    expect([...logged.snapshot()]).toEqual([1, 2]);
+  });
+
+  test('the fits-cap regime still serves repeatable reads after its own close (BODY-23, BODY-28)', async () => {
+    // The drain itself closes the delegate on this path, so "closed" must NOT mean "unreadable".
+    const logged = withResponseLogging(readableOf([1, 2, 3]), 100);
+    expect([...(await readAll(await logged.read()))]).toEqual([1, 2, 3]);
+    await logged.close();
+    expect([...(await readAll(await logged.read()))]).toEqual([1, 2, 3]);
+    expect([...logged.snapshot()]).toEqual([1, 2, 3]);
+    expect(logged.error()).toBeNull();
+  });
+});
+
 describe('snapshot is a drain trigger (BODY-22)', () => {
   test('calling snapshot starts the drain, without a read()', async () => {
     const logged = withResponseLogging(readableOf([1, 2, 3]), 100);
diff --git a/packages/core/src/body/response-body-logging.ts b/packages/core/src/body/response-body-logging.ts
index a5ebc0a..0e23e01 100644
--- a/packages/core/src/body/response-body-logging.ts
+++ b/packages/core/src/body/response-body-logging.ts
@@ -2,7 +2,10 @@
 // packages/core/src/body/response-body-logging.ts
 import {invariant} from '../invariant.js';
 import {ByteQueue} from '../io/byte-queue.js';
-import {SourceContractViolationError} from '../io/errors.js';
+import {
+  ClosedResourceError,
+  SourceContractViolationError,
+} from '../io/errors.js';
 import {MAX_BYTE_ARRAY_LENGTH} from '../io/limits.js';
 import {ConsumedBodyError} from './errors.js';
 
@@ -20,11 +23,19 @@ export interface LoggedResponseBody {
    * (BODY-22). Fits-cap regime: every call, including calls after the first, returns a fresh
    * non-consuming view over the captured bytes (BODY-23). Exceeds-cap regime: exactly one call is
    * allowed; a second throws (BODY-24). If the drain failed, every call re-throws the cached error.
+   * After `close()` in any regime but fits-cap, throws `ClosedResourceError`: the delegate is gone and
+   * the captured prefix is not the whole body (BODY-27, BODY-28).
    */
   read(): Promise<ReadableStream<Uint8Array>>;
-  /** Non-consuming; reflects whatever has been captured so far, even after a failed drain (BODY-26). */
+  /**
+   * Non-consuming; reflects whatever has been captured so far, even after a failed drain (BODY-26) and
+   * after `close()` (BODY-28), which it never restarts a drain past.
+   */
   snapshot(): Uint8Array;
-  /** The cached drain failure, or null. MUST NOT trigger a drain (BODY-26). */
+  /**
+   * The cached drain failure, or null. MUST NOT trigger a drain (BODY-26), and reports only a genuine
+   * upstream failure -- never one manufactured by reading past `close()`.
+   */
   error(): Error | null;
   /** Captured size iff fully captured within the cap, else the delegate's declared length (BODY-29). */
   readonly contentLength: number;
@@ -129,8 +140,16 @@ async function drainOnce(state: DrainState): Promise<void> {
  * The detached `.catch` matters: a snapshot-triggered drain has no awaiter, so without it a drain failure
  * becomes an unhandled rejection. Attaching a handler to a *copy* leaves the stored promise rejected, so
  * `read()` still re-throws the cached failure on every call (BODY-26).
+ *
+ * BODY-28: after `close()` there is nothing left to drain -- `closeDelegate` released the reader, so
+ * starting one here reads from a detached reader, raises a raw `TypeError: Invalid state`, and
+ * `drainOnce`'s catch caches it as this wrapper's `failure`. `error()` would then report a fabricated
+ * upstream failure forever, over a capture that never failed, and the captured bytes BODY-28 promises
+ * survive close would be reachable only past that lie. A drain already in flight is left alone: on the
+ * fits-cap path the drain closes the delegate itself, and its own promise is what `read()` awaits.
  */
 function startDrain(state: DrainState): Promise<void> {
+  if (state.closed && state.started === undefined) return Promise.resolve();
   state.started ??= drainOnce(state);
   void state.started.catch(() => undefined);
   return state.started;
@@ -226,10 +245,17 @@ export function withResponseLogging(
   return {
     async read(): Promise<ReadableStream<Uint8Array>> {
       await startDrain(state); // a cached failure re-throws here on every call (BODY-26)
+      // Ordered deliberately. `fits` first: on that path the drain closed the delegate itself, and
+      // BODY-23 still requires every later read to be a fresh non-consuming view -- "closed" there does
+      // not mean "unreadable" (BODY-28).
       if (state.regime === 'fits') return capturedStream(state);
       if (state.tailConsumed) {
         throw new ConsumedBodyError('logged-response');
       }
+      // Anything else with the delegate gone: there is no live tail to continue from, and the captured
+      // prefix is not the whole body, so serving it would hand the consumer a silently truncated
+      // response. IO-42's state error, not the raw `TypeError` a detached reader throws.
+      if (state.closed) throw new ClosedResourceError('LoggedResponseBody');
       state.tailConsumed = true;
       return tailStream(state);
     },
@@ -238,6 +264,7 @@ export function withResponseLogging(
       // so it starts the drain and returns what has been captured so far rather than awaiting it; a
       // later read() awaits the very same in-flight promise, so the delegate is still read exactly once.
       // (BODY-26's "snapshot returns the partial bytes without throwing" is why it cannot await here.)
+      // After close() `startDrain` is a no-op, so this is the post-mortem accessor BODY-28 asks for.
       void startDrain(state);
       return state.captured.snapshot();
     },
diff --git a/packages/core/src/body/stream-body.test.ts b/packages/core/src/body/stream-body.test.ts
index 3530875..b966275 100644
--- a/packages/core/src/body/stream-body.test.ts
+++ b/packages/core/src/body/stream-body.test.ts
@@ -6,11 +6,13 @@
 // delivered-of-declared, and an overrunning stream is stopped BEFORE the extra bytes reach the sink),
 // IO-3 (a contentLength below the -1 sentinel is rejected), HTTP-26/HTTP-51 (a media type is
 // header-safe), RECOV-12 (a close failure never masks the primary write failure), HTTP-1 (frozen at
-// construction so the declared length cannot be desynced from the written bytes)
+// construction so the declared length cannot be desynced from the written bytes), HTTP-39/BODY-10 again
+// (a zero-length delivery during an exact-length copy is a source-contract violation, never a no-op and
+// never spun on, and no empty chunk reaches the sink)
 import {describe, expect, test} from 'bun:test';
 import {MediaTypeParseError} from '../http/errors.js';
 import {InvariantViolation} from '../invariant.js';
-import {EndOfStreamError} from '../io/errors.js';
+import {EndOfStreamError, SourceContractViolationError} from '../io/errors.js';
 import {ConsumedBodyError} from './errors.js';
 import {streamBody} from './stream-body.js';
 
@@ -49,6 +51,16 @@ function collectingSink(): {
   };
 }
 
+/** Awaits a rejection and returns its reason, failing loudly when the promise resolves instead. */
+async function rejection(promise: Promise<unknown>): Promise<unknown> {
+  try {
+    await promise;
+  } catch (error: unknown) {
+    return error;
+  }
+  throw new Error('expected a rejection, but the promise resolved');
+}
+
 describe('caller stream ownership (BODY-8)', () => {
   test('a sink failure does not cancel the caller stream on the unknown-length path', async () => {
     // `pipeTo`'s default (`preventCancel: false`) cancels the SOURCE when the destination errors,
@@ -230,6 +242,85 @@ describe('a mis-framed body never reaches the wire (HTTP-39/BODY-10)', () => {
   });
 });
 
+describe('a zero-length delivery is a source-contract violation (HTTP-39/BODY-10)', () => {
+  /**
+   * The auditor's probe, bounded so an unguarded run terminates instead of hanging the suite: a source
+   * that delivers nothing but empty chunks, then finally ends. Unbounded is the real-world shape, and
+   * the pull count below is what proves the copy did not spin on it.
+   */
+  function emptyOnly(limit: number): {
+    stream: ReadableStream<Uint8Array>;
+    pulls: () => number;
+  } {
+    let pulls = 0;
+    const stream = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        pulls += 1;
+        if (pulls > limit) {
+          controller.close();
+          return;
+        }
+        controller.enqueue(new Uint8Array(0));
+      },
+    });
+    return {stream, pulls: () => pulls};
+  }
+
+  test('an empty-only source raises on the first empty chunk rather than spinning on it', async () => {
+    const {stream, pulls} = emptyOnly(1000);
+    const {state, sink} = probeSink();
+    const error = await rejection(
+      streamBody(stream, undefined, 3).writeTo(sink),
+    );
+
+    // Not EndOfStreamError after a thousand futile reads: an unbounded source of empty chunks never
+    // ends, so nothing downstream can ever diagnose it, and every one of those chunks reaches the sink.
+    expect(error).toBeInstanceOf(SourceContractViolationError);
+    // Two, not one: the default queuing strategy reads one chunk ahead, so the source is pulled again
+    // the moment our read drains its queue. What matters is that it is not `limit`.
+    expect(pulls()).toBeLessThanOrEqual(2);
+    expect(state.written).toEqual([]);
+    expect(state.aborted).toBe(true); // a mis-framed body is never signalled as a clean close
+  });
+
+  test('an empty chunk between real chunks is raised, and never reaches the sink', async () => {
+    const emptyBetween = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(Uint8Array.from([1, 2]));
+        controller.enqueue(new Uint8Array(0));
+        controller.enqueue(Uint8Array.from([3]));
+        controller.close();
+      },
+    });
+    const chunkLengths: number[] = [];
+    const sink = new WritableStream<Uint8Array>({
+      write: chunk => void chunkLengths.push(chunk.length),
+    });
+
+    expect(
+      await rejection(streamBody(emptyBetween, undefined, 3).writeTo(sink)),
+    ).toBeInstanceOf(SourceContractViolationError);
+    // `io/buffered-sink.ts` writeString: a zero-length chunk is HTTP/1.1 chunked encoding's TERMINATING
+    // chunk, so forwarding one ends the request body early on the wire.
+    expect(chunkLengths).toEqual([2]);
+  });
+
+  test('a declared length of 0 over a source that just closes stays a legitimate empty write (BODY-10)', async () => {
+    const {state, sink} = probeSink();
+    await streamBody(
+      new ReadableStream({
+        start: c => {
+          c.close();
+        },
+      }),
+      undefined,
+      0,
+    ).writeTo(sink);
+    expect(state.written).toEqual([]);
+    expect(state.closed).toBe(true);
+  });
+});
+
 describe('StreamBody media type and failure propagation', () => {
   test('rejects a media type carrying CR/LF (HTTP-26/HTTP-51)', () => {
     expect(() =>
diff --git a/packages/core/src/body/stream-body.ts b/packages/core/src/body/stream-body.ts
index c0ba250..093e0a7 100644
--- a/packages/core/src/body/stream-body.ts
+++ b/packages/core/src/body/stream-body.ts
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: MIT
 // packages/core/src/body/stream-body.ts
-import {EndOfStreamError} from '../io/errors.js';
+import {EndOfStreamError, SourceContractViolationError} from '../io/errors.js';
 import {invariant} from '../invariant.js';
 import type {Body} from './body.js';
 import {ConsumedBodyError} from './errors.js';
@@ -8,6 +8,33 @@ import {freezeBody} from './freeze-body.js';
 import {assertHeaderSafeMediaType} from './media-type-safety.js';
 import {withBodyWriter} from './write-body.js';
 
+/**
+ * HTTP-39/BODY-10: a zero-length delivery during an exact-length copy is a source-contract violation,
+ * never end-of-stream and never something to spin on -- `{done: true}` is the only end signal.
+ *
+ * Two failure modes, and neither one is diagnosable anywhere else. A source that only ever yields empty
+ * chunks never ends, so the delivered-of-declared check below is never reached; and every empty chunk
+ * that gets past here reaches the transport sink, where to an HTTP/1.1 chunked-encoding transport a
+ * zero-length chunk is the TERMINATING chunk (`io/buffered-sink.ts`'s `writeString`) -- so tolerating
+ * one ends the request body early on the wire while the copy still believes it is mid-body.
+ *
+ * The wording and the error type are `io/retention-window.ts`'s deliberately: `read()` carries no
+ * requested count, so the requirement's "for a positive requested count" has no literal analog, and a
+ * request body reaches both this copy and `BufferedSource` -- a divergence would make the same upstream
+ * fail or succeed depending only on which wrapper it passed through. `body/response-body-logging.ts`
+ * makes the same call for BODY-25 on the response side.
+ *
+ * A declared length of 0 is still a legitimate empty write (BODY-10): that is a source that signals
+ * `{done: true}` immediately, which never reaches this check.
+ */
+function assertNonEmptyChunk(value: Uint8Array): void {
+  if (value.length === 0) {
+    throw new SourceContractViolationError(
+      'source delivered 0 bytes without signalling end of stream',
+    );
+  }
+}
+
 /**
  * A single-use body backed by a caller-supplied stream.
  *
@@ -50,6 +77,8 @@ export class StreamBody implements Body {
    * @throws {@link ConsumedBodyError} on a second call -- this body is single-use (BODY-3).
    * @throws EndOfStreamError when a declared `contentLength` disagrees with the bytes the stream
    * actually yields, in either direction (HTTP-39/BODY-10).
+   * @throws SourceContractViolationError when the stream delivers a zero-length chunk without
+   * signalling end of stream during a declared-length write (HTTP-39/BODY-10).
    */
   async writeTo(sink: WritableStream<Uint8Array>): Promise<void> {
     if (this.#consumed) throw new ConsumedBodyError('stream');
@@ -80,6 +109,7 @@ export class StreamBody implements Body {
           // Serial by necessity: each read depends on the previous one advancing the cursor.
           const {done, value} = await reader.read();
           if (done) break;
+          assertNonEmptyChunk(value); // HTTP-39/BODY-10
           // Checked BEFORE the write, not after the loop: once a transport has stamped the declared
           // Content-Length, an overrun byte sits on the socket where the peer reads it as the start of
           // the next message, and a thrown error cannot recall bytes already written (HTTP-39/BODY-10).
@@ -109,6 +139,8 @@ export class StreamBody implements Body {
  * @throws ConsumedBodyError from `writeTo` when the body has already been written once (BODY-3).
  * @throws EndOfStreamError from `writeTo` when the stream yields a byte count other than the declared
  * `contentLength` (HTTP-39/BODY-10).
+ * @throws SourceContractViolationError from `writeTo` when the stream delivers a zero-length chunk
+ * without signalling end of stream during a declared-length write (HTTP-39/BODY-10).
  * @public
  */
 export function streamBody(
diff --git a/tests/node-conformance/README.md b/tests/node-conformance/README.md
index b9c62df..ef802e0 100644
--- a/tests/node-conformance/README.md
+++ b/tests/node-conformance/README.md
@@ -51,10 +51,16 @@ Node).
 
 **A phase that touches a runtime-divergent surface adds a case here, not only to `bun test`** (§5.9:378).
 Since Phase 4 that has meant most phases — pipelines, retry, redirect, auth, serde, SSE, pagination,
-configuration, observability, the two concrete transports, and the RxJS bridge all have cases here. Two are
-worth naming as the shape to aim for: 8a's `fetch`/`undici` transports, where this stops being precautionary
-and becomes the point, and 8b's RxJS bridge, whose reason for being hand-written is a cancellation path the
-runtime decides.
+configuration, observability, the two concrete transports, and the RxJS bridge all have cases here — as, since
+audit #67's #77, do the five Web Streams bridges that had none: `BufferedSource.toReadableStream`,
+`BufferedSink.toWritableStream`, `TeeSink.toWritableStream` in `io-byte-stream.test.mjs`, and the
+`withRequestLogging` / `withResponseLogging` body taps in `body-lifecycle.test.mjs`. Every one of them is a
+hand-written underlying source or sink object, so what they exercise is the runtime's own pull scheduling,
+cancel dispatch and reader-lock bookkeeping. Three are worth naming as the shape to aim for: 8a's
+`fetch`/`undici` transports, where this stops being precautionary and becomes the point; 8b's RxJS bridge,
+whose reason for being hand-written is a cancellation path the runtime decides; and #77's multipart
+`Content-Type`, where Bun's `Response.formData()` accepted a header Node's `Response.formData()` rejected —
+the Bun rows were green over a body no Node peer could parse, and only the case here reproduced it.
 
 ## Which cases exist
 
diff --git a/tests/node-conformance/body-lifecycle.test.mjs b/tests/node-conformance/body-lifecycle.test.mjs
index ec37f82..931f5ec 100644
--- a/tests/node-conformance/body-lifecycle.test.mjs
+++ b/tests/node-conformance/body-lifecycle.test.mjs
@@ -24,6 +24,14 @@ import {
   stringBody,
   toHttpError,
 } from '@dexpace/core';
+// The two logging taps are `@internal` -- `body/index.ts` holds them and the public barrel deliberately
+// does not, so they are reached by direct `dist/` file path, exactly as `io-byte-stream.test.mjs` reaches
+// `io/`. Still the BUILT artifact, never `src/`.
+import {withRequestLogging} from '../../packages/core/dist/body/request-body-logging.js';
+import {withResponseLogging} from '../../packages/core/dist/body/response-body-logging.js';
+
+/** `Response` above is the SDK's model class, which shadows the platform global this file also needs. */
+const PlatformResponse = globalThis.Response;
 
 function streamOf(bytes) {
   return new ReadableStream({
@@ -227,3 +235,179 @@ describe('toHttpError buffering on Node', () => {
     assert.deepEqual([...(await response.bytes())], [1, 2, 3]);
   });
 });
+
+describe("the multipart Content-Type parses in Node's own FormData reader (HTTP-51)", () => {
+  // The reproducer for the boundary-quoting fix, and the reason it belongs here rather than only in
+  // `bun test`: Bun's `Response.formData()` tolerates an unquoted `boundary=a,b`, Node's (undici's)
+  // rejects the whole body with `TypeError: Failed to parse body as FormData`. Two independent parsers
+  // disagreeing about a header this SDK generates is precisely what this tree exists to catch.
+  for (const boundary of ['a,b', 'bound ary', 'a:b', 'a=b', 'a?b', '(a)/b']) {
+    it(`round-trips a body framed with ${JSON.stringify(boundary)}`, async () => {
+      const body = multipartBody(
+        [{name: 'field', body: stringBody('value')}],
+        boundary,
+      );
+      const parsed = await new PlatformResponse(await collect(body), {
+        headers: {'content-type': body.mediaType},
+      }).formData();
+      assert.equal(parsed.get('field'), 'value');
+    });
+  }
+
+  it('leaves a boundary that is already a bare token unquoted', () => {
+    const body = multipartBody([{name: 'a', body: stringBody('x')}], 'plain-1');
+    assert.equal(body.mediaType, 'multipart/form-data; boundary=plain-1');
+  });
+});
+
+describe('an exact-length copy refuses a zero-length delivery on Node (HTTP-39/BODY-10)', () => {
+  it('raises rather than forwarding a chunked-encoding terminator to the sink', async () => {
+    const source = new ReadableStream({
+      start(controller) {
+        controller.enqueue(Uint8Array.from([1, 2]));
+        controller.enqueue(new Uint8Array(0));
+        controller.enqueue(Uint8Array.from([3]));
+        controller.close();
+      },
+    });
+    const chunkLengths = [];
+
+    await assert.rejects(
+      streamBody(source, undefined, 3).writeTo(
+        new WritableStream({write: c => void chunkLengths.push(c.length)}),
+      ),
+      error => error.name === 'SourceContractViolationError',
+    );
+    assert.deepEqual(chunkLengths, [2]);
+  });
+
+  it('still allows a declared length of 0 over a source that just closes', async () => {
+    let closed = false;
+    await streamBody(
+      new ReadableStream({
+        start: c => {
+          c.close();
+        },
+      }),
+      undefined,
+      0,
+    ).writeTo(
+      new WritableStream({
+        write: c => void c,
+        close: () => void (closed = true),
+      }),
+    );
+    assert.equal(closed, true);
+  });
+});
+
+describe('withRequestLogging over Node Web Streams (BODY-17..21)', () => {
+  it('mirrors into the tap while the full untruncated payload reaches the primary', async () => {
+    const logged = withRequestLogging(
+      byteArrayBody(Uint8Array.from([1, 2, 3, 4, 5])),
+      2,
+    );
+    assert.deepEqual([...(await collect(logged))], [1, 2, 3, 4, 5]);
+    assert.deepEqual([...logged.snapshot()], [1, 2]);
+  });
+
+  it('clears the tap between writes so a retry does not accumulate stale bytes (BODY-18)', async () => {
+    const logged = withRequestLogging(
+      byteArrayBody(Uint8Array.from([9, 9])),
+      8,
+    );
+    await collect(logged);
+    await collect(logged);
+    assert.deepEqual([...logged.snapshot()], [9, 9]);
+  });
+
+  it('aborts the real sink when the delegate refuses before ever touching the adapter', async () => {
+    // A `ConsumedBodyError` on a second write reaches neither handler on the adapter stream, so without
+    // the wrapper's own catch the primary writer stays open and locked forever -- a held connection.
+    // Whether an abort dispatched on a writer reaches the underlying sink's algorithm, and does so
+    // instead of the close algorithm, is runtime plumbing rather than logic.
+    const logged = withRequestLogging(
+      streamBody(streamOf([1, 2, 3]), undefined, 3),
+      8,
+    );
+    await collect(logged); // consumes the single-use delegate
+
+    let abortReason = 'NOT-ABORTED';
+    let closed = false;
+    const destination = new WritableStream({
+      write: chunk => void chunk,
+      close: () => void (closed = true),
+      abort: reason => void (abortReason = reason),
+    });
+    await assert.rejects(
+      logged.writeTo(destination),
+      error => error.name === 'ConsumedBodyError',
+    );
+    assert.equal(abortReason.name, 'ConsumedBodyError');
+    // Never closed: a broken message must not be committed downstream as a well-formed short one.
+    assert.equal(closed, false);
+  });
+});
+
+describe('withResponseLogging over Node Web Streams (BODY-22..28)', () => {
+  function chunked(...chunks) {
+    return new ReadableStream({
+      start(controller) {
+        for (const chunk of chunks) controller.enqueue(Uint8Array.from(chunk));
+        controller.close();
+      },
+    });
+  }
+
+  async function readAll(stream) {
+    const out = [];
+    for await (const chunk of stream) out.push(...chunk);
+    return out;
+  }
+
+  it('serves the prefix then the still-live tail, one pull at a time (BODY-24)', async () => {
+    const logged = withResponseLogging(chunked([1, 2], [3, 4, 5]), 3);
+    assert.deepEqual(await readAll(await logged.read()), [1, 2, 3, 4, 5]);
+    assert.deepEqual([...logged.snapshot()], [1, 2, 3]);
+  });
+
+  it('cancelling the tail stream cancels the delegate exactly once (BODY-27)', async () => {
+    let cancels = 0;
+    const delegate = chunked([1, 2], [3, 4]);
+    const inner = delegate.cancel.bind(delegate);
+    delegate.cancel = async reason => {
+      cancels += 1;
+      return inner(reason);
+    };
+    const logged = withResponseLogging(delegate, 1);
+
+    await (await logged.read()).cancel();
+    await logged.close();
+    assert.equal(cancels, 1);
+  });
+
+  it('close() leaves the tap inert instead of poisoning it with a detached-reader TypeError', async () => {
+    // Node reports a released reader as `TypeError [ERR_INVALID_STATE]: Invalid state: The reader is not
+    // attached to a stream`. That message used to be cached as this wrapper's drain failure and reported
+    // by error() forever, over a capture that never failed.
+    const logged = withResponseLogging(chunked([1, 2, 3]), 100);
+    await logged.close();
+
+    assert.deepEqual([...logged.snapshot()], []);
+    await new Promise(resolve => setTimeout(resolve, 0));
+    assert.equal(logged.error(), null);
+    await assert.rejects(
+      logged.read(),
+      error => error.name === 'ClosedResourceError',
+    );
+  });
+
+  it('a fits-cap capture stays repeatably readable after close (BODY-23, BODY-28)', async () => {
+    const logged = withResponseLogging(chunked([1, 2, 3]), 100);
+    assert.deepEqual(await readAll(await logged.read()), [1, 2, 3]);
+    await logged.close();
+    assert.deepEqual(await readAll(await logged.read()), [1, 2, 3]);
+    assert.deepEqual([...logged.snapshot()], [1, 2, 3]);
+    assert.equal(logged.error(), null);
+  });
+});
diff --git a/tests/node-conformance/io-byte-stream.test.mjs b/tests/node-conformance/io-byte-stream.test.mjs
index c9165ce..c86f68c 100644
--- a/tests/node-conformance/io-byte-stream.test.mjs
+++ b/tests/node-conformance/io-byte-stream.test.mjs
@@ -224,3 +224,163 @@ describe('BufferedSink and TeeSink over a Node WritableStream', () => {
     );
   });
 });
+
+// The five Web Streams bridges had no case here at all until #77. Every one of them is a hand-written
+// `ReadableStream`/`WritableStream` underlying-source or -sink object, so what they exercise is the
+// runtime's own pull scheduling, cancel dispatch and reader-lock bookkeeping — the three things §5.9
+// names and the three that two independent Streams implementations are most likely to differ on.
+
+describe('BufferedSource.toReadableStream on Node (IO-16)', () => {
+  it('pulls one chunk at a time instead of draining the source eagerly', async () => {
+    let pulls = 0;
+    const stream = new ReadableStream({
+      pull(controller) {
+        pulls += 1;
+        if (pulls > 4) {
+          controller.close();
+          return;
+        }
+        controller.enqueue(Uint8Array.from([pulls]));
+      },
+    });
+    const bridge = BufferedSource.overStream(stream).toReadableStream();
+    const reader = bridge.getReader();
+
+    const first = await reader.read();
+    assert.deepEqual([...first.value], [1]);
+    // Node's default queuing strategy reads one chunk ahead, so at most one pull beyond the one just
+    // served. The assertion that matters is that the whole 4-chunk source has not been materialized.
+    assert.ok(pulls <= 2, `expected at most 2 pulls, saw ${pulls}`);
+    await reader.cancel();
+  });
+
+  it('closes the bridge at natural EOF without tearing down the owning source', async () => {
+    // IO-19: closing the source here would invalidate every outstanding peek/slice view, defeating the
+    // bridge's most natural usage — take a preview, hand the bridge to the transport, read the preview
+    // afterwards. Only an explicit cancel closes the source (next case).
+    const source = BufferedSource.overStream(streamOfChunks([[1, 2], [3]]));
+    const preview = source.peek();
+    const collected = [];
+    for await (const chunk of source.toReadableStream())
+      collected.push(...chunk);
+
+    assert.deepEqual(collected, [1, 2, 3]);
+    assert.deepEqual([...(await preview.readBytes())], [1, 2, 3]);
+    assert.equal(source.closed, false);
+    await source.close();
+  });
+
+  it('cancelling the bridge closes the source AND releases the caller stream lock', async () => {
+    const stream = streamOfChunks([[1, 2, 3]]);
+    const source = BufferedSource.overStream(stream);
+    assert.equal(stream.locked, true);
+
+    await source.toReadableStream().cancel();
+    assert.equal(source.closed, true);
+    // cancel() cancels the stream but never releases the reader's lock; only releaseLock() does, and a
+    // leaked lock on a connection-backed source is a held socket.
+    assert.equal(stream.locked, false);
+  });
+
+  it('a mid-stream read failure closes the source rather than stranding the lock', async () => {
+    // The Streams spec does NOT invoke `cancel` on an errored stream, so the bridge has to close the
+    // source itself on this path. A runtime that dispatched cancel here would hide the bug.
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(Uint8Array.from([1, 2, 3]));
+      },
+      pull() {
+        throw new Error('mid-stream read failure');
+      },
+    });
+    const source = BufferedSource.overStream(stream);
+    const reader = source.toReadableStream().getReader();
+
+    assert.deepEqual([...(await reader.read()).value], [1, 2, 3]);
+    await assert.rejects(reader.read(), /mid-stream read failure/);
+    assert.equal(source.closed, true);
+    assert.equal(stream.locked, false);
+  });
+});
+
+describe('BufferedSink.toWritableStream on Node (IO-16)', () => {
+  it('carries a pipeTo through to the destination and closes it', async () => {
+    // `pipeTo` closes its destination on natural EOF, and IO-16 says closing the bridge closes the
+    // sink, which closes the caller's stream. Three closes chained through two runtimes' plumbing.
+    const written = [];
+    let closed = false;
+    const destination = new WritableStream({
+      write: chunk => void written.push(...chunk),
+      close: () => void (closed = true),
+    });
+    const sink = BufferedSink.overStream(destination);
+
+    await streamOfChunks([[1, 2], [3]]).pipeTo(sink.toWritableStream());
+    assert.deepEqual(written, [1, 2, 3]);
+    assert.equal(sink.closed, true);
+    assert.equal(closed, true);
+  });
+
+  it('aborting the bridge aborts the sink and carries the reason, rather than closing it', async () => {
+    // Collapsing an abort into a graceful close commits a cancelled request body downstream as a
+    // well-formed complete one, so the peer cannot tell an aborted upload from a successful short one.
+    let closed = false;
+    let abortReason = 'NOT-ABORTED';
+    const destination = new WritableStream({
+      write: chunk => void chunk,
+      close: () => void (closed = true),
+      abort: reason => void (abortReason = reason),
+    });
+    const sink = BufferedSink.overStream(destination);
+    const writer = sink.toWritableStream().getWriter();
+    await writer.write(Uint8Array.from([1, 2, 3]));
+
+    const reason = new Error('user cancelled');
+    await writer.abort(reason);
+    assert.equal(abortReason, reason);
+    assert.equal(closed, false);
+    assert.equal(sink.closed, true);
+  });
+
+  it('drops a zero-length chunk rather than forwarding a chunked-encoding terminator', async () => {
+    const {stream, written} = collectingStream();
+    const sink = BufferedSink.overStream(stream);
+    const writer = sink.toWritableStream().getWriter();
+    await writer.write(new Uint8Array(0));
+    await writer.write(Uint8Array.from([7]));
+    await writer.close();
+    assert.deepEqual([...written()], [7]);
+  });
+});
+
+describe('TeeSink.toWritableStream on Node (IO-16, IO-26)', () => {
+  it('routes through the tee, so bytes written to the bridge still reach the tap', async () => {
+    // Handing callers the PRIMARY's bridge instead would let every byte written through it bypass the
+    // tap, silently producing an empty capture.
+    const {stream, written} = collectingStream();
+    const tee = new TeeSink(BufferedSink.overStream(stream), 2);
+
+    await streamOfChunks([
+      [1, 2],
+      [3, 4, 5],
+    ]).pipeTo(tee.toWritableStream());
+    assert.deepEqual([...written()], [1, 2, 3, 4, 5]);
+    assert.deepEqual([...tee.snapshot()], [1, 2]);
+  });
+
+  it('forwards an abort to the primary while the tap survives to record what was attempted', async () => {
+    let abortReason = 'NOT-ABORTED';
+    const destination = new WritableStream({
+      write: chunk => void chunk,
+      abort: reason => void (abortReason = reason),
+    });
+    const tee = new TeeSink(BufferedSink.overStream(destination), 4);
+    const writer = tee.toWritableStream().getWriter();
+    await writer.write(Uint8Array.from([1, 2, 3]));
+
+    const reason = new Error('deadline exceeded');
+    await writer.abort(reason);
+    assert.equal(abortReason, reason);
+    assert.deepEqual([...tee.snapshot()], [1, 2, 3]);
+  });
+});