import type Anthropic from "@anthropic-ai/sdk";
import { Type } from "typebox";
import { describe, expect, it } from "vitest";
import { getModel } from "../src/models.js";
import { streamAnthropic } from "../src/providers/anthropic.js";
import type { Context, ToolCall } from "../src/types.js";
function createSseResponse(events: Array<{ event: string; data: string }>): Response {
const body = events.map(({ event, data }) => `event: ${event}\ndata: ${data}\n`).join("\n");
return new Response(body, {
status: 200,
headers: { "content-type": "text/event-stream" },
});
}
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: {
create: () => ({
asResponse: async () => response,
}),
},
} as unknown as Anthropic;
}
type CacheCreation = { ephemeral_5m_input_tokens: number; ephemeral_1h_input_tokens: number };
function createCacheUsageEvents(
cacheCreation: CacheCreation,
deltaCacheCreation?: CacheCreation,
): Array<{ event: string; data: string }> {
const tokens = (c: CacheCreation) => c.ephemeral_5m_input_tokens + c.ephemeral_1h_input_tokens;
return [
{
event: "message_start",
data: JSON.stringify({
type: "message_start",
message: {
id: "msg_cache_test",
usage: {
input_tokens: 12,
output_tokens: 0,
cache_read_input_tokens: 0,
cache_creation_input_tokens: tokens(cacheCreation),
cache_creation: cacheCreation,
},
},
}),
},
{
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: tokens(deltaCacheCreation ?? cacheCreation),
...(deltaCacheCreation ? { cache_creation: deltaCacheCreation } : {}),
},
}),
},
{ event: "message_stop", data: JSON.stringify({ type: "message_stop" }) },
];
}
describe("Anthropic raw SSE parsing", () => {
it.each([
{
name: "five-minute writes",
cacheRetention: "short" as const,
cacheCreation: { ephemeral_5m_input_tokens: 1000, ephemeral_1h_input_tokens: 0 },
expectedCacheWriteCost: 0.00125,
},
{
name: "one-hour writes",
cacheRetention: "long" as const,
cacheCreation: { ephemeral_5m_input_tokens: 0, ephemeral_1h_input_tokens: 1000 },
expectedCacheWriteCost: 0.002,
},
{
name: "mixed TTL writes",
cacheRetention: "long" as const,
cacheCreation: { ephemeral_5m_input_tokens: 250, ephemeral_1h_input_tokens: 750 },
expectedCacheWriteCost: 0.0018125,
},
])("prices $name from the reported Anthropic usage breakdown", async (testCase) => {
const model = getModel("anthropic", "claude-haiku-4-5");
const response = createSseResponse(createCacheUsageEvents(testCase.cacheCreation));
const result = await streamAnthropic(
model,
{ messages: [{ role: "user", content: "Say hello.", timestamp: Date.now() }] },
{
client: createFakeAnthropicClient(response),
cacheRetention: testCase.cacheRetention,
},
).result();
expect(result.usage.cacheWrite).toBe(1000);
expect(result.usage.cost.cacheWrite).toBeCloseTo(testCase.expectedCacheWriteCost);
});
it("reprices cache writes from a message_delta usage breakdown", async () => {
const model = getModel("anthropic", "claude-haiku-4-5");
const response = createSseResponse(
createCacheUsageEvents(
{ ephemeral_5m_input_tokens: 1000, ephemeral_1h_input_tokens: 0 },
{ ephemeral_5m_input_tokens: 0, ephemeral_1h_input_tokens: 2000 },
),
);
const result = await streamAnthropic(
model,
{ messages: [{ role: "user", content: "Say hello.", timestamp: Date.now() }] },
{ client: createFakeAnthropicClient(response), cacheRetention: "long" },
).result();
expect(result.usage.cacheWrite).toBe(2000);
// 2000 one-hour tokens at 2x input cost, not the stale 1.25x rate from message_start.
expect(result.usage.cost.cacheWrite).toBeCloseTo(0.004, 6);
});
it("preserves configured cache write pricing for non-Anthropic models", async () => {
const model = getModel("minimax", "MiniMax-M2.7-highspeed");
const response = createSseResponse(
createCacheUsageEvents({ ephemeral_5m_input_tokens: 1000, ephemeral_1h_input_tokens: 0 }),
);
const result = await streamAnthropic(
model,
{ messages: [{ role: "user", content: "Say hello.", timestamp: Date.now() }] },
{ client: createFakeAnthropicClient(response) },
).result();
expect(result.usage.cacheWrite).toBe(1000);
expect(result.usage.cost.cacheWrite).toBeCloseTo((1000 * model.cost.cacheWrite) / 1_000_000);
});
it("repairs malformed SSE JSON and malformed streamed tool JSON", async () => {
const model = getModel("anthropic", "claude-haiku-4-5");
const context: Context = {
messages: [{ role: "user", content: "Use the edit tool.", timestamp: Date.now() }],
tools: [
{
name: "edit",
description: "Edit a file.",
parameters: Type.Object({
path: Type.String(),
text: Type.String(),
}),
},
],
};
const malformedToolJsonDelta = String.raw`{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"path\":\"A\H\",\"text\":\"col1 col2\"}"}}`;
const response = createSseResponse([
{
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: "tool_use",
id: "toolu_test",
name: "edit",
input: {},
},
}),
},
{ event: "content_block_delta", data: malformedToolJsonDelta },
{
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: "tool_use" },
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" }),
},
]);
const stream = streamAnthropic(model, context, {
client: createFakeAnthropicClient(response),
});
const result = await stream.result();
expect(result.stopReason).toBe("toolUse");
expect(result.errorMessage).toBeUndefined();
const toolCall = result.content.find((block): block is ToolCall => block.type === "toolCall");
expect(toolCall).toBeDefined();
expect(toolCall?.arguments).toEqual({
path: "A\\H",
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" }]);
});
});