diff --git a/src/runtime/api/html_rewriter.rs b/src/runtime/api/html_rewriter.rs
index 0b70f34d3585..9a143e99ae3e 100644
--- a/src/runtime/api/html_rewriter.rs
+++ b/src/runtime/api/html_rewriter.rs
@@ -1252,6 +1252,22 @@ impl RewriterPipe {
         Writable::Owned(len)
     }
 
+    pub fn write_latin1(&self, data: &StreamResult) -> Writable {
+        let bytes = data.slice();
+        if bun_core::strings::is_all_ascii(bytes) {
+            return self.write(data);
+        }
+        let mut buf = Vec::new();
+        let _ = buf.write_latin1(bytes);
+        self.write(&StreamResult::Temporary(RawSlice::new(&buf)))
+    }
+
+    pub fn write_utf16(&self, data: &StreamResult) -> Writable {
+        let mut buf = Vec::new();
+        let _ = buf.write_utf16(data.slice16());
+        self.write(&StreamResult::Temporary(RawSlice::new(&buf)))
+    }
+
     /// `SinkHandle::end` entry — input EOF or terminal upstream error.
     pub fn end_from_stream(&self, err: Option<StreamError>) {
         // Detach via `detach_input_source` (not a bare `.set(None)`) so a
@@ -1632,18 +1648,10 @@ impl crate::webcore::sink::JsSinkType for RewriterPipe {
         RewriterPipe::write(self, data)
     }
     fn write_utf16(&mut self, data: &StreamResult) -> Writable {
-        let mut buf = Vec::new();
-        let _ = buf.write_utf16(data.slice16());
-        RewriterPipe::write(self, &StreamResult::Temporary(RawSlice::new(&buf)))
+        RewriterPipe::write_utf16(self, data)
     }
     fn write_latin1(&mut self, data: &StreamResult) -> Writable {
-        let bytes = data.slice();
-        if bun_core::strings::is_all_ascii(bytes) {
-            return RewriterPipe::write(self, data);
-        }
-        let mut buf = Vec::new();
-        let _ = buf.write_latin1(bytes);
-        RewriterPipe::write(self, &StreamResult::Temporary(RawSlice::new(&buf)))
+        RewriterPipe::write_latin1(self, data)
     }
     fn end(&mut self, err: Option<SysError>) -> bun_sys::Result<()> {
         self.end_from_stream(err.map(StreamError::Error));
diff --git a/src/runtime/webcore.rs b/src/runtime/webcore.rs
index 9aff312bc5b3..5013dfaf289f 100644
--- a/src/runtime/webcore.rs
+++ b/src/runtime/webcore.rs
@@ -406,6 +406,68 @@ impl SinkHandle {
         }
     }
 
+    /// Route a Latin-1 chunk to the sink's own Latin-1→UTF-8 writer so the
+    /// conversion lands directly in the sink's buffer. Every sink's
+    /// `write_latin1` has an `is_all_ascii` fast path that forwards the input
+    /// slice verbatim, so an all-ASCII `Temporary` here is zero-copy end to end.
+    ///
+    /// `ServerResponse` is unreachable: only `sink::sink_handle_from_id` (a
+    /// native transform attached to a JSSink) produces callers of this path.
+    ///
+    /// SAFETY: same pointee-liveness invariant as [`Self::write`].
+    pub fn write_latin1(&self, data: &streams::Result) -> streams::Writable {
+        match *self {
+            SinkHandle::None => streams::Writable::Done,
+            SinkHandle::ServerResponse(_) => {
+                debug_assert!(false, "ServerResponse is never a JSSink SinkID");
+                streams::Writable::Done
+            }
+            // SAFETY: live backref; ByteStream clears sink before free.
+            SinkHandle::FetchRequestBody(mut p) => unsafe { p.get_mut() }.write_latin1(data),
+            // SAFETY: live backref; ByteStream clears sink before free.
+            SinkHandle::S3Upload(mut p) => unsafe { p.get_mut() }.write_latin1(data),
+            SinkHandle::FileSink(p) => p.write_latin1(data),
+            SinkHandle::HTMLRewriter(p) => p.write_latin1(data),
+            // SAFETY: live backref; transform detaches before the JSSink is finalized.
+            SinkHandle::HttpResponse(mut p) => unsafe { p.get_mut() }.write_latin1(data),
+            // SAFETY: live backref; transform detaches before the JSSink is finalized.
+            SinkHandle::HttpsResponse(mut p) => unsafe { p.get_mut() }.write_latin1(data),
+            // SAFETY: live backref; transform detaches before the JSSink is finalized.
+            SinkHandle::H3Response(mut p) => unsafe { p.get_mut() }.write_latin1(data),
+            // SAFETY: live backref; transform detaches before the JSSink is finalized.
+            SinkHandle::ArrayBuffer(mut p) => unsafe { p.get_mut() }.write_latin1(data),
+        }
+    }
+
+    /// As [`Self::write_latin1`] but for a UTF-16 chunk (`data.slice16()`).
+    /// Every sink's `write_utf16` emits U+FFFD for unpaired surrogates, so the
+    /// caller is responsible for any cross-chunk surrogate carry.
+    ///
+    /// SAFETY: same pointee-liveness invariant as [`Self::write`].
+    pub fn write_utf16(&self, data: &streams::Result) -> streams::Writable {
+        match *self {
+            SinkHandle::None => streams::Writable::Done,
+            SinkHandle::ServerResponse(_) => {
+                debug_assert!(false, "ServerResponse is never a JSSink SinkID");
+                streams::Writable::Done
+            }
+            // SAFETY: live backref; ByteStream clears sink before free.
+            SinkHandle::FetchRequestBody(mut p) => unsafe { p.get_mut() }.write_utf16(data),
+            // SAFETY: live backref; ByteStream clears sink before free.
+            SinkHandle::S3Upload(mut p) => unsafe { p.get_mut() }.write_utf16(data),
+            SinkHandle::FileSink(p) => p.write_utf16(data),
+            SinkHandle::HTMLRewriter(p) => p.write_utf16(data),
+            // SAFETY: live backref; transform detaches before the JSSink is finalized.
+            SinkHandle::HttpResponse(mut p) => unsafe { p.get_mut() }.write_utf16(data),
+            // SAFETY: live backref; transform detaches before the JSSink is finalized.
+            SinkHandle::HttpsResponse(mut p) => unsafe { p.get_mut() }.write_utf16(data),
+            // SAFETY: live backref; transform detaches before the JSSink is finalized.
+            SinkHandle::H3Response(mut p) => unsafe { p.get_mut() }.write_utf16(data),
+            // SAFETY: live backref; transform detaches before the JSSink is finalized.
+            SinkHandle::ArrayBuffer(mut p) => unsafe { p.get_mut() }.write_utf16(data),
+        }
+    }
+
     /// Signal end-of-stream (or terminal error) to the attached sink.
     ///
     /// SAFETY: same pointee-liveness invariant as [`Self::write`].
diff --git a/src/runtime/webcore/TextEncoderStreamEncoder.rs b/src/runtime/webcore/TextEncoderStreamEncoder.rs
index b19cc103dc23..1ad84ce67c0c 100644
--- a/src/runtime/webcore/TextEncoderStreamEncoder.rs
+++ b/src/runtime/webcore/TextEncoderStreamEncoder.rs
@@ -263,13 +263,19 @@ pub extern "C" fn TextEncoderStreamEncoder__flushForStream(
 /// that much memory for the life of the encoder.
 const SCRATCH_CAP: usize = 64 * 1024;
 
-/// Native-sink transform step: encodes `chunk` into the encoder's reusable
-/// scratch buffer and writes it straight to the sink, so a
-/// `ByteStream → TextEncoderStream → JSSink` chain allocates no
-/// `JSUint8Array` per chunk. Returns the sink's `write_bytes` result (see
-/// nativeSinkWriteIsBackpressure for the backpressure-signal shapes),
-/// `undefined` for an empty output, or `JSValue::zero` with the exception
-/// pending on `global`.
+/// Native-sink transform step: hands the chunk's `WTFStringImpl` bytes
+/// straight to the sink via [`SinkHandle::write_latin1`] /
+/// [`SinkHandle::write_utf16`], so a `ByteStream → TextEncoderStream → JSSink`
+/// chain does the Latin-1/UTF-16 → UTF-8 conversion directly into the sink's
+/// own buffer with no per-chunk scratch allocation (and zero-copy for an
+/// all-ASCII 8-bit chunk — every sink's `write_latin1` has that fast path).
+/// A carried lead surrogate from the previous chunk is the only case that
+/// still needs a scratch buffer (the replacement / combined astral prefix
+/// must be part of the same sink write so only one `Writable` is produced).
+///
+/// Returns the sink's write result (see `nativeSinkWriteIsBackpressure` for
+/// the backpressure-signal shapes), `undefined` for an empty output, or
+/// `JSValue::zero` with the exception pending on `global`.
 #[unsafe(no_mangle)]
 #[allow(clippy::not_unsafe_ptr_arg_deref)]
 pub extern "C" fn TextEncoderStreamEncoder__encodeIntoSink(
@@ -279,14 +285,65 @@ pub extern "C" fn TextEncoderStreamEncoder__encodeIntoSink(
     sink_id: u8,
     sink_ptr: *mut core::ffi::c_void,
 ) -> JSValue {
-    let Ok(str) = chunk.get_zig_string(global) else {
+    // WebIDL DOMString coercion runs user JS; hold the resulting JSString cell
+    // so its `WTFStringImpl` survives the sink write even if the sink itself
+    // runs user JS (HTMLRewriter content handlers) or reaches a GC point
+    // (ArrayBufferSink's onReady) — same pattern as `JSSink::js_write`.
+    let Ok(js_str) = chunk.to_js_string(global) else {
         return JSValue::ZERO;
     };
-    // SAFETY: `this` is the live encoder owned by the calling JS cell; taken
-    // after the coercion so no user JS runs while the borrow is live.
+    let str = js_str.view(global);
+    let _keep = bun_jsc::EnsureStillAlive(js_str.to_js());
+    // SAFETY: `this` is the live encoder owned by the calling JS cell (on the
+    // C++ caller's stack, so it cannot be finalized mid-call). The sink write
+    // below can run user JS, but re-entry only forms another shared `&*this`;
+    // all state is `Cell`/`RefCell` and `scratch` is moved out before the write.
     let this = unsafe { &*this };
-    // Move the Vec out of the RefCell for the duration of the sink write so a
-    // (theoretical) re-entrant encode-into-sink call cannot BorrowMut-panic.
+    let Some(ptr) = NonNull::new(sink_ptr) else {
+        return JSValue::UNDEFINED;
+    };
+    // SAFETY: `sink_ptr` is a live JSSink of type `sink_id` (the C++ caller
+    // null-checks it before attaching); the sink copies what it needs.
+    let handle = unsafe { sink_handle_from_id(sink_id, ptr) };
+    if handle.is_none() {
+        return JSValue::UNDEFINED;
+    }
+
+    if this.pending_lead_surrogate.get().is_none() {
+        if !str.is_16bit() {
+            let bytes = str.slice();
+            if bytes.is_empty() {
+                return JSValue::UNDEFINED;
+            }
+            return handle
+                .write_latin1(&streams::Result::Temporary(RawSlice::new(bytes)))
+                .to_js(global);
+        }
+
+        let mut utf16 = str.utf16_slice_aligned();
+        // The sink's `write_utf16` emits U+FFFD for every unpaired surrogate
+        // (matching this encoder for mid-chunk ones); a trailing lone lead is
+        // the one position where the encoder differs — it carries it to the
+        // next chunk instead.
+        if let Some(&last) = utf16.last() {
+            if strings::u16_is_lead(last) {
+                this.pending_lead_surrogate.set(Some(last));
+                utf16 = &utf16[..utf16.len() - 1];
+            }
+        }
+        if utf16.is_empty() {
+            return JSValue::UNDEFINED;
+        }
+        let bytes: &[u8] = bytemuck::cast_slice(utf16);
+        return handle
+            .write_utf16(&streams::Result::Temporary(RawSlice::new(bytes)))
+            .to_js(global);
+    }
+
+    // A lead surrogate was carried from the previous chunk: the prefix and
+    // body go out as one write so only one `Writable` is produced. Move the
+    // Vec out of the RefCell for the duration so a (theoretical) re-entrant
+    // encode-into-sink call cannot BorrowMut-panic.
     let mut buf = this.scratch.take();
     buf.clear();
     if str.is_16bit() {
@@ -303,17 +360,6 @@ pub extern "C" fn TextEncoderStreamEncoder__encodeIntoSink(
         this.scratch.replace(buf);
         return JSValue::UNDEFINED;
     }
-    let Some(ptr) = NonNull::new(sink_ptr) else {
-        this.scratch.replace(buf);
-        return JSValue::UNDEFINED;
-    };
-    // SAFETY: `sink_ptr` is a live JSSink of type `sink_id` (the C++ caller
-    // null-checks it before attaching); the sink copies what it needs.
-    let handle = unsafe { sink_handle_from_id(sink_id, ptr) };
-    if handle.is_none() {
-        this.scratch.replace(buf);
-        return JSValue::UNDEFINED;
-    }
     let wrote = handle
         .write(&streams::Result::Temporary(RawSlice::new(&buf)))
         .to_js(global);
diff --git a/test/js/web/encoding/text-decoder-stream.test.ts b/test/js/web/encoding/text-decoder-stream.test.ts
index cf603bfc031d..b056c2eef8d2 100644
--- a/test/js/web/encoding/text-decoder-stream.test.ts
+++ b/test/js/web/encoding/text-decoder-stream.test.ts
@@ -264,6 +264,32 @@ test("utf-8 fatal: truncated sequence rejects at flush", async () => {
   await expect(Bun.readableStreamToArray(out)).rejects.toBeInstanceOf(TypeError);
 });
 
+// Readable-side backpressure: the readable queue's HWM is 1, so a single write
+// completes without a reader, and a second write parks until the readable side
+// is drained. Exercises both the utf-8 fast path and the Rust-decoder path.
+for (const opts of [{}, { fatal: true }] as const) {
+  test(`TextDecoderStream readable-side backpressure (fatal=${!!opts.fatal}): second write stays pending until drained`, async () => {
+    const tds = new TextDecoderStream("utf-8", opts);
+    const writer = tds.writable.getWriter();
+    const reader = tds.readable.getReader();
+
+    await writer.write(new TextEncoder().encode("first"));
+    const second = writer.write(new TextEncoder().encode("second"));
+    expect(Bun.peek.status(second)).toBe("pending");
+
+    const { value } = await reader.read();
+    expect(value).toBe("first");
+    expect(Bun.peek.status(second)).toBe("pending");
+    await second;
+
+    const closed = writer.close();
+    const rest: string[] = [];
+    for (let r; !(r = await reader.read()).done; ) rest.push(r.value);
+    await closed;
+    expect(rest.join("")).toBe("second");
+  });
+}
+
 // The transform/flush arm runs the native decoder directly, so monkeypatching
 // TextDecoder.prototype.decode no longer reaches it.
 test("TextDecoderStream does not call a patched TextDecoder.prototype.decode", async () => {
diff --git a/test/js/web/encoding/text-encoder-stream.test.ts b/test/js/web/encoding/text-encoder-stream.test.ts
index 5c2578ec9479..a65536ca2a46 100644
--- a/test/js/web/encoding/text-encoder-stream.test.ts
+++ b/test/js/web/encoding/text-encoder-stream.test.ts
@@ -1,4 +1,4 @@
-import { expect, test } from "bun:test";
+import { describe, expect, test } from "bun:test";
 import { readableStreamFromArray } from "harness";
 
 // META: global=window,worker
@@ -267,3 +267,170 @@ test("TextEncoderStream -> TextDecoderStream -> TextEncoderStream -> native HTTP
   expect(out.byteLength).toBe(expected.byteLength);
   expect(Buffer.compare(out, Buffer.from(expected))).toBe(0);
 });
+
+// Two source shapes: `start` enqueues everything before the HTTP sink attaches
+// (readMany drains a pre-filled queue); `async pull` with a macrotask yield
+// keeps the source incomplete, so each chunk goes through encodeIntoSink while
+// the response is actively streaming.
+const sourceShapes = {
+  start(chunks: readonly unknown[]) {
+    return new ReadableStream({
+      start(c) {
+        for (const s of chunks) c.enqueue(s);
+        c.close();
+      },
+    });
+  },
+  "async pull"(chunks: readonly unknown[]) {
+    let i = 0;
+    return new ReadableStream({
+      async pull(c) {
+        await Bun.sleep(0);
+        if (i < chunks.length) c.enqueue(chunks[i++]);
+        else c.close();
+      },
+    });
+  },
+};
+
+describe.each(Object.entries(sourceShapes))("TextEncoderStream -> native sink (%s source)", (_, source) => {
+  // 8-bit (Latin-1) chunks go straight to the sink's own write_latin1. An
+  // all-ASCII chunk is zero-copy; a chunk with bytes 128-255 is widened to
+  // 2-byte UTF-8 inside the sink.
+  test("HTTP sink: 8-bit chunks (ASCII + Latin-1 non-ASCII)", async () => {
+    const ascii = Buffer.alloc(300_000, "plain ascii run.").toString();
+    const latin1 = "caf\xe9 \xffna\xefve \xbd"; // é ÿ ï ½
+    const chunks = [ascii, latin1, "mid", latin1, ascii];
+    const expected = Buffer.from(new TextEncoder().encode(chunks.join("")));
+    await using server = Bun.serve({
+      port: 0,
+      fetch: () => new Response(source(chunks).pipeThrough(new TextEncoderStream())),
+    });
+    const out = Buffer.from(await (await fetch(server.url)).arrayBuffer());
+    expect(out.byteLength).toBe(expected.byteLength);
+    expect(Buffer.compare(out, expected)).toBe(0);
+  });
+
+  // 16-bit chunks go straight to the sink's own write_utf16. The encoder still
+  // owns cross-chunk surrogate carry (trailing lone lead), and mid-chunk lone
+  // surrogates become U+FFFD via the sink.
+  test("HTTP sink: 16-bit chunks with surrogate edge cases", async () => {
+    const chunks = [
+      "A" + leading + leading, // 8-bit "A" roped with a 16-bit tail: mid-chunk lone lead + trailing lone lead
+      trailing + "B", // completes the carried pair
+      trailing, // lone trail, no carry -> FFFD
+      leading + leading + trailing, // mid-chunk lone lead + valid pair
+      Buffer.alloc(32_000, "\u{1F499}").toString(), // long 16-bit run
+      leading + trailing + leading, // valid pair then trailing lone lead
+      trailing, // completes
+      leading, // dangling -> FFFD at flush
+    ];
+    // Reference via the non-sink encodeForStream path (JSUint8Array per chunk).
+    const ref = Buffer.from(
+      await new Response(readableStreamFromArray(chunks).pipeThrough(new TextEncoderStream())).arrayBuffer(),
+    );
+    await using server = Bun.serve({
+      port: 0,
+      fetch: () => new Response(source(chunks).pipeThrough(new TextEncoderStream())),
+    });
+    const out = Buffer.from(await (await fetch(server.url)).arrayBuffer());
+    expect(out.toString("hex")).toBe(ref.toString("hex"));
+  });
+
+  // HTMLRewriter's sink runs user JS (content handlers) while the input slice
+  // is still being parsed; a non-string chunk exercises the ToString-created
+  // JSString being kept rooted across that call.
+  test("HTMLRewriter sink: non-string chunks + handler that allocates", async () => {
+    const ascii = Buffer.alloc(4096, "a").toString();
+    const chunks = [
+      "<div>",
+      { toString: () => "<p>" + ascii + "</p>" }, // 8-bit all-ASCII, ToString-coerced (RewriterPipe zero-copy branch)
+      { toString: () => "<p>caf\xe9</p>" }, // 8-bit non-ASCII, ToString-coerced (RewriterPipe Vec-copy branch)
+      { toString: () => "<p>\u{1F499}</p>" }, // 16-bit, ToString-coerced
+      "</div>",
+    ];
+    let seen = 0;
+    const out = await new HTMLRewriter()
+      .on("p", {
+        element() {
+          seen++;
+          Bun.gc(true);
+        },
+      })
+      .transform(new Response(source(chunks).pipeThrough(new TextEncoderStream())))
+      .text();
+    expect(seen).toBe(3);
+    expect(out).toBe("<div><p>" + ascii + "</p><p>caf\xe9</p><p>\u{1F499}</p></div>");
+  });
+});
+
+// Native-sink backpressure: when the HTTP response sink's socket buffer fills
+// (slow client), the sink's write_latin1 returns Backpressure, the transform
+// arm parks on m_nativeSinkReadyPromise, and the next writer.write() stays
+// pending until the client drains. Drive the writable side directly so the
+// parked write is observable via Bun.peek.status (no time-based polling).
+test("TextEncoderStream -> native HTTP sink applies backpressure to the writer", async () => {
+  const chunk = Buffer.alloc(256 * 1024, "x").toString();
+  const MAX = 128; // upper bound; backpressure engages well before this (tens of 256 KiB chunks)
+  const gotWriter = Promise.withResolvers<WritableStreamDefaultWriter<string>>();
+  await using server = Bun.serve({
+    port: 0,
+    fetch() {
+      const tes = new TextEncoderStream();
+      gotWriter.resolve(tes.writable.getWriter());
+      return new Response(tes.readable);
+    },
+  });
+  // The response headers flow once the body produces its first bytes, so write
+  // before awaiting fetch() to avoid a writer↔fetch deadlock.
+  const resP = fetch(server.url);
+  const writer = await gotWriter.promise;
+  await writer.write(chunk);
+  const reader = (await resP).body!.getReader();
+
+  let parked: Promise<void> | undefined;
+  let written = 1;
+  for (; written < MAX; written++) {
+    const w = writer.write(chunk);
+    // One macrotask: the transform + native-sink write has run and either
+    // resolved w (sink accepted) or left it pending on m_nativeSinkReadyPromise.
+    await new Promise<void>(r => setImmediate(r));
+    if (Bun.peek.status(w) === "pending") {
+      parked = w;
+      break;
+    }
+  }
+  expect(written).toBeLessThan(MAX);
+  expect(parked && Bun.peek.status(parked)).toBe("pending");
+
+  const closed = writer.close();
+  let received = 0;
+  for (let r; !(r = await reader.read()).done; ) received += r.value!.byteLength;
+  await parked;
+  await closed;
+  expect(received).toBe((written + 1) * chunk.length);
+});
+
+// Readable-side backpressure (no native sink): the readable queue's HWM is 1,
+// so a single write completes without a reader, and a second write parks until
+// the readable side is drained.
+test("TextEncoderStream readable-side backpressure: second write stays pending until drained", async () => {
+  const tes = new TextEncoderStream();
+  const writer = tes.writable.getWriter();
+  const reader = tes.readable.getReader();
+
+  await writer.write("first");
+  const second = writer.write("second");
+  expect(Bun.peek.status(second)).toBe("pending");
+
+  const { value } = await reader.read();
+  expect(new TextDecoder().decode(value)).toBe("first");
+  expect(Bun.peek.status(second)).toBe("pending");
+  await second;
+
+  const closed = writer.close();
+  const rest: Uint8Array[] = [];
+  for (let r; !(r = await reader.read()).done; ) rest.push(r.value);
+  await closed;
+  expect(new TextDecoder().decode(Buffer.concat(rest))).toBe("second");
+});