From 2467ced060aedad159d9f1bb0ba390e097865666 Mon Sep 17 00:00:00 2001 From: Ingwannu Date: Thu, 23 Jul 2026 06:17:09 +0000 Subject: [PATCH] fix(anthropic): preserve terminal SSE frames --- .../src/content/docs/reference/adapters.md | 3 +- src/adapters/anthropic.ts | 59 ++++----- src/lib/sse-decoder.ts | 75 ++++++++++++ tests/anthropic-compatible-stream.test.ts | 112 ++++++++++++++++++ tests/sse-decoder.test.ts | 48 ++++++++ 5 files changed, 259 insertions(+), 38 deletions(-) create mode 100644 src/lib/sse-decoder.ts create mode 100644 tests/anthropic-compatible-stream.test.ts create mode 100644 tests/sse-decoder.test.ts diff --git a/docs-site/src/content/docs/reference/adapters.md b/docs-site/src/content/docs/reference/adapters.md index ed1b54bfb6d..4f1f7ed2192 100644 --- a/docs-site/src/content/docs/reference/adapters.md +++ b/docs-site/src/content/docs/reference/adapters.md @@ -62,7 +62,8 @@ streams the response back **untranslated**. output headroom, and **drops `temperature`/`top_p`** when thinking is enabled (Anthropic forbids them there). - Always sends `anthropic-version: 2023-06-01`. Streams `content_block_delta` (`text_delta`, - `thinking_delta`, `input_json_delta`). + `thinking_delta`, compatible `reasoning_delta`, `input_json_delta`). The SSE decoder preserves + event state across fetch chunks and accepts a terminal `message_stop` without a trailing newline. ## `google` diff --git a/src/adapters/anthropic.ts b/src/adapters/anthropic.ts index 5eb0ab718d7..b5c70d8b328 100644 --- a/src/adapters/anthropic.ts +++ b/src/adapters/anthropic.ts @@ -21,6 +21,7 @@ import { normalizeAnthropicImages } from "./anthropic-image-normalize"; import { neutralizeIdentity } from "./identity"; import { CLAUDE_CODE_HEADERS, claudeCodeSessionId } from "./client-fingerprint"; import { buildNonOpenAIToolCatalogNudgeForTools } from "./tool-catalog-nudge"; +import { decodeServerSentEvents } from "../lib/sse-decoder"; /** Map a user content part to an Anthropic content block (text or image source). */ function toAnthropicContentPart(p: OcxContentPart): unknown { @@ -719,9 +720,6 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti return; } - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; let currentBlockType = ""; let currentToolCallId = ""; let currentToolCallName = ""; @@ -739,34 +737,19 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti }; }; - try { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - buffer += decoder.decode(value, { stream: true }); + for await (const record of decodeServerSentEvents(response.body)) { + const payload = record.data.trim(); + if (!payload) continue; - const lines = buffer.split("\n"); - buffer = lines.pop() ?? ""; - - let currentEventType = ""; - for (const line of lines) { - if (line.startsWith("event: ")) { - currentEventType = line.slice(7).trim(); - continue; - } - if (!line.startsWith("data: ")) continue; - const payload = line.slice(6).trim(); - if (!payload) continue; - - let data: Record; - try { - data = JSON.parse(payload) as Record; - } catch { - debugDroppedFrame("anthropic", payload); - continue; - } + let data: Record; + try { + data = JSON.parse(payload) as Record; + } catch { + debugDroppedFrame("anthropic", payload); + continue; + } - switch (currentEventType || data.type) { + switch (record.event || data.type) { case "message_start": { const message = data.message as { usage?: Record } | undefined; pendingUsage = mergeAnthropicUsage(pendingUsage, message?.usage); @@ -794,7 +777,12 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti yield { type: "text_delta", text: delta.text }; } else if (delta.type === "thinking_delta" && typeof delta.thinking === "string") { yield { type: "thinking_delta", thinking: delta.thinking }; - } else if (delta.type === "signature_delta" && typeof delta.signature === "string" && currentBlockType === "thinking") { + } else if (delta.type === "reasoning_delta" && typeof delta.reasoning === "string") { + // Some Anthropic-compatible reasoning models use `reasoning` names for the + // otherwise equivalent thinking block. Preserve it as raw reasoning and keep + // later text blocks independent. + yield { type: "thinking_delta", thinking: delta.reasoning }; + } else if (delta.type === "signature_delta" && typeof delta.signature === "string" && (currentBlockType === "thinking" || currentBlockType === "reasoning")) { // Arrives once, just before the thinking block's content_block_stop; block-scoped // so a stray signature on a non-thinking block can never be captured. yield { type: "thinking_signature", signature: delta.signature }; @@ -827,20 +815,15 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti yield { type: "error", message: err?.message ?? "Anthropic error" }; return; } - } - currentEventType = ""; - } } - if (pendingUsage && !emittedDone) yield* emitDone(); - } finally { - reader.releaseLock(); } + if (pendingUsage && !emittedDone) yield* emitDone(); }, async parseResponse(response: Response): Promise { const json = await response.json() as Record; const events: AdapterEvent[] = []; - const content = json.content as { type: string; text?: string; id?: string; name?: string; input?: unknown; thinking?: string; signature?: string; data?: string }[] | undefined; + const content = json.content as { type: string; text?: string; id?: string; name?: string; input?: unknown; thinking?: string; reasoning?: string; signature?: string; data?: string }[] | undefined; if (content) { for (const block of content) { if (block.type === "text" && block.text) { @@ -850,6 +833,8 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti if (typeof block.signature === "string" && block.signature) { events.push({ type: "thinking_signature", signature: block.signature }); } + } else if (block.type === "reasoning" && typeof block.reasoning === "string") { + events.push({ type: "thinking_delta", thinking: block.reasoning }); } else if (block.type === "redacted_thinking" && typeof block.data === "string") { events.push({ type: "redacted_thinking", data: block.data }); } else if (block.type === "tool_use") { diff --git a/src/lib/sse-decoder.ts b/src/lib/sse-decoder.ts new file mode 100644 index 00000000000..1eb9a539c6b --- /dev/null +++ b/src/lib/sse-decoder.ts @@ -0,0 +1,75 @@ +export interface ServerSentEvent { + event?: string; + data: string; +} + +/** + * Decode text/event-stream records across arbitrary fetch chunk boundaries. + * + * The final record is dispatched at EOF even when the upstream omits the trailing blank line or + * final newline. That matters for compatible APIs that place a terminal event in the last bytes of + * the body: dropping that record turns a successful response into an adapter_eof failure. + */ +export async function* decodeServerSentEvents( + source: ReadableStream, +): AsyncGenerator { + const reader = source.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let event: string | undefined; + let dataLines: string[] = []; + + const dispatch = (): ServerSentEvent | undefined => { + if (dataLines.length === 0) { + event = undefined; + return undefined; + } + const record = { ...(event ? { event } : {}), data: dataLines.join("\n") }; + event = undefined; + dataLines = []; + return record; + }; + + const acceptLine = (rawLine: string): ServerSentEvent | undefined => { + const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; + if (line === "") return dispatch(); + if (line.startsWith(":")) return undefined; + + const colon = line.indexOf(":"); + const field = colon < 0 ? line : line.slice(0, colon); + let value = colon < 0 ? "" : line.slice(colon + 1); + if (value.startsWith(" ")) value = value.slice(1); + + if (field === "event") event = value; + else if (field === "data") dataLines.push(value); + return undefined; + }; + + try { + while (true) { + const { done, value } = await reader.read(); + if (value) buffer += decoder.decode(value, { stream: !done }); + + let newline: number; + while ((newline = buffer.indexOf("\n")) >= 0) { + const record = acceptLine(buffer.slice(0, newline)); + buffer = buffer.slice(newline + 1); + if (record) yield record; + } + + if (!done) continue; + buffer += decoder.decode(); + if (buffer.length > 0) { + const record = acceptLine(buffer); + buffer = ""; + if (record) yield record; + } + const finalRecord = dispatch(); + if (finalRecord) yield finalRecord; + break; + } + } finally { + try { await reader.cancel(); } catch { /* already closed/errored */ } + try { reader.releaseLock(); } catch { /* already released */ } + } +} diff --git a/tests/anthropic-compatible-stream.test.ts b/tests/anthropic-compatible-stream.test.ts new file mode 100644 index 00000000000..9bd2e076f68 --- /dev/null +++ b/tests/anthropic-compatible-stream.test.ts @@ -0,0 +1,112 @@ +import { describe, expect, test } from "bun:test"; +import { createAnthropicAdapter } from "../src/adapters/anthropic"; +import { bridgeToResponsesSSE } from "../src/bridge"; +import { responsesSseToAnthropicSse } from "../src/claude/outbound"; +import type { AdapterEvent, OcxProviderConfig } from "../src/types"; + +const provider = { + adapter: "anthropic", + baseUrl: "https://api.kimi.com/coding", + apiKey: "test-key", + authMode: "key", +} as OcxProviderConfig; + +const kimiCompatibleSse = [ + 'event: message_start\ndata: {"type":"message_start","message":{}}', + 'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"reasoning","reasoning":""}}', + 'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"reasoning_delta","reasoning":"think"}}', + 'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}', + 'event: content_block_start\ndata: {"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}', + 'event: content_block_delta\ndata: {"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"visible"}}', + 'event: content_block_stop\ndata: {"type":"content_block_stop","index":1}', + // Deliberately no usage snapshot and no newline after the terminal data line. + 'event: message_stop\ndata: {"type":"message_stop"}', +].join("\n\n"); + +function arbitrarilyChunkedResponse(text = kimiCompatibleSse): Response { + const encoder = new TextEncoder(); + const bytes = encoder.encode(text); + const cuts = [3, 19, 47, 101, 173, 251, 337, 419, bytes.length]; + return new Response(new ReadableStream({ + start(controller) { + let offset = 0; + for (const end of cuts) { + if (end <= offset) continue; + controller.enqueue(bytes.slice(offset, Math.min(end, bytes.length))); + offset = end; + if (offset >= bytes.length) break; + } + controller.close(); + }, + })); +} + +async function collectAdapterEvents(response: Response): Promise { + const events: AdapterEvent[] = []; + for await (const event of createAnthropicAdapter(provider).parseStream(response)) events.push(event); + return events; +} + +describe("Anthropic-compatible reasoning stream termination (#312)", () => { + test("preserves reasoning and visible text, then emits done from final message_stop", async () => { + const events = await collectAdapterEvents(arbitrarilyChunkedResponse()); + + expect(events).toContainEqual({ type: "thinking_delta", thinking: "think" }); + expect(events).toContainEqual({ type: "text_delta", text: "visible" }); + expect(events.at(-1)).toEqual({ type: "done", usage: undefined }); + }); + + test("Responses bridge completes instead of reporting adapter_eof", async () => { + const responses = bridgeToResponsesSSE( + createAnthropicAdapter(provider).parseStream(arbitrarilyChunkedResponse()), + "kimi/k3", + undefined, + undefined, + undefined, + undefined, + 0, + ); + const text = await new Response(responses).text(); + + expect(text).toContain("response.reasoning_summary_text.delta"); + expect(text).toContain("response.output_text.delta"); + expect(text).toContain("response.completed"); + expect(text).not.toContain("response.incomplete"); + expect(text).toContain("visible"); + }); + + test("Claude Messages translation keeps content blocks and message_stop", async () => { + const responses = bridgeToResponsesSSE( + createAnthropicAdapter(provider).parseStream(arbitrarilyChunkedResponse()), + "kimi/k3", + undefined, + undefined, + undefined, + undefined, + 0, + ); + const anthropic = responsesSseToAnthropicSse(responses, "k3", { pingIntervalMs: 0 }); + const text = await new Response(anthropic).text(); + + expect(text).toContain('"type":"thinking_delta","thinking":"think"'); + expect(text).toContain('"type":"text_delta","text":"visible"'); + expect(text).toContain("event: message_stop"); + }); + + test("non-streaming compatible reasoning blocks map without hiding later text", async () => { + const events = await createAnthropicAdapter(provider).parseResponse(new Response(JSON.stringify({ + content: [ + { type: "reasoning", reasoning: "think" }, + { type: "text", text: "visible" }, + ], + usage: { input_tokens: 2, output_tokens: 3 }, + stop_reason: "end_turn", + }))); + + expect(events).toEqual([ + { type: "thinking_delta", thinking: "think" }, + { type: "text_delta", text: "visible" }, + { type: "done", usage: { inputTokens: 2, outputTokens: 3 }, stopReason: "end_turn" }, + ]); + }); +}); diff --git a/tests/sse-decoder.test.ts b/tests/sse-decoder.test.ts new file mode 100644 index 00000000000..b2396098bf3 --- /dev/null +++ b/tests/sse-decoder.test.ts @@ -0,0 +1,48 @@ +import { describe, expect, test } from "bun:test"; +import { decodeServerSentEvents } from "../src/lib/sse-decoder"; + +function chunkedStream(chunks: string[]): ReadableStream { + const encoder = new TextEncoder(); + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(encoder.encode(chunk)); + controller.close(); + }, + }); +} + +async function collect(chunks: string[]) { + const records = []; + for await (const record of decodeServerSentEvents(chunkedStream(chunks))) records.push(record); + return records; +} + +describe("text/event-stream decoder", () => { + test("preserves event state across arbitrary reader chunks", async () => { + expect(await collect([ + "event: content_", + "block_delta\n", + "data: {\"type\":\"content_block_delta\"}\n\n", + ])).toEqual([{ + event: "content_block_delta", + data: '{"type":"content_block_delta"}', + }]); + }); + + test("dispatches the terminal record without a final newline or blank delimiter", async () => { + expect(await collect([ + "event: message_stop\n", + 'data: {"type":"message_stop"}', + ])).toEqual([{ + event: "message_stop", + data: '{"type":"message_stop"}', + }]); + }); + + test("joins multiline data and accepts CRLF framing", async () => { + expect(await collect([ + "event: custom\r\ndata: first\r\n", + "data: second\r\n\r\n", + ])).toEqual([{ event: "custom", data: "first\nsecond" }]); + }); +});