diff --git a/src/adapters/anthropic.ts b/src/adapters/anthropic.ts index 1ecf9e1c35b..420c0206755 100644 --- a/src/adapters/anthropic.ts +++ b/src/adapters/anthropic.ts @@ -63,6 +63,14 @@ function usageFromAnthropic(usage: Record | undefined): OcxUsage }; } +function mergeAnthropicUsage( + base: Record | undefined, + next: Record | undefined, +): Record | undefined { + if (!next) return base; + return { ...(base ?? {}), ...next }; +} + function buildToolNameTransforms(provider: OcxProviderConfig): { toWire: (name: string) => string; fromWire: (name: string) => string } { if (provider.authMode === "oauth") { return { toWire: applyClaudeToolPrefix, fromWire: stripClaudeToolPrefix }; @@ -269,6 +277,14 @@ export function createAnthropicAdapter(provider: OcxProviderConfig): ProviderAda let currentBlockType = ""; let currentToolCallId = ""; let currentToolCallName = ""; + let pendingUsage: Record | undefined; + let emittedDone = false; + + const emitDone = function* (): Generator { + if (emittedDone) return; + emittedDone = true; + yield { type: "done", usage: usageFromAnthropic(pendingUsage) }; + }; try { while (true) { @@ -298,6 +314,11 @@ export function createAnthropicAdapter(provider: OcxProviderConfig): ProviderAda } switch (currentEventType || data.type) { + case "message_start": { + const message = data.message as { usage?: Record } | undefined; + pendingUsage = mergeAnthropicUsage(pendingUsage, message?.usage); + break; + } case "content_block_start": { const block = data.content_block as { type: string; id?: string; name?: string } | undefined; if (!block) break; @@ -331,15 +352,11 @@ export function createAnthropicAdapter(provider: OcxProviderConfig): ProviderAda } case "message_delta": { const usage = data.usage as Record | undefined; - if (usage) { - yield { - type: "done", - usage: usageFromAnthropic(usage), - }; - } + pendingUsage = mergeAnthropicUsage(pendingUsage, usage); break; } case "message_stop": { + yield* emitDone(); break; } case "error": { @@ -351,6 +368,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig): ProviderAda currentEventType = ""; } } + if (pendingUsage && !emittedDone) yield* emitDone(); } finally { reader.releaseLock(); } diff --git a/tests/adapter-usage.test.ts b/tests/adapter-usage.test.ts index d27d1bc00cc..07c7ea10aeb 100644 --- a/tests/adapter-usage.test.ts +++ b/tests/adapter-usage.test.ts @@ -118,6 +118,44 @@ describe("usage and content retention (F2)", () => { expect(events.at(-1)).toEqual({ type: "done", usage: { inputTokens: 5, outputTokens: 1 } }); }); + test("anthropic stream merges message_start input usage with message_delta output usage", async () => { + const adapter = createAnthropicAdapter({ ...provider, adapter: "anthropic" }); + const response = new Response([ + 'event: message_start\n', + 'data: {"type":"message_start","message":{"usage":{"input_tokens":20,"cache_read_input_tokens":3,"cache_creation_input_tokens":2}}}\n\n', + 'event: content_block_delta\n', + 'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"hi"}}\n\n', + 'event: message_delta\n', + 'data: {"type":"message_delta","usage":{"output_tokens":4}}\n\n', + 'event: message_stop\n', + 'data: {"type":"message_stop"}\n\n', + ].join("")); + + const events = []; + for await (const event of adapter.parseStream(response)) events.push(event); + const dones = events.filter(e => e.type === "done"); + expect(events).toContainEqual({ type: "text_delta", text: "hi" }); + expect(dones).toHaveLength(1); + expect(dones[0]).toEqual({ + type: "done", + usage: { inputTokens: 20, outputTokens: 4, cachedInputTokens: 5 }, + }); + }); + + test("anthropic stream emits terminal usage on EOF when message_stop is missing", async () => { + const adapter = createAnthropicAdapter({ ...provider, adapter: "anthropic" }); + const response = new Response([ + 'event: message_start\n', + 'data: {"type":"message_start","message":{"usage":{"input_tokens":7}}}\n\n', + 'event: message_delta\n', + 'data: {"type":"message_delta","usage":{"output_tokens":1}}\n\n', + ].join("")); + + const events = []; + for await (const event of adapter.parseStream(response)) events.push(event); + expect(events.at(-1)).toEqual({ type: "done", usage: { inputTokens: 7, outputTokens: 1 } }); + }); + test("google emits exactly one done carrying usage", async () => { const adapter = createGoogleAdapter({ ...provider, adapter: "google" }); const response = new Response(