From 3e7ffff184e805930e23c7440fa9571407a485ee Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Sat, 25 Apr 2026 16:57:03 +0200 Subject: [PATCH] fix(ai): ignore unknown anthropic sse events closes #3708 --- packages/ai/CHANGELOG.md | 1 + packages/ai/src/providers/anthropic.ts | 17 ++++- .../ai/test/anthropic-sse-parsing.test.ts | 76 +++++++++++++++++++ .../anthropic-tool-name-normalization.test.ts | 2 +- packages/ai/test/context-overflow.test.ts | 2 +- ...-shared-gemini3-unsigned-tool-call.test.ts | 4 +- packages/ai/test/lazy-module-load.test.ts | 4 +- packages/ai/test/stream.test.ts | 4 +- packages/ai/test/total-tokens.test.ts | 2 +- 9 files changed, 99 insertions(+), 13 deletions(-) diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index e41da3a6..8d789f7b 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -8,6 +8,7 @@ ### Fixed +- Fixed Anthropic SSE parsing to ignore unknown proxy events such as OpenAI-style `done` terminators ([#3708](https://github.com/badlogic/pi-mono/issues/3708)). - Fixed OpenAI-compatible prompt cache tests to cover proxies that explicitly disable long cache retention. - Stopped sending `tools: []` on OpenAI-compatible, Anthropic, OpenAI Responses, OpenAI Codex Responses, and Azure OpenAI Responses requests when no tools are active (e.g. `pi --no-tools`). DashScope/Aliyun Qwen (OpenAI-compatible) rejects empty tools arrays with `"[] is too short - 'tools'"` (HTTP 400); the field is now omitted unless the conversation has tool history (the existing LiteLLM/Anthropic-proxy workaround). - Fixed `supportsXhigh()` to recognize DeepSeek V4 Pro, preserving `xhigh` reasoning requests so they map to DeepSeek's `max` effort ([#3662](https://github.com/badlogic/pi-mono/issues/3662)) diff --git a/packages/ai/src/providers/anthropic.ts b/packages/ai/src/providers/anthropic.ts index 72d82d40..1c597a59 100644 --- a/packages/ai/src/providers/anthropic.ts +++ b/packages/ai/src/providers/anthropic.ts @@ -237,6 +237,15 @@ interface SseDecoderState { raw: string[]; } +const ANTHROPIC_MESSAGE_EVENTS: ReadonlySet = new Set([ + "message_start", + "message_delta", + "message_stop", + "content_block_start", + "content_block_delta", + "content_block_stop", +]); + function flushSseEvent(state: SseDecoderState): ServerSentEvent | null { if (!state.event && state.data.length === 0) { return null; @@ -376,14 +385,14 @@ async function* iterateAnthropicEvents( } for await (const sse of iterateSseMessages(response.body, signal)) { - if (!sse.event || sse.event === "ping") { - continue; - } - if (sse.event === "error") { throw new Error(sse.data); } + if (!ANTHROPIC_MESSAGE_EVENTS.has(sse.event ?? "")) { + continue; + } + try { yield parseJsonWithRepair(sse.data); } catch (error) { diff --git a/packages/ai/test/anthropic-sse-parsing.test.ts b/packages/ai/test/anthropic-sse-parsing.test.ts index 48599408..5bbc7d72 100644 --- a/packages/ai/test/anthropic-sse-parsing.test.ts +++ b/packages/ai/test/anthropic-sse-parsing.test.ts @@ -13,6 +13,61 @@ function createSseResponse(events: Array<{ event: string; data: string }>): Resp }); } +const minimalAnthropicEvents = [ + { + event: "message_start", + data: JSON.stringify({ + type: "message_start", + message: { + id: "msg_test", + usage: { + input_tokens: 12, + output_tokens: 0, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + }, + }, + }), + }, + { + event: "content_block_start", + data: JSON.stringify({ + type: "content_block_start", + index: 0, + content_block: { type: "text", text: "" }, + }), + }, + { + event: "content_block_delta", + data: JSON.stringify({ + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: "Hello" }, + }), + }, + { + event: "content_block_stop", + data: JSON.stringify({ type: "content_block_stop", index: 0 }), + }, + { + event: "message_delta", + data: JSON.stringify({ + type: "message_delta", + delta: { stop_reason: "end_turn" }, + usage: { + input_tokens: 12, + output_tokens: 5, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + }, + }), + }, + { + event: "message_stop", + data: JSON.stringify({ type: "message_stop" }), + }, +]; + function createFakeAnthropicClient(response: Response): Anthropic { return { messages: { @@ -110,4 +165,25 @@ describe("Anthropic raw SSE parsing", () => { text: "col1\tcol2", }); }); + + it("ignores unknown SSE events after message_stop", async () => { + const model = getModel("anthropic", "claude-haiku-4-5"); + const context: Context = { + messages: [{ role: "user", content: "Say hello.", timestamp: Date.now() }], + }; + const response = createSseResponse([ + ...minimalAnthropicEvents, + { event: "done", data: "[DONE]" }, + { event: "proxy.stats", data: "not json" }, + ]); + + const stream = streamAnthropic(model, context, { + client: createFakeAnthropicClient(response), + }); + const result = await stream.result(); + + expect(result.stopReason).toBe("stop"); + expect(result.errorMessage).toBeUndefined(); + expect(result.content).toEqual([{ type: "text", text: "Hello" }]); + }); }); diff --git a/packages/ai/test/anthropic-tool-name-normalization.test.ts b/packages/ai/test/anthropic-tool-name-normalization.test.ts index 0a6c6bd3..1ea6bdb2 100644 --- a/packages/ai/test/anthropic-tool-name-normalization.test.ts +++ b/packages/ai/test/anthropic-tool-name-normalization.test.ts @@ -24,7 +24,7 @@ const oauthToken = await resolveApiKey("anthropic"); * - Result: tool call has name "Glob" but no tool exists with that name */ describe.skipIf(!oauthToken)("Anthropic OAuth tool name normalization", () => { - const model = getModel("anthropic", "claude-sonnet-4-20250514"); + const model = getModel("anthropic", "claude-sonnet-4-6"); it("should normalize user-defined tool matching CC name (todowrite -> TodoWrite -> todowrite)", async () => { // User defines a tool named "todowrite" (lowercase) diff --git a/packages/ai/test/context-overflow.test.ts b/packages/ai/test/context-overflow.test.ts index 5b1e08ef..ab58ea07 100644 --- a/packages/ai/test/context-overflow.test.ts +++ b/packages/ai/test/context-overflow.test.ts @@ -113,7 +113,7 @@ describe("Context overflow error handling", () => { describe.skipIf(!process.env.ANTHROPIC_OAUTH_TOKEN)("Anthropic (OAuth)", () => { it("claude-sonnet-4 - should detect overflow via isContextOverflow", async () => { - const model = getModel("anthropic", "claude-sonnet-4-20250514"); + const model = getModel("anthropic", "claude-sonnet-4-6"); const result = await testContextOverflow(model, process.env.ANTHROPIC_OAUTH_TOKEN!); logResult(result); diff --git a/packages/ai/test/google-shared-gemini3-unsigned-tool-call.test.ts b/packages/ai/test/google-shared-gemini3-unsigned-tool-call.test.ts index 179fe75d..5e51b876 100644 --- a/packages/ai/test/google-shared-gemini3-unsigned-tool-call.test.ts +++ b/packages/ai/test/google-shared-gemini3-unsigned-tool-call.test.ts @@ -39,7 +39,7 @@ describe("google-shared convertMessages — Gemini 3 unsigned tool calls", () => ], api: "google-gemini-cli", provider: "google-antigravity", - model: "claude-sonnet-4-20250514", + model: "claude-sonnet-4-6", usage: { input: 0, output: 0, @@ -146,7 +146,7 @@ describe("google-shared convertMessages — Gemini 3 unsigned tool calls", () => ], api: "google-gemini-cli", provider: "google-antigravity", - model: "claude-sonnet-4-20250514", + model: "claude-sonnet-4-6", usage: { input: 0, output: 0, diff --git a/packages/ai/test/lazy-module-load.test.ts b/packages/ai/test/lazy-module-load.test.ts index 84d82752..c94669ab 100644 --- a/packages/ai/test/lazy-module-load.test.ts +++ b/packages/ai/test/lazy-module-load.test.ts @@ -72,7 +72,7 @@ describe("lazy provider module loading", () => { it("loads only the Anthropic SDK when calling the root lazy wrapper", () => { const result = runProbe(` const model = { - id: "claude-sonnet-4-20250514", + id: "claude-sonnet-4-6", name: "Claude Sonnet 4", api: "anthropic-messages", provider: "anthropic", @@ -92,7 +92,7 @@ describe("lazy provider module loading", () => { it("loads only the Anthropic SDK when dispatching through streamSimple", () => { const result = runProbe(` - const model = mod.getModel("anthropic", "claude-sonnet-4-20250514"); + const model = mod.getModel("anthropic", "claude-sonnet-4-6"); const context = { messages: [{ role: "user", content: "hi" }] }; await mod.streamSimple(model, context).result(); `); diff --git a/packages/ai/test/stream.test.ts b/packages/ai/test/stream.test.ts index f23a05fd..d007f49f 100644 --- a/packages/ai/test/stream.test.ts +++ b/packages/ai/test/stream.test.ts @@ -879,8 +879,8 @@ describe("Generate E2E Tests", () => { // Tokens are resolved at module level (see oauthTokens above) // ========================================================================= - describe("Anthropic OAuth Provider (claude-sonnet-4-20250514)", () => { - const model = getModel("anthropic", "claude-sonnet-4-20250514"); + describe("Anthropic OAuth Provider (claude-sonnet-4-6)", () => { + const model = getModel("anthropic", "claude-sonnet-4-6"); it.skipIf(!anthropicOAuthToken)("should complete basic text generation", { retry: 3 }, async () => { await basicTextGeneration(model, { apiKey: anthropicOAuthToken }); diff --git a/packages/ai/test/total-tokens.test.ts b/packages/ai/test/total-tokens.test.ts index e81cc125..fd1ced76 100644 --- a/packages/ai/test/total-tokens.test.ts +++ b/packages/ai/test/total-tokens.test.ts @@ -132,7 +132,7 @@ describe("totalTokens field", () => { "claude-sonnet-4 - should return totalTokens equal to sum of components", { retry: 3, timeout: 60000 }, async () => { - const llm = getModel("anthropic", "claude-sonnet-4-20250514"); + const llm = getModel("anthropic", "claude-sonnet-4-6"); console.log(`\nAnthropic OAuth / ${llm.id}:`); const { first, second } = await testTotalTokensWithCache(llm, { apiKey: anthropicOAuthToken });