Encode string chunks as UTF-8 when pumping TypeScript streams

src/tests/streams/cache/AGENTS.md

@@ -21,6 +21,7 @@ the record back through its MOCK service binding (/last-put).
 | Test | Shape |
 | --- | --- |
 | `putJsValueStreamBody` | value stream of Uint8Array chunks, byte-exact at the backend |
+| `putStringChunkBody` | a string chunk stored as its UTF-8 encoding beside a byte chunk, byte-exact at the backend |
 | `putJsByteStreamBody` | 64 KiB chunked byte stream |
 | `putFixedLengthStreamBody` | FixedLengthStream body; the declared length arrives as a concrete Content-Length in the serialized head |
 | `putIdentityStreamBody` | identity body fed by a concurrent writer |

src/tests/streams/cache/cache-streams.js

@@ -57,6 +57,29 @@ export const putJsValueStreamBody = {
   },
 };
 
+// A string chunk is stored as its UTF-8 encoding beside the byte chunk
+// after it: the pattern's first 100 bytes are ASCII, so the string of
+// those characters encodes to exactly them.
+export const putStringChunkBody = {
+  async test(ctrl, env) {
+    const ascii = String.fromCharCode(...patternChunk(0, 100));
+    const rs = new ReadableStream({
+      start(c) {
+        c.enqueue(ascii);
+        c.enqueue(patternChunk(100, 100));
+        c.close();
+      },
+    });
+    await caches.default.put(
+      'https://example.com/string-chunk',
+      new Response(rs)
+    );
+    const record = await lastPut(env);
+    strictEqual(record.byteLength, 200);
+    strictEqual(record.patternOk, true);
+  },
+};
+
 // A JS byte stream, chunked.
 export const putJsByteStreamBody = {
   async test(ctrl, env) {

src/tests/streams/cache/main.js

@@ -7,6 +7,7 @@
 
 export {
   putJsValueStreamBody,
+  putStringChunkBody,
   putJsByteStreamBody,
   putFixedLengthStreamBody,
   putIdentityStreamBody,

src/tests/streams/htmlrewriter/AGENTS.md

@@ -19,6 +19,7 @@ under the MODERN `fixup-transform-stream-backpressure`.
 | `handlerAcrossChunkBoundaries` | chunks split MID-TAG; the parser reassembles and the handler mutates both elements |
 | `rewrittenBodyIsReadableStream` | output body drained incrementally via a reader |
 | `contentFromReadableStream` | element.replace(ReadableStream) — streamed replacement content |
+| `stringChunksAreUtf8` | string chunks in the transformed body and in streamed replacement content read as UTF-8 beside byte chunks (parity) |
 | `identityStreamBody` | identity body fed by a concurrent writer |
 | `cancelReachesSourceAfterNextChunk` | cancel remains pending at the source until another chunk wakes the pump, then reaches the source |
 | `erroringSourceRejectsConsumption` | source error surfaces from .text() |

src/tests/streams/htmlrewriter/main.js

@@ -11,6 +11,7 @@ export {
   handlerAcrossChunkBoundaries,
   rewrittenBodyIsReadableStream,
   contentFromReadableStream,
+  stringChunksAreUtf8,
   identityStreamBody,
   cancelReachesSourceAfterNextChunk,
   erroringSourceRejectsConsumption,

src/tests/streams/htmlrewriter/rewrite-streams.js

@@ -125,6 +125,41 @@ export const contentFromReadableStream = {
   },
 };
 
+// String chunks, in the transformed body and in streamed replacement
+// content, are read as their UTF-8 encoding beside the byte chunks
+// around them (parity).
+export const stringChunksAreUtf8 = {
+  async test() {
+    const mixed = (first, second) =>
+      new ReadableStream({
+        start(c) {
+          c.enqueue(enc.encode(first));
+          c.enqueue(second);
+          c.close();
+        },
+      });
+
+    strictEqual(
+      await new HTMLRewriter()
+        .transform(new Response(mixed('<p>caf', '\u00e9</p>')))
+        .text(),
+      '<p>caf\u00e9</p>'
+    );
+
+    strictEqual(
+      await new HTMLRewriter()
+        .on('div', {
+          element(element) {
+            element.replace(mixed('<b>new', '</b>'), { html: true });
+          },
+        })
+        .transform(new Response('<div>old</div>'))
+        .text(),
+      '<b>new</b>'
+    );
+  },
+};
+
 // An IdentityTransformStream body fed by a concurrent writer.
 export const identityStreamBody = {
   async test() {

src/tests/streams/readable/AGENTS.md

@@ -106,6 +106,7 @@ C++ implementation; `draining-reader.js` asserts both sides.
 | `streams_enable_constructors` (2022-11-30) | yes (the suite's subject) | `readable-cpp-legacy` cell: ctor and .from() throw plain Errors naming the flag; ReadableStreamDefaultController not exposed; native-backed streams (Response bodies) fully usable incl. tee() |
 | `transformstream_enable_standard_constructor` | yes (tee-reentrancy drives tee through TransformStreams) | transform suite's legacy cells |
 | `capture_async_api_throws`, `workers_api_getters_setters_on_prototype`, `set_tostring_tag`, `fixup-transform-stream-backpressure`, `writable_stream_spec_compliant_writer` | yes | other suites' legacy cells |
+| `fetch_iterable_type_support` (2026-02-19) | yes (string-chunks.js sends an async generator as a request body) | not exercised here: iterable bodies are fetch's surface, not this suite's |
 | `pedantic_wpt` (dateless opt-in) | `readable-cpp-pedantic` cell | zero observable deltas on this suite's probed surface (cpp == pedantic in every probe) |
 
 ## Module map
@@ -138,6 +139,7 @@ C++ implementation; `draining-reader.js` asserts both sides.
 | `data-volumes.js` | value-stream volume axes: 4096-chunk counts, a 1 MiB single string chunk, 8 MiB aggregate (128 × 64 KiB), and 1 MiB through tee on both branches — every chunk index-encoded |
 | `pollution.js` | prototype pollution neither implementation observes: a patched array iterator or replaced Number global leaves bodies intact; a patched controller error() still errors the stream; omitted dictionaries read nothing from Object.prototype, nor does from() for the source and strategy it builds; @@asyncIterator shape (TS: the values() function object, per WebIDL; C++: a separate function) and for-await ignoring a patched values() |
 | `unhandled-rejection.js` | a reader's closed promise is marked handled where it is made (on a readable stream, on an errored stream; replaced at release from a closed stream): rejected with nothing looking at it, it reports no unhandledrejection (parity) |
+| `string-chunks.js` | string chunks in a body the runtime sends on (a fetch request body, in one batch and one chunk per pull; an async-generator request body; a response the fetch handler returns, served by SELF at `/strings`) go out as their UTF-8 encoding beside the byte chunks, lone surrogates replaced with U+FFFD and embedded NULs kept (parity; a body read in JavaScript rejects them, `integration-body.js`) |
 
 Consumed sources: streams-async-iterator-test.js (deleted),
 streams-tee-edge-cases-test.js (value half), streams-test.js (from/

src/tests/streams/readable/main.js

@@ -9,9 +9,11 @@
 // tests: it echoes the request body back, letting tests round-trip a
 // ReadableStream through the fetch machinery in both directions, and
 // reports the Content-Length the request carried. /gzip and /short serve
-// expected-length.js's compressed and short responses instead.
+// expected-length.js's compressed and short responses, and /strings
+// string-chunks.js's response, instead.
 
 import { compressedResponse, shortResponse } from 'expected-length';
+import { stringsResponse } from 'string-chunks';
 
 export default {
   async fetch(request) {
@@ -22,6 +24,9 @@ export default {
     if (url.pathname === '/short') {
       return shortResponse(url);
     }
+    if (url.pathname === '/strings') {
+      return stringsResponse();
+    }
     return new Response(request.body, {
       headers: {
         'content-type': 'application/octet-stream',
@@ -307,3 +312,9 @@ export {
 } from 'pollution';
 
 export { readerClosedIsMarkedHandled } from 'unhandled-rejection';
+
+export {
+  stringChunksInRequestBody,
+  stringChunksInIterableRequestBody,
+  stringChunksInReturnedResponse,
+} from 'string-chunks';

src/tests/streams/readable/readable-cpp-pedantic.wd-test

@@ -21,6 +21,7 @@ const unitTests :Workerd.Config = (
           "set_tostring_tag",
           "writable_stream_spec_compliant_writer",
           "pedantic_wpt",
+          "fetch_iterable_type_support",
           "enable_weak_ref",
         ],
         bindings = [

src/tests/streams/readable/readable-cpp.wd-test

@@ -20,6 +20,8 @@ using ReadableModules = import "readable-modules.capnp";
 # - set_tostring_tag: Symbol.toStringTag observed by api-surface tests.
 # - writable_stream_spec_compliant_writer: writer behavior in the
 #   integration tests' pipe interactions.
+# - fetch_iterable_type_support: string-chunks.js sends an async
+#   generator as a request body.
 const unitTests :Workerd.Config = (
   services = [
     ( name = "streams-readable-cpp",
@@ -34,6 +36,7 @@ const unitTests :Workerd.Config = (
           "workers_api_getters_setters_on_prototype",
           "set_tostring_tag",
           "writable_stream_spec_compliant_writer",
+          "fetch_iterable_type_support",
           "enable_weak_ref",
         ],
         bindings = [

src/tests/streams/readable/readable-modules.capnp

@@ -35,4 +35,5 @@ const modules :List(Workerd.Worker.Module) = [
   (name = "transfer", esModule = embed "transfer.js"),
   (name = "pollution", esModule = embed "pollution.js"),
   (name = "unhandled-rejection", esModule = embed "unhandled-rejection.js"),
+  (name = "string-chunks", esModule = embed "string-chunks.js"),
 ];

src/tests/streams/readable/readable-ts.wd-test

@@ -15,6 +15,7 @@ const unitTests :Workerd.Config = (
           "typescript_implemented_streams",
           "expose_draining_reader",
           "experimental",
+          "fetch_iterable_type_support",
           "enable_weak_ref",
         ],
         bindings = [

src/tests/streams/readable/string-chunks.js

@@ -0,0 +1,92 @@
+// Copyright (c) 2026 Cloudflare, Inc.
+// Licensed under the Apache 2.0 license found in the LICENSE file or at:
+//     https://opensource.org/licenses/Apache-2.0
+
+// String chunks in a body the runtime sends on: a fetch request body, an
+// iterable request body, and a response the fetch handler returns,
+// observed through the SELF service.
+//
+// PARITY: a string chunk is sent as its UTF-8 encoding, lone surrogates
+// replaced with U+FFFD and embedded NULs kept, alongside the byte chunks
+// around it. Strings are not valid body chunks (a body read in JavaScript,
+// text() and the like, rejects them in both; integration-body.js), but
+// the C++ implementation has always encoded them when it sends a body on,
+// and code such as server-sent-event handlers relies on it.
+
+import { deepStrictEqual, strictEqual } from 'node:assert';
+
+const enc = new TextEncoder();
+
+// Bytes and strings in one stream, with the UTF-8 the runtime sends for
+// them.
+const MIXED_CHUNKS = [enc.encode('ab'), 'cd', '', '\uD800', 'x\0y'];
+const MIXED_BYTES = [
+  ...enc.encode('abcd'),
+  0xef,
+  0xbf,
+  0xbd,
+  ...enc.encode('x\0y'),
+];
+
+function streamOf(chunks) {
+  return new ReadableStream({
+    start(c) {
+      for (const chunk of chunks) c.enqueue(chunk);
+      c.close();
+    },
+  });
+}
+
+async function echoedBytes(env, body) {
+  const resp = await env.SELF.fetch('http://example.org', {
+    method: 'POST',
+    body,
+  });
+  return [...new Uint8Array(await resp.arrayBuffer())];
+}
+
+// Served by main.js at /strings: an event stream whose second event is a
+// string chunk.
+export function stringsResponse() {
+  return new Response(
+    streamOf([enc.encode('data: one\n\n'), 'data: two\n\n']),
+    { headers: { 'content-type': 'text/event-stream' } }
+  );
+}
+
+export const stringChunksInRequestBody = {
+  async test(ctrl, env) {
+    deepStrictEqual(
+      await echoedBytes(env, streamOf(MIXED_CHUNKS)),
+      MIXED_BYTES
+    );
+
+    // One chunk per pull, so the strings reach the pump in batches of
+    // their own.
+    let i = 0;
+    const pulled = new ReadableStream({
+      pull(c) {
+        if (i < MIXED_CHUNKS.length) c.enqueue(MIXED_CHUNKS[i++]);
+        else c.close();
+      },
+    });
+    deepStrictEqual(await echoedBytes(env, pulled), MIXED_BYTES);
+  },
+};
+
+export const stringChunksInIterableRequestBody = {
+  async test(ctrl, env) {
+    async function* body() {
+      yield enc.encode('ab');
+      yield 'cd';
+    }
+    deepStrictEqual(await echoedBytes(env, body()), [...enc.encode('abcd')]);
+  },
+};
+
+export const stringChunksInReturnedResponse = {
+  async test(ctrl, env) {
+    const resp = await env.SELF.fetch('http://example.org/strings');
+    strictEqual(await resp.text(), 'data: one\n\ndata: two\n\n');
+  },
+};

src/workerd/api/js-readable-stream-test.c++

@@ -1909,7 +1909,7 @@ KJ_TEST("JsReadableStream pumpTo rejects when a queued stream produces non-byte
             js.v8Context(), [](jsg::Lock& js, const v8::FunctionCallbackInfo<v8::Value>& info) {
       auto controller = KJ_ASSERT_NONNULL(jsg::JsValue(info[0]).tryCast<jsg::JsObject>());
       auto enqueue = KJ_ASSERT_NONNULL(controller.get(js, "enqueue"_kj).tryCast<jsg::JsFunction>());
-      enqueue.call(js, controller, js.str("not bytes"_kj));
+      enqueue.call(js, controller, js.num(42));
     })));
 
     auto stream = makeTsStream(js, underlying);
@@ -2080,6 +2080,35 @@ KJ_TEST("JsReadableStream pumpTo of a queued TypeScript stream dropped in IoCont
   KJ_EXPECT(!ended);
 }
 
+KJ_TEST("JsReadableStream pumpTo writes a queued stream's string chunks as UTF-8") {
+  auto fixture = makeTsStreamsFixture();
+  kj::Vector<kj::byte> collected;
+  bool ended = false;
+  fixture.runInIoContext([&](const TestFixture::Environment& env) -> kj::Promise<void> {
+    auto& js = env.js;
+
+    // Bytes, then strings, in one batch, as the legacy pump's valueToBytes() writes them.
+    auto underlying = js.obj();
+    underlying.set(js, "start"_kj,
+        jsg::JsValue(js.wrapSimpleFunction(
+            js.v8Context(), [](jsg::Lock& js, const v8::FunctionCallbackInfo<v8::Value>& info) {
+      auto controller = KJ_ASSERT_NONNULL(jsg::JsValue(info[0]).tryCast<jsg::JsObject>());
+      auto enqueue = KJ_ASSERT_NONNULL(controller.get(js, "enqueue"_kj).tryCast<jsg::JsFunction>());
+      enqueue.call(js, controller, jsg::JsUint8Array::create(js, "ab"_kjb));
+      enqueue.call(js, controller, js.str("cd"_kj));
+      enqueue.call(js, controller, js.str("\xc3\xa9"_kj));
+      auto close = KJ_ASSERT_NONNULL(controller.get(js, "close"_kj).tryCast<jsg::JsFunction>());
+      close.call(js, controller);
+    })));
+
+    auto stream = makeTsStream(js, underlying);
+    return stream.pumpTo(js, kj::heap<CollectingSink>(collected, ended), EndStream::YES)
+        .then([](DeferredProxy<void> proxy) { return kj::mv(proxy.proxyTask); });
+  });
+  KJ_EXPECT(collected.asPtr() == "abcd\xc3\xa9"_kjb);
+  KJ_EXPECT(ended);
+}
+
 KJ_TEST("JsReadableStream text() over the limit cancels a queued TypeScript stream") {
   auto fixture = makeTsStreamsFixture();
   kj::Maybe<kj::String> cancelReason;

src/workerd/api/js-readable-stream.c++

@@ -545,16 +545,19 @@ jsg::Promise<void> queuedPumpStep(jsg::Lock& js, kj::Rc<QueuedPumpState> state,
         "ReadableStreamDrainingReader.read() promise did not resolve to an object "
         "with a chunks array.");
 
-    // Only ArrayBuffer/ArrayBufferView chunks are usable as bytes. Everything else
-    // (including strings) rejects, matching the legacy pump's historical behavior and
-    // error text exactly.
+    // ArrayBuffer/ArrayBufferView chunks are written as their bytes, and strings as their
+    // UTF-8 encoding (lone surrogates replaced with U+FFFD), as the legacy pump's
+    // valueToBytes() writes them. Everything else rejects. JS body consumption (text() and
+    // the like) rejects strings in both implementations; only pumps encode them.
     auto pieces = kj::Vector<kj::Array<const kj::byte>>(chunks.size());
     for (uint32_t i = 0; i < chunks.size(); i++) {
       auto chunk = chunks.get(js, i);
       KJ_IF_SOME(view, JSG_TRY_CAST(chunk, JsArrayBufferView)) {
         pieces.add(jsg::JsBufferSource(view).copy());
       } else KJ_IF_SOME(buffer, JSG_TRY_CAST_ARRAYBUFFER(chunk)) {
         pieces.add(jsg::JsBufferSource(buffer).copy());
+      } else KJ_IF_SOME(str, chunk.tryCast<jsg::JsString>()) {
+        pieces.add(webstreams::stringToBytes(str.toUSVString(js)));
       } else {
         JSG_FAIL_REQUIRE(TypeError, "This ReadableStream did not return bytes.");
       }

src/workerd/api/tests/BUILD.bazel

@@ -753,6 +753,12 @@ wd_test(
     data = ["eventsource-test.js"],
 )
 
+wd_test(
+    src = "eventsource-ts-test.wd-test",
+    args = ["--experimental"],
+    data = ["eventsource-test.js"],
+)
+
 wd_test(
     src = "legacy-dispatch-exceptions-test.wd-test",
     args = ["--experimental"],

src/workerd/api/tests/eventsource-test.js

@@ -357,6 +357,31 @@ export const eventSourceFromTest = {
   },
 };
 
+// String chunks are read as their UTF-8 encoding, beside byte chunks.
+export const eventSourceFromStringChunksTest = {
+  async test() {
+    const rs = new ReadableStream({
+      start(c) {
+        c.enqueue(new TextEncoder().encode('data: first\n\n'));
+        c.enqueue('data: caf\u00e9\n\n');
+        c.close();
+      },
+    });
+    const { promise, resolve } = Promise.withResolvers();
+    const eventsource = EventSource.from(rs);
+    const events = [];
+    eventsource.onmessage = (event) => {
+      events.push(event.data);
+      if (events.length === 2) {
+        eventsource.close();
+        resolve();
+      }
+    };
+    await promise;
+    deepStrictEqual(events, ['first', 'caf\u00e9']);
+  },
+};
+
 export const eventSourceFromWithBOMTest = {
   async test() {
     const enc = new TextEncoder();