diff --git a/packages/core/src/gateway/features/codex-multi-agent-bridge.ts b/packages/core/src/gateway/features/codex-multi-agent-bridge.ts index 57f15fa3..3914cbc9 100644 --- a/packages/core/src/gateway/features/codex-multi-agent-bridge.ts +++ b/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): { 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 = ""; diff --git a/packages/core/src/gateway/features/codex-patch-bridge.ts b/packages/core/src/gateway/features/codex-patch-bridge.ts index a79be245..c3aa1d88 100644 --- a/packages/core/src/gateway/features/codex-patch-bridge.ts +++ b/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 = ""; diff --git a/packages/core/src/gateway/features/hosted-web-search/response-transform.ts b/packages/core/src/gateway/features/hosted-web-search/response-transform.ts index 0074cb05..0397402c 100644 --- a/packages/core/src/gateway/features/hosted-web-search/response-transform.ts +++ b/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) { diff --git a/packages/core/test/unit/gateway/sse-utf8-chunk-boundary.test.mjs b/packages/core/test/unit/gateway/sse-utf8-chunk-boundary.test.mjs new file mode 100644 index 00000000..b71d99fc --- /dev/null +++ b/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, + "🎉" + ); +});