Merge pull request #1672 from pacocartones/fix/sse-utf8-chunk-boundary

fix(gateway): SSE transforms corrupt multi-byte characters split across chunks

packages/core/src/gateway/features/codex-multi-agent-bridge.ts

@@ -1,5 +1,6 @@
 import type { IncomingHttpHeaders } from "node:http";
 import { Readable, Transform } from "node:stream";
+import { StringDecoder } from "node:string_decoder";
 import type { AppConfig } from "@ccr/core/contracts/app";
 import { normalizeRouteSelector } from "@ccr/core/routing/model-registry";
 import { isRecord, rawStringValue, stringValue } from "@ccr/core/gateway/internal/value";
@@ -345,9 +346,17 @@ function transformMultiAgentFunctionCall(item: Record<string, unknown>): { value
   };
 }
 
+type CodexMultiAgentBridgeSseTransform = Transform & {
+  __ccrCodexMultiAgentBridgeSseDecoder?: StringDecoder;
+  __ccrCodexMultiAgentBridgeSsePending?: string;
+};
+
 function transformSseChunk(stream: Transform, chunk: Buffer | string): void {
-  const state = stream as Transform & { __ccrCodexMultiAgentBridgeSsePending?: string };
-  state.__ccrCodexMultiAgentBridgeSsePending = (state.__ccrCodexMultiAgentBridgeSsePending ?? "") + chunk.toString();
+  const state = stream as CodexMultiAgentBridgeSseTransform;
+  const decoder = state.__ccrCodexMultiAgentBridgeSseDecoder ?? new StringDecoder("utf8");
+  state.__ccrCodexMultiAgentBridgeSseDecoder = decoder;
+  state.__ccrCodexMultiAgentBridgeSsePending = (state.__ccrCodexMultiAgentBridgeSsePending ?? "") +
+    decoder.write(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
   while (state.__ccrCodexMultiAgentBridgeSsePending) {
     const match = /\r?\n\r?\n/.exec(state.__ccrCodexMultiAgentBridgeSsePending);
     if (!match || match.index === undefined) {
@@ -361,7 +370,9 @@ function transformSseChunk(stream: Transform, chunk: Buffer | string): void {
 }
 
 function flushSseTransform(stream: Transform): void {
-  const state = stream as Transform & { __ccrCodexMultiAgentBridgeSsePending?: string };
+  const state = stream as CodexMultiAgentBridgeSseTransform;
+  state.__ccrCodexMultiAgentBridgeSsePending = (state.__ccrCodexMultiAgentBridgeSsePending ?? "") +
+    (state.__ccrCodexMultiAgentBridgeSseDecoder?.end() ?? "");
   if (state.__ccrCodexMultiAgentBridgeSsePending) {
     stream.push(transformCodexMultiAgentBridgeSseEvent(state.__ccrCodexMultiAgentBridgeSsePending));
     state.__ccrCodexMultiAgentBridgeSsePending = "";

packages/core/src/gateway/features/codex-patch-bridge.ts

@@ -3,6 +3,7 @@
  */
 import type { IncomingHttpHeaders } from "node:http";
 import { Readable, Transform } from "node:stream";
+import { StringDecoder } from "node:string_decoder";
 import type { AppConfig } from "@ccr/core/contracts/app";
 import { normalizeRouteSelector } from "@ccr/core/routing/model-registry";
 import { isRecord, rawStringValue, stringValue } from "@ccr/core/gateway/internal/value";
@@ -416,9 +417,17 @@ function patchInputFromVirtualApplyPatchArguments(value: unknown): string | unde
 }
 
 
+type CodexPatchBridgeSseTransform = Transform & {
+  __ccrCodexPatchBridgeSseDecoder?: StringDecoder;
+  __ccrCodexPatchBridgeSsePending?: string;
+};
+
 function transformSseChunk(stream: Transform, chunk: Buffer | string): void {
-  const state = stream as Transform & { __ccrCodexPatchBridgeSsePending?: string };
-  state.__ccrCodexPatchBridgeSsePending = (state.__ccrCodexPatchBridgeSsePending ?? "") + chunk.toString();
+  const state = stream as CodexPatchBridgeSseTransform;
+  const decoder = state.__ccrCodexPatchBridgeSseDecoder ?? new StringDecoder("utf8");
+  state.__ccrCodexPatchBridgeSseDecoder = decoder;
+  state.__ccrCodexPatchBridgeSsePending = (state.__ccrCodexPatchBridgeSsePending ?? "") +
+    decoder.write(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
   while (state.__ccrCodexPatchBridgeSsePending) {
     const match = /\r?\n\r?\n/.exec(state.__ccrCodexPatchBridgeSsePending);
     if (!match || match.index === undefined) {
@@ -433,7 +442,9 @@ function transformSseChunk(stream: Transform, chunk: Buffer | string): void {
 
 
 function flushSseTransform(stream: Transform): void {
-  const state = stream as Transform & { __ccrCodexPatchBridgeSsePending?: string };
+  const state = stream as CodexPatchBridgeSseTransform;
+  state.__ccrCodexPatchBridgeSsePending = (state.__ccrCodexPatchBridgeSsePending ?? "") +
+    (state.__ccrCodexPatchBridgeSseDecoder?.end() ?? "");
   if (state.__ccrCodexPatchBridgeSsePending) {
     stream.push(transformCodexApplyPatchBridgeSseEvent(state.__ccrCodexPatchBridgeSsePending));
     state.__ccrCodexPatchBridgeSsePending = "";

packages/core/src/gateway/features/hosted-web-search/response-transform.ts

@@ -1,4 +1,5 @@
 import { Readable, Transform } from "node:stream";
+import { StringDecoder } from "node:string_decoder";
 import type { GatewayProviderProtocol } from "@ccr/core/contracts/app";
 import { isRecord, numberValue, stringValue } from "@ccr/core/gateway/internal/value";
 import { formatError } from "@ccr/core/gateway/http/io";
@@ -65,6 +66,7 @@ function hostedWebSearchProtocolSseStream(
   const recordsPromise = context.records?.length
     ? Promise.resolve(context.records)
     : selectHostedWebSearchProtocolRecords(context, integration);
+  const decoder = new StringDecoder("utf8");
   let records: BrowserWebSearchProtocolRecord[] | undefined;
   let pending = "";
   let passThrough = false;
@@ -84,7 +86,7 @@ function hostedWebSearchProtocolSseStream(
 
   return input.pipe(new Transform({
     transform(chunk, _encoding, callback) {
-      const text = chunk.toString();
+      const text = decoder.write(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
       const rawText = pending + text;
       void (async () => {
         await ensureRecords();
@@ -103,6 +105,7 @@ function hostedWebSearchProtocolSseStream(
       }).finally(() => callback());
     },
     flush(callback) {
+      pending += decoder.end();
       void (async () => {
         await ensureRecords();
         if (passThrough || !records) {
@@ -263,6 +266,7 @@ function anthropicHostedWebSearchProtocolSseStream(
   const recordsPromise = context.records?.length
     ? Promise.resolve(context.records)
     : selectHostedWebSearchProtocolRecords(context, integration);
+  const decoder = new StringDecoder("utf8");
   let records: BrowserWebSearchProtocolRecord[] | undefined;
   let pending = "";
   let passThrough = false;
@@ -286,7 +290,7 @@ function anthropicHostedWebSearchProtocolSseStream(
 
   return input.pipe(new Transform({
     transform(chunk, _encoding, callback) {
-      const text = chunk.toString();
+      const text = decoder.write(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
       const rawText = pending + text;
       void (async () => {
         await ensureRecords();
@@ -305,6 +309,7 @@ function anthropicHostedWebSearchProtocolSseStream(
       }).finally(() => callback());
     },
     flush(callback) {
+      pending += decoder.end();
       void (async () => {
         await ensureRecords();
         if (passThrough || !records) {

packages/core/test/unit/gateway/sse-utf8-chunk-boundary.test.mjs

@@ -0,0 +1,114 @@
+import assert from "node:assert/strict";
+import { Readable } from "node:stream";
+import test from "node:test";
+import { codexMultiAgentBridgeResponseStream } from "@ccr/core/gateway/features/codex-multi-agent-bridge.ts";
+import { codexApplyPatchBridgeResponseStream } from "@ccr/core/gateway/features/codex-patch-bridge.ts";
+import { hostedWebSearchProtocolResponseStream } from "@ccr/core/gateway/features/hosted-web-search/index.ts";
+
+const sseHeaders = () => new Headers({ "content-type": "text/event-stream" });
+
+const webSearchRecords = [{
+  engine: "test",
+  query: "北京天气",
+  results: [{ content: "多云", snippet: "多云", title: "天气", url: "https://example.test/weather" }],
+  searchUrl: "https://example.test/search"
+}];
+
+function hostedWebSearchContext(protocol) {
+  return {
+    maxUses: 1,
+    protocol,
+    queryHint: "北京天气",
+    records: webSearchRecords,
+    requestId: "req-utf8",
+    sinceMs: 0,
+    toolName: "web_search"
+  };
+}
+
+async function streamText(stream) {
+  const chunks = [];
+  for await (const chunk of stream) {
+    chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
+  }
+  return Buffer.concat(chunks).toString("utf8");
+}
+
+// Feeds `text` through `makeStream` once per interior byte offset of the first
+// multi-byte character, so at least one chunk boundary always lands inside it.
+async function assertSurvivesEveryByteSplit(makeStream, text, marker) {
+  const bytes = Buffer.from(text, "utf8");
+  const markerBytes = Buffer.from(marker, "utf8");
+  const markerStart = bytes.indexOf(markerBytes);
+  assert.ok(markerStart >= 0, "marker must be present in the event");
+  assert.ok(markerBytes.length > 1, "marker must be a multi-byte character");
+
+  for (let offset = 1; offset < markerBytes.length; offset += 1) {
+    const splitAt = markerStart + offset;
+    const output = await streamText(makeStream(
+      Readable.from([bytes.subarray(0, splitAt), bytes.subarray(splitAt)])
+    ));
+    assert.equal(
+      output.includes("�"),
+      false,
+      `split at byte ${splitAt} produced U+FFFD: ${JSON.stringify(output)}`
+    );
+    assert.equal(output, text, `split at byte ${splitAt} changed the event`);
+  }
+}
+
+test("Codex multi-agent bridge SSE keeps multi-byte characters split across chunks", async () => {
+  await assertSurvivesEveryByteSplit(
+    (input) => codexMultiAgentBridgeResponseStream(input, sseHeaders()),
+    'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"北京今天多云"}\n\n',
+    "北"
+  );
+});
+
+test("Codex apply_patch bridge SSE keeps multi-byte characters split across chunks", async () => {
+  await assertSurvivesEveryByteSplit(
+    (input) => codexApplyPatchBridgeResponseStream(input, sseHeaders()),
+    'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"北京今天多云"}\n\n',
+    "北"
+  );
+});
+
+test("Hosted web search Anthropic SSE keeps multi-byte characters split across chunks", async () => {
+  await assertSurvivesEveryByteSplit(
+    (input) => hostedWebSearchProtocolResponseStream(
+      input,
+      sseHeaders(),
+      hostedWebSearchContext("anthropic_messages"),
+      undefined
+    ),
+    'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"北京今天多云"}}\n\n',
+    "北"
+  );
+});
+
+test("Hosted web search OpenAI Responses SSE keeps multi-byte characters split across chunks", async () => {
+  await assertSurvivesEveryByteSplit(
+    (input) => hostedWebSearchProtocolResponseStream(
+      input,
+      sseHeaders(),
+      hostedWebSearchContext("openai_responses"),
+      undefined
+    ),
+    'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","output_index":0,"delta":"北京今天多云"}\n\n',
+    "北"
+  );
+});
+
+test("Codex bridges keep emoji surrogate pairs split across chunks", async () => {
+  const event = 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"done 🎉"}\n\n';
+  await assertSurvivesEveryByteSplit(
+    (input) => codexMultiAgentBridgeResponseStream(input, sseHeaders()),
+    event,
+    "🎉"
+  );
+  await assertSurvivesEveryByteSplit(
+    (input) => codexApplyPatchBridgeResponseStream(input, sseHeaders()),
+    event,
+    "🎉"
+  );
+});