From 1283afd0d0685d1ffe88aa56a725c0e4ad3cfc7b Mon Sep 17 00:00:00 2001 From: David Brailovsky Date: Thu, 17 Sep 2026 10:22:47 +0200 Subject: [PATCH] fix(ai): preserve thinking replay through renamed Anthropic models Keep the requested model ID on assistant messages and record the provider-reported model separately, while retaining fallback pricing. Fixes #9188 --- packages/ai/CHANGELOG.md | 1 + packages/ai/src/api/anthropic-messages.ts | 9 +- packages/ai/src/types.ts | 2 +- .../ai/test/anthropic-sse-parsing.test.ts | 88 ++++++++++++++++++- 4 files changed, 94 insertions(+), 6 deletions(-) diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index d8e5f7339..c2aa1544f 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -10,6 +10,7 @@ ### Fixed - Fixed Google Generative AI and Vertex AI using unsupported thinking levels when reasoning is omitted or when model capabilities differ within a Gemini family ([#9455](https://github.com/earendil-works/pi/issues/9455)). +- Fixed Anthropic-compatible relays breaking signed thinking replay when they report a different response model, while preserving fallback pricing ([#9188](https://github.com/earendil-works/pi/issues/9188)). - Fixed Amazon Bedrock one-hour cache writes being priced at the five-minute rate ([#9457](https://github.com/earendil-works/pi/issues/9457)). - Added `RetryPolicy.maxAgentDelayMs` support to cap shared assistant retry backoff for summarization calls ([#8826](https://github.com/earendil-works/pi/issues/8826)). - Fixed quadratic CPU usage when draining buffered `EventStream` events ([#9055](https://github.com/earendil-works/pi/issues/9055)). diff --git a/packages/ai/src/api/anthropic-messages.ts b/packages/ai/src/api/anthropic-messages.ts index f571a6e5a..2192e754a 100644 --- a/packages/ai/src/api/anthropic-messages.ts +++ b/packages/ai/src/api/anthropic-messages.ts @@ -603,14 +603,15 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = ( output.responseId = event.message.id; const transformations = event.message.input_transformations; if (Array.isArray(transformations)) inputTransformations = transformations; - output.model = event.message.model; + const responseModel = event.message.model; + if (responseModel !== model.id) output.responseModel = responseModel; const fallbackCost = - output.model === model.id + responseModel === model.id ? undefined : model.compat?.allowedFallbackModels?.find( - (fallback) => fallback.provider === model.provider && fallback.model === output.model, + (fallback) => fallback.provider === model.provider && fallback.model === responseModel, )?.cost; - usageModel = fallbackCost ? { ...model, id: output.model, cost: fallbackCost } : model; + usageModel = fallbackCost ? { ...model, id: responseModel, cost: fallbackCost } : model; // Capture initial token usage from message_start event // This ensures we have input token counts even if the stream is aborted early output.usage.input = event.message.usage.input_tokens || 0; diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index b04438153..4dcfd08d5 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -472,7 +472,7 @@ export interface AssistantMessage { api: Api; provider: ProviderId; model: string; - responseModel?: string; // Concrete `chunk.model` when different from the requested `model` (e.g. OpenRouter `auto` -> `anthropic/...`) + responseModel?: string; // Concrete model reported by the provider when different from the requested `model` responseId?: string; // Provider-specific response/message identifier when the upstream API exposes one /** Exact provider-native effort level used for this response. Absent for legacy or unmanaged responses. */ providerThinkingLevel?: string; diff --git a/packages/ai/test/anthropic-sse-parsing.test.ts b/packages/ai/test/anthropic-sse-parsing.test.ts index 872f2886b..36de46d26 100644 --- a/packages/ai/test/anthropic-sse-parsing.test.ts +++ b/packages/ai/test/anthropic-sse-parsing.test.ts @@ -2,8 +2,9 @@ import type Anthropic from "@anthropic-ai/sdk"; import { Type } from "typebox"; import { describe, expect, it } from "vitest"; import { stream as streamAnthropic } from "../src/api/anthropic-messages.ts"; +import { transformMessages } from "../src/api/transform-messages.ts"; import { getModel, normalizeContext } from "../src/compat.ts"; -import type { ToolCall } from "../src/types.ts"; +import type { Model, ToolCall } from "../src/types.ts"; function createSseResponse(events: Array<{ event: string; data: string }>): Response { const body = events.map(({ event, data }) => `event: ${event}\ndata: ${data}\n`).join("\n"); @@ -80,7 +81,92 @@ function createFakeAnthropicClient(response: Response): Anthropic { } as unknown as Anthropic; } +type ResponseContentBlock = { type: "thinking"; thinking: string; signature: string } | { type: "text"; text: string }; + +function createResponseModelSseResponse(model: string, contentBlock: ResponseContentBlock): Response { + return createSseResponse([ + { + event: "message_start", + data: JSON.stringify({ + type: "message_start", + message: { id: "msg_response_model", model, usage: { input_tokens: 100, output_tokens: 0 } }, + }), + }, + { + event: "content_block_start", + data: JSON.stringify({ type: "content_block_start", index: 0, content_block: contentBlock }), + }, + { 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: 100, output_tokens: 20 }, + }), + }, + { event: "message_stop", data: JSON.stringify({ type: "message_stop" }) }, + ]); +} + describe("Anthropic raw SSE parsing", () => { + it("keeps signed thinking replayable when a proxy relabels the model", async () => { + // Regression test for earendil-works/pi#9188. + const model = getModel("anthropic", "claude-opus-5"); + const responseModel = "kimi-for-coding"; + const initialContext = normalizeContext({ + messages: [{ role: "user", content: "Hello", timestamp: 1 }], + }); + const first = await streamAnthropic(model, initialContext, { + client: createFakeAnthropicClient( + createResponseModelSseResponse(responseModel, { + type: "thinking", + thinking: "reasoning", + signature: "signature", + }), + ), + }).result(); + + expect(first.model).toBe(model.id); + expect(first.responseModel).toBe(responseModel); + + const transformed = transformMessages([...initialContext.messages, first], model); + const replayedAssistant = transformed.find((message) => message.role === "assistant"); + expect(replayedAssistant?.content).toEqual([ + { type: "thinking", thinking: "reasoning", thinkingSignature: "signature" }, + ]); + }); + + it("uses a returned fallback model for cost attribution", async () => { + const fallbackModel = "fallback-model"; + const model: Model<"anthropic-messages"> = { + ...getModel("anthropic", "claude-opus-5"), + compat: { + allowedFallbackModels: [ + { + provider: "anthropic", + model: fallbackModel, + cost: { input: 3, output: 5, cacheRead: 0, cacheWrite: 0 }, + }, + ], + }, + }; + const result = await streamAnthropic( + model, + normalizeContext({ messages: [{ role: "user", content: "Hello", timestamp: 1 }] }), + { + client: createFakeAnthropicClient( + createResponseModelSseResponse(fallbackModel, { type: "text", text: "done" }), + ), + }, + ).result(); + + expect(result.model).toBe(model.id); + expect(result.responseModel).toBe(fallbackModel); + expect(result.usage.cost.input).toBeCloseTo(0.0003, 10); + expect(result.usage.cost.output).toBeCloseTo(0.0001, 10); + }); + it("fails safely when Anthropic falls back after output begins", async () => { const model = getModel("anthropic", "claude-opus-5"); const response = createSseResponse([