From 83592bb2df67a1dacdbeb6c7aefef8ae2ff89ef3 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Wed, 29 Apr 2026 23:30:29 +0200 Subject: [PATCH] fix(ai): detect incomplete Anthropic streams closes #3936 --- packages/ai/CHANGELOG.md | 1 + packages/ai/src/providers/anthropic.ts | 15 ++++++++++++++- 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 0e8f5e402..1330e4e20 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -4,6 +4,7 @@ ### Fixed +- Fixed Anthropic streams that end before `message_stop` to be treated as errors instead of successful partial responses ([#3936](https://github.com/badlogic/pi-mono/issues/3936)). - Fixed generated OpenAI-compatible DeepSeek V4 models to carry the provider-specific reasoning effort mapping outside the direct DeepSeek provider ([#3940](https://github.com/badlogic/pi-mono/issues/3940)). - Fixed DeepSeek V4 Flash and V4 Pro pricing metadata to match current official rates ([#3910](https://github.com/badlogic/pi-mono/issues/3910)). - Fixed DeepSeek prompt cache hits to be tracked from `prompt_cache_hit_tokens` in OpenAI-compatible usage responses ([#3880](https://github.com/badlogic/pi-mono/issues/3880)). diff --git a/packages/ai/src/providers/anthropic.ts b/packages/ai/src/providers/anthropic.ts index 1c597a595..367e19755 100644 --- a/packages/ai/src/providers/anthropic.ts +++ b/packages/ai/src/providers/anthropic.ts @@ -384,6 +384,9 @@ async function* iterateAnthropicEvents( throw new Error("Attempted to iterate over an Anthropic response with no body"); } + let sawMessageStart = false; + let sawMessageEnd = false; + for await (const sse of iterateSseMessages(response.body, signal)) { if (sse.event === "error") { throw new Error(sse.data); @@ -394,7 +397,13 @@ async function* iterateAnthropicEvents( } try { - yield parseJsonWithRepair(sse.data); + const event = parseJsonWithRepair(sse.data); + if (event.type === "message_start") { + sawMessageStart = true; + } else if (event.type === "message_stop") { + sawMessageEnd = true; + } + yield event; } catch (error) { const message = error instanceof Error ? error.message : String(error); throw new Error( @@ -402,6 +411,10 @@ async function* iterateAnthropicEvents( ); } } + + if (sawMessageStart && !sawMessageEnd) { + throw new Error("Anthropic stream ended before message_stop"); + } } export const streamAnthropic: StreamFunction<"anthropic-messages", AnthropicOptions> = (