diff --git a/open-sse/handlers/responseSanitizer.ts b/open-sse/handlers/responseSanitizer.ts index 5c97d76e8cde..d2d139bd5bff 100644 --- a/open-sse/handlers/responseSanitizer.ts +++ b/open-sse/handlers/responseSanitizer.ts @@ -20,6 +20,7 @@ import { stripObfuscationZeroWidth } from "../utils/zeroWidth.ts"; import { normalizeArrayContentChunk } from "../utils/arrayContentDelta.ts"; import { assignAliasCacheWrite } from "../utils/pickCacheCreationTokens.ts"; import { findBalancedJsonEnd } from "./responseSanitizer/balancedJson.ts"; +import { normalizeResponsesContextUsage } from "../utils/normalizeResponsesContextUsage.ts"; export { extractThinkingFromContent, shouldParseTextualReasoningTags, @@ -561,7 +562,7 @@ function sanitizeResponsesUsage(usage: unknown): unknown { sanitized.output_tokens = outputTokens; sanitized.total_tokens = totalTokens; - return sanitized; + return normalizeResponsesContextUsage(sanitized); } /** @@ -773,6 +774,10 @@ function sanitizeResponsesStreamingEvent(parsedRecord: JsonRecord): JsonRecord { } else { delete sanitizedResponse.output_text; } + const responseUsage = toRecord(responseRecord.usage); + if (responseUsage) { + sanitizedResponse.usage = normalizeResponsesContextUsage(responseUsage); + } sanitized.response = sanitizedResponse; } diff --git a/open-sse/transformer/responsesTransformer.ts b/open-sse/transformer/responsesTransformer.ts index 014bc4215be5..3bfbe6a32c69 100644 --- a/open-sse/transformer/responsesTransformer.ts +++ b/open-sse/transformer/responsesTransformer.ts @@ -10,6 +10,7 @@ import * as fs from "fs"; import * as path from "path"; import { plaintextCollaborationFields } from "../translator/response/openai-responses/collaborationPlaintextMarker.ts"; import { finalizeResponsesTerminalStatus } from "../translator/helpers/responsesTerminalStatus.ts"; +import { normalizeResponsesContextUsage } from "../utils/normalizeResponsesContextUsage.ts"; // #10223: threshold for detecting corrupted request_id fields. Normal // request IDs are <100 chars. DeepSeek's SSE encoder bug produces 200+ @@ -613,7 +614,8 @@ export function createResponsesApiTransformStream( }; if (state.usage) { - response.usage = state.usage; + // Keep aggregation in upstream units; only emitted Responses usage includes cache writes. + response.usage = normalizeResponsesContextUsage(state.usage); } const eventType = finalizeResponsesTerminalStatus(response, state.finishReason); diff --git a/open-sse/translator/response/openai-responses.ts b/open-sse/translator/response/openai-responses.ts index 93f4ee8bbe6e..671138669ca6 100644 --- a/open-sse/translator/response/openai-responses.ts +++ b/open-sse/translator/response/openai-responses.ts @@ -12,6 +12,7 @@ import { finalizeResponsesTerminalStatus } from "../helpers/responsesTerminalSta import { shouldParseTextualReasoningTags } from "../../handlers/responseSanitizer.ts"; import { getReadableReasoningValue } from "../../utils/reasoningFields.ts"; import { resolveResponsesCacheUsageDetails } from "../../utils/resolveResponsesCacheUsageDetails.ts"; +import { normalizeResponsesContextUsage } from "../../utils/normalizeResponsesContextUsage.ts"; import { pickCacheCreationInPrompt } from "../../utils/pickCacheCreationTokens.ts"; import { isInternalReasoningPlaceholder, @@ -870,7 +871,8 @@ function sendCompleted(state, emit) { } if (state.usage) { - response.usage = state.usage; + // Preserve upstream accounting in state; expose the full context only to Responses clients. + response.usage = normalizeResponsesContextUsage(state.usage); } const eventType = finalizeResponsesTerminalStatus(response, state.finishReason, !!upstreamErr); diff --git a/open-sse/utils/normalizeResponsesContextUsage.ts b/open-sse/utils/normalizeResponsesContextUsage.ts new file mode 100644 index 000000000000..47d1e9ede1f6 --- /dev/null +++ b/open-sse/utils/normalizeResponsesContextUsage.ts @@ -0,0 +1,34 @@ +/** + * Responses clients use input_tokens as the full context size. The internal + * Claude-to-Chat billing contract can exclude cache writes and mark that fact + * explicitly; fold those writes back in only at the Responses boundary. + * Mark the result inclusive so subsequent sanitization cannot add them twice. + */ +export function normalizeResponsesContextUsage>(usage: T): T { + const details = usage.input_tokens_details; + if (!details || typeof details !== "object" || Array.isArray(details)) return usage; + const inputDetails = details as Record; + const creation = inputDetails.cache_creation_tokens; + const input = usage.input_tokens; + if ( + inputDetails.cache_creation_in_prompt !== false || + typeof creation !== "number" || + !Number.isFinite(creation) || + creation <= 0 || + typeof input !== "number" || + !Number.isFinite(input) + ) + return usage; + + const output = typeof usage.output_tokens === "number" ? usage.output_tokens : 0; + const total = + typeof usage.total_tokens === "number" && Number.isFinite(usage.total_tokens) + ? usage.total_tokens + : input + output; + return { + ...usage, + input_tokens: input + creation, + total_tokens: total + creation, + input_tokens_details: { ...inputDetails, cache_creation_in_prompt: true }, + }; +} diff --git a/tests/unit/cache-creation-marker.test.ts b/tests/unit/cache-creation-marker.test.ts index fe1a38aa5dad..06d6ae5de8d9 100644 --- a/tests/unit/cache-creation-marker.test.ts +++ b/tests/unit/cache-creation-marker.test.ts @@ -56,25 +56,27 @@ for (const included of [false, true]) { ); }); - test(`nonstream Chat -> Responses preserves ${included} without changing token counts`, () => { + test(`nonstream Responses includes the whole context for cache-in-prompt ${included}`, () => { const converted = translateNonStreamingResponse( chatBody(input, included), FORMATS.OPENAI, FORMATS.OPENAI_RESPONSES ); const result = sanitizeResponsesApiResponse(converted); - assert.equal(readPath(result, "usage", "input_tokens"), input); + assert.equal(readPath(result, "usage", "input_tokens"), 714029); + assert.equal(readPath(result, "usage", "total_tokens"), 714164); + assert.equal(readPath(converted, "usage", "prompt_tokens"), input); assert.equal( readPath(result, "usage", "input_tokens_details", "cache_creation_tokens"), 713119 ); assert.equal( readPath(result, "usage", "input_tokens_details", "cache_creation_in_prompt"), - included + true ); }); - test(`stream Chat -> Responses preserves ${included} without changing token counts`, () => { + test(`stream Responses includes context and preserves internal cache-in-prompt ${included}`, () => { const state = initState(FORMATS.OPENAI_RESPONSES); const usage = chatBody(input, included).usage; const events: unknown[] = []; @@ -104,11 +106,17 @@ for (const included of [false, true]) { .map((event) => record(event).data ?? event) .find((event) => record(event).type === "response.completed"); assert.ok(terminal, "missing terminal Responses event"); - assert.equal(readPath(terminal, "response", "usage", "input_tokens"), input); + assert.equal(readPath(terminal, "response", "usage", "input_tokens"), 714029); + assert.equal(readPath(terminal, "response", "usage", "total_tokens"), 714164); + assert.equal(readPath(state, "usage", "input_tokens"), input); assert.equal( - readPath(terminal, "response", "usage", "input_tokens_details", "cache_creation_in_prompt"), + readPath(state, "usage", "input_tokens_details", "cache_creation_in_prompt"), included ); + assert.equal( + readPath(terminal, "response", "usage", "input_tokens_details", "cache_creation_in_prompt"), + true + ); }); } @@ -235,10 +243,11 @@ for (const included of [false, true]) { assert.equal(readPath(usage, "prompt_tokens_details", "cache_creation_in_prompt"), included); }); - test(`Responses transform retains cache-in-prompt ${included}`, async () => { + test(`Responses transform includes the whole context for cache-in-prompt ${included}`, async () => { const usage = await streamUsage(chunks, createResponsesApiTransformStream()); - assert.equal(usage.input_tokens, input); - assert.equal(readPath(usage, "input_tokens_details", "cache_creation_in_prompt"), included); + assert.equal(usage.input_tokens, 714029); + assert.equal(usage.total_tokens, 714164); + assert.equal(readPath(usage, "input_tokens_details", "cache_creation_in_prompt"), true); }); } diff --git a/tests/unit/responses-context-cache-creation.test.ts b/tests/unit/responses-context-cache-creation.test.ts new file mode 100644 index 000000000000..ccc2a5e52597 --- /dev/null +++ b/tests/unit/responses-context-cache-creation.test.ts @@ -0,0 +1,252 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { + sanitizeResponsesApiResponse, + sanitizeStreamingChunk, +} from "../../open-sse/handlers/responseSanitizer.ts"; +import { openaiToOpenAIResponsesResponse } from "../../open-sse/translator/response/openai-responses.ts"; +import { createResponsesApiTransformStream } from "../../open-sse/transformer/responsesTransformer.ts"; + +import { getLoggedInputTokens } from "../../src/lib/usage/tokenAccounting.ts"; + +type Usage = Record; + +const cases = [ + { + name: "cold Claude cache reports the whole context to Responses clients", + usage: { + prompt_tokens: 769, + completion_tokens: 1120, + total_tokens: 1889, + prompt_tokens_details: { cache_creation_tokens: 170215, cache_creation_in_prompt: false }, + }, + input: 170984, + total: 172104, + }, + { + name: "mixed cache reads and writes include the write exactly once", + usage: { + prompt_tokens: 110, + completion_tokens: 7, + total_tokens: 117, + prompt_tokens_details: { + cached_tokens: 100, + cache_creation_tokens: 20, + cache_creation_in_prompt: false, + }, + }, + input: 130, + total: 137, + }, + { + name: "warm cache totals already include cached input", + usage: { + prompt_tokens: 170984, + completion_tokens: 211, + total_tokens: 171195, + prompt_tokens_details: { cached_tokens: 170215 }, + }, + input: 170984, + total: 171195, + }, + { + name: "inclusive cache writes are not counted twice", + usage: { + prompt_tokens: 130, + completion_tokens: 7, + total_tokens: 137, + prompt_tokens_details: { cache_creation_tokens: 20, cache_creation_in_prompt: true }, + }, + input: 130, + total: 137, + }, + { + name: "unmarked cache writes keep the upstream inclusion contract", + usage: { + prompt_tokens: 130, + completion_tokens: 7, + total_tokens: 137, + prompt_tokens_details: { cache_creation_tokens: 20 }, + }, + input: 130, + total: 137, + }, +]; + +function assertUsage(actual: Usage, input: number, total: number, source: Usage) { + assert.equal(actual.input_tokens, input); + assert.equal(actual.total_tokens, total); + const originalDetails = source.prompt_tokens_details as Record; + const details = actual.input_tokens_details as Record; + assert.equal(details?.cache_creation_tokens, originalDetails.cache_creation_tokens); + if (originalDetails.cache_creation_in_prompt === false) { + assert.equal(details.cache_creation_in_prompt, true); + } +} + +for (const fixture of cases) { + test(`native Responses SSE: ${fixture.name}`, () => { + const source = structuredClone(fixture.usage); + const event = sanitizeStreamingChunk({ + type: "response.completed", + response: { + id: "resp-native", + object: "response", + status: "completed", + output: [], + usage: { + input_tokens: source.prompt_tokens, + output_tokens: source.completion_tokens, + total_tokens: source.total_tokens, + input_tokens_details: source.prompt_tokens_details, + }, + }, + }) as { response: { usage: Usage } }; + assertUsage(event.response.usage, fixture.input, fixture.total, fixture.usage); + assert.deepEqual(source, fixture.usage); + }); + + test(`non-streaming: ${fixture.name}`, () => { + const source = structuredClone(fixture.usage); + const response = sanitizeResponsesApiResponse({ + id: "chatcmpl-context", + choices: [{ index: 0, message: { role: "assistant", content: "OK" }, finish_reason: "stop" }], + usage: source, + }) as { usage: Usage }; + assertUsage(response.usage, fixture.input, fixture.total, fixture.usage); + assert.deepEqual(source, fixture.usage); + const twice = sanitizeResponsesApiResponse(response) as { usage: Usage }; + assert.deepEqual( + twice.usage, + response.usage, + "a second boundary must not add cache writes again" + ); + }); + + test(`live SSE: ${fixture.name}`, () => { + const state: Record = { + seq: 0, + responseId: "resp-context", + created: 1, + started: false, + msgTextBuf: {}, + msgItemAdded: {}, + msgContentAdded: {}, + msgItemDone: {}, + reasoningId: "", + reasoningIndex: -1, + reasoningBuf: "", + reasoningPartAdded: false, + reasoningDone: false, + inThinking: false, + funcArgsBuf: {}, + funcArgsEscapeState: {}, + funcNames: {}, + funcCallIds: {}, + funcArgsDone: {}, + funcItemAdded: {}, + funcItemDone: {}, + completedOutputItems: [], + completedSent: false, + }; + const source = structuredClone(fixture.usage); + const first = openaiToOpenAIResponsesResponse( + { + id: "chatcmpl-context", + choices: [{ index: 0, delta: { content: "OK" }, finish_reason: "stop" }], + usage: source, + }, + state + ); + const events = [...(first || []), ...(openaiToOpenAIResponsesResponse(null, state) || [])]; + const completed = events.find((event) => event.event === "response.completed"); + assert.ok(completed); + assertUsage(completed.data.response.usage, fixture.input, fixture.total, fixture.usage); + assert.equal( + getLoggedInputTokens(state.usage), + fixture.usage.prompt_tokens, + "internal accounting keeps upstream units" + ); + assert.deepEqual(source, fixture.usage); + }); + + test(`legacy SSE: ${fixture.name}`, async () => { + const chunk = { + id: "chatcmpl-context", + choices: [{ index: 0, delta: { content: "OK" }, finish_reason: "stop" }], + usage: structuredClone(fixture.usage), + }; + const stream = new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode(`data: ${JSON.stringify(chunk)}\n\ndata: [DONE]\n\n`) + ); + controller.close(); + }, + }).pipeThrough(createResponsesApiTransformStream()); + const text = await new Response(stream).text(); + const events = text + .split("\n") + .filter((line) => line.startsWith("data: ") && line !== "data: [DONE]") + .map((line) => JSON.parse(line.slice(6))); + const completed = events.find((event) => event.type === "response.completed"); + assert.ok(completed); + assertUsage(completed.response.usage, fixture.input, fixture.total, fixture.usage); + }); +} + +for (const update of [ + { + name: "fresh input without repeated cache details", + usage: { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 }, + }, + { name: "output-only usage update", usage: { completion_tokens: 5 } }, + { + name: "repeated raw cache details", + usage: { + prompt_tokens: 10, + completion_tokens: 5, + total_tokens: 15, + prompt_tokens_details: { cache_creation_tokens: 100, cache_creation_in_prompt: false }, + }, + }, +]) { + test(`legacy SSE preserves cache provenance across ${update.name}`, async () => { + const chunks = [ + { + id: "chatcmpl-context", + choices: [{ index: 0, delta: { content: "OK" }, finish_reason: null }], + usage: { + prompt_tokens: 10, + completion_tokens: 0, + total_tokens: 10, + prompt_tokens_details: { cache_creation_tokens: 100, cache_creation_in_prompt: false }, + }, + }, + { choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: update.usage }, + ]; + const stream = new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + chunks.map((chunk) => `data: ${JSON.stringify(chunk)}\n\n`).join("") + + "data: [DONE]\n\n" + ) + ); + controller.close(); + }, + }).pipeThrough(createResponsesApiTransformStream()); + const text = await new Response(stream).text(); + const events = text + .split("\n") + .filter((line) => line.startsWith("data: ") && line !== "data: [DONE]") + .map((line) => JSON.parse(line.slice(6))); + const completed = events.find((event) => event.type === "response.completed"); + assert.ok(completed); + assert.equal(completed.response.usage.input_tokens, 110); + assert.equal(completed.response.usage.output_tokens, 5); + assert.equal(completed.response.usage.total_tokens, 115); + assert.equal(completed.response.usage.input_tokens_details.cache_creation_in_prompt, true); + }); +}