From 49fc27d8726e173f93b6ef362997f36b67acfcfc Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 18 Sep 2026 21:00:34 +0900 Subject: [PATCH] fix(claude): report the prompt this proxy counted on message_start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Anthropic surface published `usage.input_tokens: 0` on `message_start` whenever the upstream had not reported usage before the first content frame. Third-party clients read the prompt size there — Paseo's context meter takes input from that frame and output from `message_delta` — so a turn whose own `/context` reported ~97k rendered as a nearly-empty ring. #4891 fixed the destinations that report usage up front. The internal bridge attaches `usage: null` to its lifecycle frames, so those paths had nothing to publish and kept sending zero. `responsesSseToAnthropicSse` now accepts an input-token floor: the count this proxy made of the prompt it forwarded, published only when no confirmed upstream usage arrived first. `claude-messages.ts` supplies it from `estimateClaudeRequestTokens`, the same estimate the usage log already trusts as a floor, computed at most once per request and shared with the log path. This narrows a recorded decision rather than ignoring it. The pinned case said zero is the honest placeholder and an estimate must not replace it. Zero is not honest about a prompt that exists — it asserts an empty one — and the Anthropic schema makes the field required, so the choice is between a false measurement and a real one of the request side. The second half of that decision stands: the first content frame is still never delayed to await usage, and `message_delta` remains authoritative. Closes #4857. --- src/claude/outbound.ts | 39 +++++- src/server/claude-messages.ts | 26 +++- .../claude-outbound.test.ts | 111 +++++++++++++++++- 3 files changed, 166 insertions(+), 10 deletions(-) diff --git a/src/claude/outbound.ts b/src/claude/outbound.ts index 8b98a8ea954..5756f68a360 100644 --- a/src/claude/outbound.ts +++ b/src/claude/outbound.ts @@ -189,7 +189,27 @@ function webSearchPairFromItem(item: Rec): { id: string; input: Rec; resultConte return { id, input, resultContent, completed }; } -function messageSnapshot(model: string, confirmedUsage?: Rec): Rec { +/** + * `inputTokenFloor` is this proxy's own count of the prompt it forwarded, used only when the + * upstream sent no confirmed usage before the first frame. + * + * Real Anthropic fills `message_start.message.usage.input_tokens` with the turn's prompt size, + * and third-party clients read it there — Paseo's context meter takes input from this frame and + * output from `message_delta`, so a hardcoded zero showed a nearly-empty ring for a turn whose + * `/context` reported ~97k (#4857). #4891 fixed the destinations that report usage up front; + * the internal bridge attaches `usage: null` to its lifecycle frames, so those paths had + * nothing to report and kept sending zero. + * + * A floor is a measurement, which is what makes it publishable here: it counts the prompt this + * proxy actually sent, the same estimate `claude-messages.ts` already trusts as a log floor. It + * is not a claim about upstream's tokenizer, and it is not final — `message_delta` carries the + * authoritative count for every reader that waits for it, exactly as before. + */ +function messageSnapshot(model: string, confirmedUsage?: Rec, inputTokenFloor?: number): Rec { + const usage = confirmedUsage + ?? (typeof inputTokenFloor === "number" && Number.isFinite(inputTokenFloor) && inputTokenFloor > 0 + ? { input_tokens: Math.trunc(inputTokenFloor), output_tokens: 0 } + : { input_tokens: 0, output_tokens: 0 }); return { id: `msg_${uuid()}`, type: "message", @@ -198,7 +218,7 @@ function messageSnapshot(model: string, confirmedUsage?: Rec): Rec { model, stop_reason: null, stop_sequence: null, - usage: confirmedUsage ?? { input_tokens: 0, output_tokens: 0 }, + usage, }; } @@ -228,7 +248,15 @@ interface OpenBlock { export function responsesSseToAnthropicSse( upstream: ReadableStream, model: string, - opts: { pingIntervalMs?: number; translatorBudget: TranslatorBudget }, + opts: { + pingIntervalMs?: number; + translatorBudget: TranslatorBudget; + /** + * This proxy's count of the prompt it forwarded, published on `message_start` only when the + * upstream sent no confirmed usage before the first frame. See `messageSnapshot` (#4857). + */ + inputTokenFloor?: number; + }, ): ReadableStream { const translatorBudget = opts.translatorBudget; const pingIntervalMs = opts?.pingIntervalMs ?? 20_000; @@ -283,7 +311,10 @@ export function responsesSseToAnthropicSse( const ensureStarted = () => { if (started) return; started = true; - emit("message_start", { type: "message_start", message: messageSnapshot(model, earlyAnthropicUsage) }); + emit("message_start", { + type: "message_start", + message: messageSnapshot(model, earlyAnthropicUsage, opts.inputTokenFloor), + }); emit("ping", { type: "ping" }); }; // Keepalive pings protect remote deployments behind LB/NAT idle timeouts even diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts index 83b6eb52647..b44fab2a95c 100644 --- a/src/server/claude-messages.ts +++ b/src/server/claude-messages.ts @@ -790,6 +790,22 @@ async function handleClaudeMessagesWithBudget( if (!requestedModel) requestedModel = (anthropicBody as Rec).model as string; const stream = internalBody.stream === true; + /** + * This proxy's count of the prompt it is about to forward, computed at most once. + * + * Two readers want it and they want it under different rules. The usage log takes it as a + * floor only for estimated-usage adapters, because its merge is `max(reported, estimate)` and + * would otherwise overwrite real usage. `message_start` takes it whenever the upstream sent + * no confirmed usage before the first frame, where nothing is merged and the terminal + * `message_delta` still corrects it (#4857). + */ + let requestTokenFloor: number | undefined; + const claudeRequestTokenFloor = (): number => { + if (requestTokenFloor === undefined) { + requestTokenFloor = estimateClaudeRequestTokens(anthropicBody as Rec, requestedModel); + } + return requestTokenFloor; + }; // Routed adapters only support streamed turns; always stream internally and fold // the translated Anthropic SSE into a message JSON for non-streaming clients. internalBody.stream = true; @@ -815,7 +831,7 @@ async function handleClaudeMessagesWithBudget( // accurate-usage adapters — the request-log merge is max(reported, estimate) and // would overwrite real usage (audit 133 R1#7). if (route.provider.adapter === "cursor" || route.provider.adapter === "kiro") { - logCtx.usageLogInputTokens = estimateClaudeRequestTokens(anthropicBody as Rec, requestedModel); + logCtx.usageLogInputTokens = claudeRequestTokenFloor(); } // Effort safety valve (devlog 136 B6, audit 139 R2#2): opus-shaped aliases make // every routed model look like a reasoning model to Claude clients, so a forced @@ -988,7 +1004,13 @@ async function handleClaudeMessagesWithBudget( const contentType = response.headers.get("content-type") ?? ""; if (contentType.includes("text/event-stream") && response.body) { - const anthropicSse = responsesSseToAnthropicSse(response.body, requestedModel, { translatorBudget }); + const anthropicSse = responsesSseToAnthropicSse(response.body, requestedModel, { + translatorBudget, + // Only a floor, and only for the first frame: an upstream that reports usage early wins + // over it inside the translator, and the terminal `message_delta` carries the + // authoritative count either way (#4857). + inputTokenFloor: claudeRequestTokenFloor(), + }); if (stream) { return new Response(anthropicSse, { status: 200, diff --git a/tests/claude-integration/claude-outbound.test.ts b/tests/claude-integration/claude-outbound.test.ts index 769cae779d5..b07a7b9cf0d 100644 --- a/tests/claude-integration/claude-outbound.test.ts +++ b/tests/claude-integration/claude-outbound.test.ts @@ -20,7 +20,7 @@ const streamBudgets = new WeakMap, TranslatorBudget>( function responsesSseToAnthropicSse( upstream: ReadableStream, model: string, - opts: { pingIntervalMs?: number; translatorBudget?: TranslatorBudget } = {}, + opts: { pingIntervalMs?: number; translatorBudget?: TranslatorBudget; inputTokenFloor?: number } = {}, ): ReadableStream { const translatorBudget = opts.translatorBudget ?? createTestTranslatorBudget(); const stream = responsesSseToAnthropicSseProduction(upstream, model, { @@ -233,6 +233,44 @@ describe("claude outbound SSE", () => { cache_read_input_tokens: 100, cache_creation_input_tokens: 5, }); + expect(events.find(event => event.name === "message_delta")!.data.usage).toEqual({ + input_tokens: 15, + output_tokens: 30, + cache_read_input_tokens: 100, + cache_creation_input_tokens: 5, + }); + }); + + test("a caller-counted prompt reaches message_start when the upstream reports none early", async () => { + // The path this report came from: the internal bridge attaches usage: null to its lifecycle + // frames, so #4891 has nothing to publish and the first frame used to claim an empty prompt. + const upstream = [ + sse("response.created", { response: { id: "resp_floor", status: "in_progress", usage: null } }), + sse("response.in_progress", { response: { id: "resp_floor", status: "in_progress", usage: null } }), + sse("response.output_text.delta", { delta: "ready" }), + sse("response.completed", { + response: { + status: "completed", + usage: { + input_tokens: 120, + output_tokens: 30, + input_tokens_details: { cached_tokens: 100, cache_write_tokens: 5 }, + }, + }, + }), + ].join(""); + + const events = await collectEvents(responsesSseToAnthropicSse( + streamFrom(upstream), + "claude-ocx-test", + { inputTokenFloor: 97_000 }, + )); + expect(events.find(event => event.name === "message_start")!.data.message.usage).toEqual({ + input_tokens: 97_000, + output_tokens: 0, + }); + // The floor is a first-frame courtesy, never a claim about the upstream tokenizer: the + // terminal frame is still the authoritative count and is untouched by it. expect(events.find(event => event.name === "message_delta")!.data.usage).toEqual({ input_tokens: 15, output_tokens: 30, @@ -241,7 +279,63 @@ describe("claude outbound SSE", () => { }); }); - test("message_start documents unknown pre-content usage as zero while terminal usage stays authoritative", async () => { + test("confirmed early usage outranks the caller floor", async () => { + // An upstream measurement beats the proxy counting its own outbound prompt, always. The + // floor exists for the destinations that report nothing before content, not beside them. + const earlyUsage = { + input_tokens: 120, + output_tokens: 0, + input_tokens_details: { cached_tokens: 100, cache_write_tokens: 5 }, + }; + const upstream = [ + sse("response.in_progress", { response: { id: "resp_both", status: "in_progress", usage: earlyUsage } }), + sse("response.output_text.delta", { delta: "ready" }), + sse("response.completed", { response: { status: "completed", usage: { ...earlyUsage, output_tokens: 30 } } }), + ].join(""); + + const events = await collectEvents(responsesSseToAnthropicSse( + streamFrom(upstream), + "claude-ocx-test", + { inputTokenFloor: 97_000 }, + )); + expect(events.find(event => event.name === "message_start")!.data.message.usage).toEqual({ + input_tokens: 15, + output_tokens: 0, + cache_read_input_tokens: 100, + cache_creation_input_tokens: 5, + }); + }); + + test("a floor that measures nothing is not published", async () => { + // Zero and negative are not measurements of a prompt, and a fractional token is not a token. + // A caller that cannot count must not be able to turn that into a number on the wire. + const upstream = [ + sse("response.created", { response: { id: "resp_no_floor", status: "in_progress", usage: null } }), + sse("response.output_text.delta", { delta: "ready" }), + sse("response.completed", { response: { status: "completed", usage: { input_tokens: 5, output_tokens: 1 } } }), + ].join(""); + + for (const inputTokenFloor of [0, -1, Number.NaN]) { + const events = await collectEvents(responsesSseToAnthropicSse( + streamFrom(upstream), + "claude-ocx-test", + { inputTokenFloor }, + )); + expect({ inputTokenFloor, usage: events.find(event => event.name === "message_start")!.data.message.usage }) + .toEqual({ inputTokenFloor, usage: { input_tokens: 0, output_tokens: 0 } }); + } + + const fractional = await collectEvents(responsesSseToAnthropicSse( + streamFrom(upstream), + "claude-ocx-test", + { inputTokenFloor: 12.7 }, + )); + expect(fractional.find(event => event.name === "message_start")!.data.message.usage) + .toEqual({ input_tokens: 12, output_tokens: 0 }); + }); + + + test("message_start reports zero only when the caller supplies no measurement either", async () => { const upstream = [ sse("response.created", { response: { id: "resp_terminal_usage", status: "in_progress", usage: null } }), sse("response.output_text.delta", { delta: "ready" }), @@ -258,8 +352,17 @@ describe("claude outbound SSE", () => { ].join(""); const events = await collectEvents(responsesSseToAnthropicSse(streamFrom(upstream), "claude-ocx-test")); - // Zero is the documented honest placeholder when no input measurement has arrived. Do not - // replace it with an estimate or delay the first content frame to await terminal usage. + // No caller floor and no early upstream usage: there is nothing to report, and the + // Anthropic schema makes the field required, so zero is what goes out. This is the case the + // Lab conformance executor exercises, and it is unchanged. + // + // This case used to read "zero is the documented honest placeholder ... do not replace it + // with an estimate", and the position narrowed rather than reversed. Zero is not honest + // about a prompt that exists: it asserts an empty one, which is what showed a Paseo context + // ring at a few hundred tokens for a turn whose own context command reported ~97k (#4857). + // When the caller HAS counted the prompt it forwarded, publishing that count beats + // publishing a false one -- see the three floor cases in this file. What survives from the + // old position is its second half: the first content frame is never delayed to await usage. expect(events.find(event => event.name === "message_start")!.data.message.usage).toEqual({ input_tokens: 0, output_tokens: 0,