From 46e5dfdc8fcaf65bd3b4d05ddba006bd3aa883ed Mon Sep 17 00:00:00 2001
From: Diego Rodrigues de Sa e Souza <diegosouza.pw@gmail.com>
Date: Fri, 7 Aug 2026 18:08:26 -0300
Subject: [PATCH] fix(lmarena): emit Uint8Array SSE chunks instead of strings
 to satisfy shared pipeline contract (#9237)

Co-authored-by: diegosouzapw <diegosouzapw@users.noreply.github.com>
---
 changelog.d/fixes/9237-fix.plan.md            |  1 +
 open-sse/executors/lmarena/response.ts        |  6 +-
 tests/unit/lmarena-string-chunk-repro.test.ts | 75 +++++++++++++++++++
 3 files changed, 79 insertions(+), 3 deletions(-)
 create mode 100644 changelog.d/fixes/9237-fix.plan.md
 create mode 100644 tests/unit/lmarena-string-chunk-repro.test.ts

diff --git a/changelog.d/fixes/9237-fix.plan.md b/changelog.d/fixes/9237-fix.plan.md
new file mode 100644
index 0000000000..fde574eb17
--- /dev/null
+++ b/changelog.d/fixes/9237-fix.plan.md
@@ -0,0 +1 @@
+- fix(lmarena): emit Uint8Array SSE chunks instead of strings to satisfy shared pipeline contract (#9237)
\ No newline at end of file
diff --git a/open-sse/executors/lmarena/response.ts b/open-sse/executors/lmarena/response.ts
index 64aef907c9..da058e9eee 100644
--- a/open-sse/executors/lmarena/response.ts
+++ b/open-sse/executors/lmarena/response.ts
@@ -165,7 +165,7 @@ function baseChunk(model: string) {
 }
 
 function enqueueSse(controller: ReadableStreamDefaultController, chunk: Record<string, unknown>) {
-  controller.enqueue(`data: ${JSON.stringify(chunk)}\n\n`);
+  controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(chunk)}\n\n`));
 }
 
 function emitStopAndDone(controller: ReadableStreamDefaultController, model: string) {
@@ -173,7 +173,7 @@ function emitStopAndDone(controller: ReadableStreamDefaultController, model: str
     ...baseChunk(model),
     choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
   });
-  controller.enqueue("data: [DONE]\n\n");
+  controller.enqueue(new TextEncoder().encode("data: [DONE]\n\n"));
   controller.close();
 }
 
@@ -213,7 +213,7 @@ export function createOpenAIArenaStream(opts: {
   model: string;
   signal?: AbortSignal;
   log?: { error?: (scope: string, msg: string) => void };
-}): ReadableStream<Uint8Array | string> {
+}): ReadableStream<Uint8Array> {
   const { reader, model, signal, log } = opts;
   const decoder = new TextDecoder();
   let buffer = "";
diff --git a/tests/unit/lmarena-string-chunk-repro.test.ts b/tests/unit/lmarena-string-chunk-repro.test.ts
new file mode 100644
index 0000000000..7f76a4321c
--- /dev/null
+++ b/tests/unit/lmarena-string-chunk-repro.test.ts
@@ -0,0 +1,75 @@
+/**
+ * TDD repro for #9237: Arena SSE stream emits string chunks (not Uint8Array),
+ * which causes TextDecoder.decode in the shared pipeline to throw
+ * TypeError ERR_INVALID_ARG_TYPE.
+ */
+import { describe, it } from "node:test";
+import { ok, deepEqual, rejects } from "node:assert/strict";
+import { createOpenAIArenaStream } from "../../open-sse/executors/lmarena/response.ts";
+
+/**
+ * Build a fake upstream reader that yields SSE lines as Uint8Array,
+ * simulating what the Arena executor's upstream reader does.
+ */
+function fakeReader(lines: string[]): ReadableStreamDefaultReader<Uint8Array> {
+  let idx = 0;
+  const stream = new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (idx < lines.length) {
+        controller.enqueue(new TextEncoder().encode(lines[idx] + "\n"));
+        idx++;
+      } else {
+        controller.close();
+      }
+    },
+  });
+  return stream.getReader();
+}
+
+/**
+ * Drive the Arena stream through the real ensureStreamReadiness path
+ * to verify the contract: TextDecoder.decode must not throw on any chunk.
+ */
+async function collectArenaStream(
+  reader: ReadableStreamDefaultReader<Uint8Array>
+): Promise<string> {
+  const decoder = new TextDecoder();
+  let result = "";
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    // This is the exact call that throws ERR_INVALID_ARG_TYPE on string chunks
+    result += decoder.decode(value, { stream: true });
+  }
+  // flush
+  result += decoder.decode();
+  return result;
+}
+
+describe("Arena SSE stream — string vs Uint8Array contract (#9237)", () => {
+  it("should emit Uint8Array chunks that survive TextDecoder.decode without throwing", async () => {
+    const reader = fakeReader([
+      'data: a0:{"text":"Hello"}',
+      'data: ad:{}',
+    ]);
+    const arenaStream = createOpenAIArenaStream({
+      reader,
+      model: "test-model",
+    });
+
+    // verify the stream type is Uint8Array, not string
+    const collected = await collectArenaStream(
+      arenaStream.getReader()
+    );
+
+    // Should contain the content text and the [DONE] marker
+    ok(
+      collected.includes("Hello"),
+      `Expected collected output to include "Hello", got: ${collected.slice(0, 200)}`
+    );
+    ok(
+      collected.includes("[DONE]"),
+      `Expected collected output to include "[DONE]", got: ${collected.slice(0, 200)}`
+    );
+  });
+});
\ No newline at end of file