diff --git a/packages/__tests__/cost/usageProcessor.test.ts b/packages/__tests__/cost/usageProcessor.test.ts index 277d0056a9..424ed7e090 100644 --- a/packages/__tests__/cost/usageProcessor.test.ts +++ b/packages/__tests__/cost/usageProcessor.test.ts @@ -3,6 +3,7 @@ import { OpenAIUsageProcessor } from "@helicone-package/cost/usage/openAIUsagePr import { AnthropicUsageProcessor } from "@helicone-package/cost/usage/anthropicUsageProcessor"; import { GroqUsageProcessor } from "@helicone-package/cost/usage/groqUsageProcessor"; import { XAIUsageProcessor } from "@helicone-package/cost/usage/xaiUsageProcessor"; +import { OpenRouterUsageProcessor } from "@helicone-package/cost/usage/openRouterUsageProcessor"; import { DeepSeekUsageProcessor } from "@helicone-package/cost/usage/deepseekUsageProcessor"; import { GoogleUsageProcessor } from "@helicone-package/cost/usage/googleUsageProcessor"; import { @@ -97,7 +98,7 @@ describe("OpenAIUsageProcessor", () => { it("should parse real GPT-4o response with cached tokens", async () => { const responseData = fs.readFileSync( path.join(__dirname, "testData", "gpt4o-response-cached.snapshot"), - "utf-8" + "utf-8", ); const result = await processor.parse({ @@ -121,7 +122,7 @@ describe("OpenAIUsageProcessor", () => { it("should parse real GPT-4o stream response", async () => { const streamData = fs.readFileSync( path.join(__dirname, "testData", "gpt4o-stream-response.snapshot"), - "utf-8" + "utf-8", ); const result = await processor.parse({ @@ -137,13 +138,92 @@ describe("OpenAIUsageProcessor", () => { }); }); + it("should parse OpenAI-compatible top-level cache hit tokens", async () => { + const result = await processor.parse({ + responseBody: JSON.stringify({ + usage: { + prompt_tokens: 100, + completion_tokens: 20, + prompt_cache_hit_tokens: 40, + prompt_cache_miss_tokens: 60, + }, + }), + isStream: false, + model: "deepseek/deepseek-chat", + }); + + expect(result.error).toBeNull(); + expect(result.data).toEqual({ + input: 60, + output: 20, + cacheDetails: { + cachedInput: 40, + write5m: 0, + write1h: 0, + }, + }); + }); + + it("should derive prompt tokens from top-level cache hit and miss tokens", async () => { + const result = await processor.parse({ + responseBody: JSON.stringify({ + usage: { + completion_tokens: 20, + prompt_cache_hit_tokens: 40, + prompt_cache_miss_tokens: 60, + }, + }), + isStream: false, + model: "deepseek/deepseek-chat", + }); + + expect(result.error).toBeNull(); + expect(result.data).toEqual({ + input: 60, + output: 20, + cacheDetails: { + cachedInput: 40, + write5m: 0, + write1h: 0, + }, + }); + }); + + it("should prefer nested cached tokens over top-level cache hit tokens", async () => { + const result = await processor.parse({ + responseBody: JSON.stringify({ + usage: { + prompt_tokens: 100, + completion_tokens: 20, + prompt_cache_hit_tokens: 40, + prompt_tokens_details: { + cached_tokens: 25, + }, + }, + }), + isStream: false, + model: "deepseek/deepseek-chat", + }); + + expect(result.error).toBeNull(); + expect(result.data).toEqual({ + input: 75, + output: 20, + cacheDetails: { + cachedInput: 25, + write5m: 0, + write1h: 0, + }, + }); + }); + it("usage processing snapshot", async () => { const testCases = [ { name: "cached-response", data: fs.readFileSync( path.join(__dirname, "testData", "gpt4o-response-cached.snapshot"), - "utf-8" + "utf-8", ), isStream: false, }, @@ -151,7 +231,7 @@ describe("OpenAIUsageProcessor", () => { name: "stream-response", data: fs.readFileSync( path.join(__dirname, "testData", "gpt4o-stream-response.snapshot"), - "utf-8" + "utf-8", ), isStream: true, }, @@ -172,13 +252,46 @@ describe("OpenAIUsageProcessor", () => { }); }); +describe("OpenRouterUsageProcessor", () => { + const processor = new OpenRouterUsageProcessor(); + + it("should preserve top-level cache hit tokens for analytics", async () => { + const result = await processor.parse({ + responseBody: JSON.stringify({ + usage: { + prompt_tokens: 100, + completion_tokens: 20, + prompt_cache_hit_tokens: 40, + prompt_cache_miss_tokens: 60, + cost: 0.001, + }, + }), + isStream: false, + model: "deepseek/deepseek-chat", + }); + + expect(result.error).toBeNull(); + expect(result.data).toEqual({ + input: 60, + output: 20, + cost: 0.001, + cost_details: undefined, + provider: undefined, + is_byok: undefined, + cacheDetails: { + cachedInput: 40, + }, + }); + }); +}); + describe("Azure Usage Processing", () => { const processor = new OpenAIUsageProcessor(); // Azure uses OpenAI processor it("should parse Azure regular response", async () => { const responseData = fs.readFileSync( path.join(__dirname, "testData", "azure-response.snapshot"), - "utf-8" + "utf-8", ); const result = await processor.parse({ @@ -197,7 +310,7 @@ describe("Azure Usage Processing", () => { it("should parse Azure stream response", async () => { const streamData = fs.readFileSync( path.join(__dirname, "testData", "azure-stream-response.snapshot"), - "utf-8" + "utf-8", ); const result = await processor.parse({ @@ -220,7 +333,7 @@ describe("Azure Usage Processing", () => { name: "azure-response", data: fs.readFileSync( path.join(__dirname, "testData", "azure-response.snapshot"), - "utf-8" + "utf-8", ), isStream: false, }, @@ -228,7 +341,7 @@ describe("Azure Usage Processing", () => { name: "azure-stream-response", data: fs.readFileSync( path.join(__dirname, "testData", "azure-stream-response.snapshot"), - "utf-8" + "utf-8", ), isStream: true, }, @@ -255,7 +368,7 @@ describe("AnthropicUsageProcessor", () => { it("should parse Anthropic response with cache details", async () => { const responseData = fs.readFileSync( path.join(__dirname, "testData", "anthropic-response.snapshot"), - "utf-8" + "utf-8", ); const result = await processor.parse({ @@ -278,7 +391,7 @@ describe("AnthropicUsageProcessor", () => { it("should parse Anthropic stream response", async () => { const streamData = fs.readFileSync( path.join(__dirname, "testData", "anthropic-stream-response.snapshot"), - "utf-8" + "utf-8", ); const result = await processor.parse({ @@ -304,7 +417,7 @@ describe("AnthropicUsageProcessor", () => { name: "anthropic-response", data: fs.readFileSync( path.join(__dirname, "testData", "anthropic-response.snapshot"), - "utf-8" + "utf-8", ), isStream: false, }, @@ -314,9 +427,9 @@ describe("AnthropicUsageProcessor", () => { path.join( __dirname, "testData", - "anthropic-stream-response.snapshot" + "anthropic-stream-response.snapshot", ), - "utf-8" + "utf-8", ), isStream: true, }, @@ -343,7 +456,7 @@ describe("XAI/Grok specific features", () => { it("should parse XAI response with web search", async () => { const xaiResponse = fs.readFileSync( path.join(__dirname, "testData", "xai-response-websearch.snapshot"), - "utf-8" + "utf-8", ); const result = await xaiProcessor.parse({ @@ -366,7 +479,7 @@ describe("XAI/Grok specific features", () => { it("should parse XAI response with reasoning tokens", async () => { const xaiResponse = fs.readFileSync( path.join(__dirname, "testData", "xai-response-reasoning.snapshot"), - "utf-8" + "utf-8", ); const result = await xaiProcessor.parse({ @@ -391,7 +504,7 @@ describe("XAI/Grok specific features", () => { it("should parse XAI stream response with web search", async () => { const streamData = fs.readFileSync( path.join(__dirname, "testData", "xai-stream-response.snapshot"), - "utf-8" + "utf-8", ); const result = await xaiProcessor.parse({ @@ -418,7 +531,7 @@ describe("Groq specific features", () => { it("should parse Groq non-streaming response", async () => { const groqResponse = fs.readFileSync( path.join(__dirname, "testData", "groq-response.snapshot"), - "utf-8" + "utf-8", ); const result = await groqProcessor.parse({ @@ -437,7 +550,7 @@ describe("Groq specific features", () => { it("should parse Groq streaming response with usage in x_groq", async () => { const streamData = fs.readFileSync( path.join(__dirname, "testData", "groq-stream-response.snapshot"), - "utf-8" + "utf-8", ); const result = await groqProcessor.parse({ @@ -459,7 +572,7 @@ describe("Groq specific features", () => { it("should parse real DeepSeek non-streaming response", async () => { const responseData = fs.readFileSync( path.join(__dirname, "testData", "deepseek-non-stream.snapshot"), - "utf-8" + "utf-8", ); const result = await deepseekProcessor.parse({ @@ -478,7 +591,7 @@ describe("Groq specific features", () => { it("should parse DeepSeek response with cache hits", async () => { const responseData = fs.readFileSync( path.join(__dirname, "testData", "deepseek-cached.snapshot"), - "utf-8" + "utf-8", ); const result = await deepseekProcessor.parse({ @@ -500,7 +613,7 @@ describe("Groq specific features", () => { it("should parse DeepSeek reasoner response with thinking tokens", async () => { const responseData = fs.readFileSync( path.join(__dirname, "testData", "deepseek-reasoner.snapshot"), - "utf-8" + "utf-8", ); const result = await deepseekProcessor.parse({ @@ -523,7 +636,7 @@ describe("Groq specific features", () => { it("should parse DeepSeek streaming response", async () => { const streamData = fs.readFileSync( path.join(__dirname, "testData", "deepseek-stream.snapshot"), - "utf-8" + "utf-8", ); const result = await deepseekProcessor.parse({ @@ -677,9 +790,7 @@ describe("VertexUsageProcessor", () => { { modality: "TEXT", tokenCount: 6 }, { modality: "IMAGE", tokenCount: 8 }, ], - candidatesTokensDetails: [ - { modality: "TEXT", tokenCount: 19 }, - ], + candidatesTokensDetails: [{ modality: "TEXT", tokenCount: 19 }], }, modelVersion: "gemini-2.5-flash", responseId: "abc", diff --git a/packages/cost/usage/cacheTokenUtils.ts b/packages/cost/usage/cacheTokenUtils.ts new file mode 100644 index 0000000000..25f00e9efa --- /dev/null +++ b/packages/cost/usage/cacheTokenUtils.ts @@ -0,0 +1,86 @@ +type UsageDetails = { + cached_tokens?: number; + audio_tokens?: number; + cache_write_tokens?: number; + cache_write_details?: { + write_5m_tokens?: number; + write_1h_tokens?: number; + }; +}; + +type CacheTokenUsage = { + promptTokens: number; + cachedTokens: number; + promptAudioTokens: number; + cacheWrite5mTokens: number; + cacheWrite1hTokens: number; +}; + +function toFiniteNumber(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) + ? value + : undefined; +} + +function firstFiniteNumber(...values: unknown[]): number | undefined { + for (const value of values) { + const numberValue = toFiniteNumber(value); + if (numberValue !== undefined) { + return numberValue; + } + } + return undefined; +} + +export function getCacheTokenUsage(usage: any): CacheTokenUsage { + const promptDetails: UsageDetails = + usage?.prompt_tokens_details || usage?.input_tokens_details || {}; + + const cacheHitTokens = + firstFiniteNumber( + promptDetails.cached_tokens, + usage?.cache_read_input_tokens, + usage?.prompt_cache_hit_tokens, + ) ?? 0; + + const cacheMissTokens = toFiniteNumber(usage?.prompt_cache_miss_tokens); + const promptTokens = + firstFiniteNumber( + usage?.prompt_tokens, + usage?.input_tokens, + cacheMissTokens !== undefined + ? cacheHitTokens + cacheMissTokens + : undefined, + ) ?? 0; + + const cacheWriteTokensTotal = + firstFiniteNumber( + promptDetails.cache_write_tokens, + usage?.cache_creation_input_tokens, + ) ?? 0; + const cacheWriteDetails = promptDetails.cache_write_details; + + return { + promptTokens, + cachedTokens: cacheHitTokens, + promptAudioTokens: firstFiniteNumber(promptDetails.audio_tokens) ?? 0, + cacheWrite5mTokens: + firstFiniteNumber(cacheWriteDetails?.write_5m_tokens) ?? + cacheWriteTokensTotal, + cacheWrite1hTokens: + firstFiniteNumber(cacheWriteDetails?.write_1h_tokens) ?? 0, + }; +} + +export function getEffectivePromptTokens({ + promptTokens, + cachedTokens, + promptAudioTokens, +}: Pick< + CacheTokenUsage, + "promptTokens" | "cachedTokens" | "promptAudioTokens" +>): number { + return cachedTokens > promptTokens + ? Math.max(0, promptTokens - promptAudioTokens) + : Math.max(0, promptTokens - cachedTokens - promptAudioTokens); +} diff --git a/packages/cost/usage/openAIUsageProcessor.ts b/packages/cost/usage/openAIUsageProcessor.ts index ca69b10adb..679b01da04 100644 --- a/packages/cost/usage/openAIUsageProcessor.ts +++ b/packages/cost/usage/openAIUsageProcessor.ts @@ -1,6 +1,10 @@ import { IUsageProcessor, ParseInput } from "./IUsageProcessor"; import { ModelUsage } from "./types"; import { Result } from "../../common/result"; +import { + getCacheTokenUsage, + getEffectivePromptTokens, +} from "./cacheTokenUtils"; export class OpenAIUsageProcessor implements IUsageProcessor { public async parse( @@ -140,33 +144,27 @@ export class OpenAIUsageProcessor implements IUsageProcessor { // OpenAIUsage from "@helicone-package/llm-mapper/transform/types/common"; // ResponsesUsage from "@helicone-package/llm-mapper/transform/types/responses"; - const promptTokens = usage.prompt_tokens ?? usage.input_tokens ?? 0; const completionTokens = usage.completion_tokens ?? usage.output_tokens ?? 0; - const promptDetails = - usage.prompt_tokens_details || usage.input_tokens_details || {}; const completionDetails = usage.completion_tokens_details || usage.output_tokens_details || {}; - const cachedTokens = promptDetails.cached_tokens ?? 0; - const promptAudioTokens = promptDetails.audio_tokens ?? 0; + const { + promptTokens, + cachedTokens, + promptAudioTokens, + cacheWrite5mTokens, + cacheWrite1hTokens, + } = getCacheTokenUsage(usage); const completionAudioTokens = completionDetails.audio_tokens ?? 0; const reasoningTokens = completionDetails.reasoning_tokens ?? 0; - // AI Gateway fields - cache write tokens - // First try to get the detailed breakdown (5m vs 1h), then fall back to total cache_write_tokens - const cacheWriteDetails = promptDetails.cache_write_details; - const cacheWriteTokensTotal = promptDetails.cache_write_tokens ?? 0; - - // If we have detailed breakdown, use it; otherwise treat all cache writes as 5m (the common case) - const cacheWrite5mTokens = cacheWriteDetails?.write_5m_tokens ?? cacheWriteTokensTotal; - const cacheWrite1hTokens = cacheWriteDetails?.write_1h_tokens ?? 0; - - // Guard: if cached > prompt_tokens, data is already non-cached (Anthropic convention) - const effectivePromptTokens = cachedTokens > promptTokens - ? Math.max(0, promptTokens - promptAudioTokens) - : Math.max(0, promptTokens - cachedTokens - promptAudioTokens); + const effectivePromptTokens = getEffectivePromptTokens({ + promptTokens, + cachedTokens, + promptAudioTokens, + }); const effectiveCompletionTokens = Math.max( 0, completionTokens - completionAudioTokens - reasoningTokens, diff --git a/packages/cost/usage/openRouterUsageProcessor.ts b/packages/cost/usage/openRouterUsageProcessor.ts index 44336a0008..47a9c14955 100644 --- a/packages/cost/usage/openRouterUsageProcessor.ts +++ b/packages/cost/usage/openRouterUsageProcessor.ts @@ -1,6 +1,10 @@ import { IUsageProcessor, ParseInput } from "./IUsageProcessor"; import { ModelUsage } from "./types"; import { Result } from "../../common/result"; +import { + getCacheTokenUsage, + getEffectivePromptTokens, +} from "./cacheTokenUtils"; export interface OpenRouterCostDetails { upstream_inference_cost?: number; @@ -10,7 +14,7 @@ export interface OpenRouterCostDetails { export function getOpenRouterDeclaredCost( cost?: number, - cost_details?: OpenRouterCostDetails + cost_details?: OpenRouterCostDetails, ): number | undefined { // Priority 1: Direct cost field if (cost && cost > 0) { @@ -18,7 +22,10 @@ export function getOpenRouterDeclaredCost( } // Priority 2: Upstream inference cost - if (cost_details?.upstream_inference_cost && cost_details.upstream_inference_cost > 0) { + if ( + cost_details?.upstream_inference_cost && + cost_details.upstream_inference_cost > 0 + ) { return cost_details.upstream_inference_cost; } @@ -29,7 +36,10 @@ export function getOpenRouterDeclaredCost( cost_details.upstream_inference_prompt_cost > 0 && cost_details.upstream_inference_completions_cost > 0 ) { - return cost_details.upstream_inference_prompt_cost + cost_details.upstream_inference_completions_cost; + return ( + cost_details.upstream_inference_prompt_cost + + cost_details.upstream_inference_completions_cost + ); } return undefined; @@ -42,7 +52,9 @@ export interface OpenRouterUsage extends ModelUsage { } export class OpenRouterUsageProcessor implements IUsageProcessor { - public async parse(parseInput: ParseInput): Promise> { + public async parse( + parseInput: ParseInput, + ): Promise> { try { if (parseInput.isStream) { return this.parseStreamResponse(parseInput.responseBody); @@ -57,7 +69,9 @@ export class OpenRouterUsageProcessor implements IUsageProcessor { } } - protected parseNonStreamResponse(responseBody: string): Result { + protected parseNonStreamResponse( + responseBody: string, + ): Result { try { const parsedResponse = JSON.parse(responseBody); const usage = this.extractUsageFromResponse(parsedResponse); @@ -73,11 +87,16 @@ export class OpenRouterUsageProcessor implements IUsageProcessor { } } - protected parseStreamResponse(responseBody: string): Result { + protected parseStreamResponse( + responseBody: string, + ): Result { try { const lines = responseBody .split("\n") - .filter((line) => line.trim() !== "" && !line.includes("OPENROUTER PROCESSING")) + .filter( + (line) => + line.trim() !== "" && !line.includes("OPENROUTER PROCESSING"), + ) .map((line) => { if (line === "data: [DONE]") return null; try { @@ -105,7 +124,9 @@ export class OpenRouterUsageProcessor implements IUsageProcessor { protected consolidateStreamData(streamData: any[]): any { // Look for the last chunk with usage data - const lastChunkWithUsage = [...streamData].reverse().find(chunk => chunk?.usage); + const lastChunkWithUsage = [...streamData] + .reverse() + .find((chunk) => chunk?.usage); if (lastChunkWithUsage?.usage) { return lastChunkWithUsage; } @@ -148,20 +169,26 @@ export class OpenRouterUsageProcessor implements IUsageProcessor { const is_byok = usage.is_byok; // OpenRouter still provides token counts for compatibility - const promptTokens = usage.prompt_tokens ?? usage.input_tokens ?? 0; - const completionTokens = usage.completion_tokens ?? usage.output_tokens ?? 0; + const completionTokens = + usage.completion_tokens ?? usage.output_tokens ?? 0; - const promptDetails = usage.prompt_tokens_details || {}; const completionDetails = usage.completion_tokens_details || {}; - const cachedTokens = promptDetails.cached_tokens ?? 0; - const promptAudioTokens = promptDetails.audio_tokens ?? 0; + const { promptTokens, cachedTokens, promptAudioTokens } = + getCacheTokenUsage(usage); const completionAudioTokens = completionDetails.audio_tokens ?? 0; const reasoningTokens = completionDetails.reasoning_tokens ?? 0; // Calculate effective tokens (for logging/analytics, not for cost) - const effectivePromptTokens = Math.max(0, promptTokens - cachedTokens - promptAudioTokens); - const effectiveCompletionTokens = Math.max(0, completionTokens - completionAudioTokens - reasoningTokens); + const effectivePromptTokens = getEffectivePromptTokens({ + promptTokens, + cachedTokens, + promptAudioTokens, + }); + const effectiveCompletionTokens = Math.max( + 0, + completionTokens - completionAudioTokens - reasoningTokens, + ); // Get declared cost and apply passthrough billing markup if needed let declaredCost = getOpenRouterDeclaredCost(cost, cost_details); @@ -196,4 +223,4 @@ export class OpenRouterUsageProcessor implements IUsageProcessor { return modelUsage; } -} \ No newline at end of file +} diff --git a/valhalla/jawn/src/lib/shared/bodyProcessors/__tests__/genericBodyProcessor.test.ts b/valhalla/jawn/src/lib/shared/bodyProcessors/__tests__/genericBodyProcessor.test.ts new file mode 100644 index 0000000000..e40462c42d --- /dev/null +++ b/valhalla/jawn/src/lib/shared/bodyProcessors/__tests__/genericBodyProcessor.test.ts @@ -0,0 +1,60 @@ +import { GenericBodyProcessor } from "../genericBodyProcessor"; +import { ParseInput } from "../IBodyProcessor"; + +describe("GenericBodyProcessor", () => { + const processor = new GenericBodyProcessor(); + + async function parse(body: any): Promise { + const input: ParseInput = { + responseBody: JSON.stringify(body), + }; + + const result = await processor.parse(input); + expect(result.error).toBeNull(); + return result.data; + } + + it("handles OpenAI-compatible top-level cache hit tokens", async () => { + const { usage } = await parse({ + usage: { + prompt_tokens: 100, + completion_tokens: 20, + prompt_cache_hit_tokens: 40, + prompt_cache_miss_tokens: 60, + }, + }); + + expect(usage).toEqual({ + promptTokens: 60, + promptCacheReadTokens: 40, + promptCacheWriteTokens: 0, + completionTokens: 20, + totalTokens: undefined, + heliconeCalculated: false, + cost: undefined, + }); + }); + + it("prefers nested cached tokens over top-level cache hit tokens", async () => { + const { usage } = await parse({ + usage: { + prompt_tokens: 100, + completion_tokens: 20, + prompt_cache_hit_tokens: 40, + prompt_tokens_details: { + cached_tokens: 25, + }, + }, + }); + + expect(usage).toEqual({ + promptTokens: 75, + promptCacheReadTokens: 25, + promptCacheWriteTokens: 0, + completionTokens: 20, + totalTokens: undefined, + heliconeCalculated: false, + cost: undefined, + }); + }); +}); diff --git a/valhalla/jawn/src/lib/shared/bodyProcessors/__tests__/openAIStreamProcessor.test.ts b/valhalla/jawn/src/lib/shared/bodyProcessors/__tests__/openAIStreamProcessor.test.ts new file mode 100644 index 0000000000..49544ea821 --- /dev/null +++ b/valhalla/jawn/src/lib/shared/bodyProcessors/__tests__/openAIStreamProcessor.test.ts @@ -0,0 +1,43 @@ +import { OpenAIStreamProcessor } from "../openAIStreamProcessor"; +import { ParseInput } from "../IBodyProcessor"; + +describe("OpenAIStreamProcessor", () => { + const processor = new OpenAIStreamProcessor(); + + async function parseFromLines(lines: any[]): Promise { + const responseBody = lines + .map((line) => `data: ${JSON.stringify(line)}`) + .join("\n"); + + const input: ParseInput = { + responseBody, + }; + + const result = await processor.parse(input); + expect(result.error).toBeNull(); + return result.data; + } + + it("handles OpenAI-compatible top-level cache hit tokens in streamed usage", async () => { + const { usage } = await parseFromLines([ + { + usage: { + prompt_tokens: 100, + completion_tokens: 20, + prompt_cache_hit_tokens: 40, + prompt_cache_miss_tokens: 60, + }, + }, + ]); + + expect(usage).toEqual({ + totalTokens: undefined, + completionTokens: 20, + promptTokens: 60, + promptCacheReadTokens: 40, + promptCacheWriteTokens: 0, + heliconeCalculated: false, + cost: undefined, + }); + }); +}); diff --git a/valhalla/jawn/src/lib/shared/bodyProcessors/genericBodyProcessor.ts b/valhalla/jawn/src/lib/shared/bodyProcessors/genericBodyProcessor.ts index 4ec7831afb..a56f2fae99 100644 --- a/valhalla/jawn/src/lib/shared/bodyProcessors/genericBodyProcessor.ts +++ b/valhalla/jawn/src/lib/shared/bodyProcessors/genericBodyProcessor.ts @@ -1,10 +1,14 @@ import { Usage } from "../../handlers/HandlerContext"; import { PromiseGenericResult, ok } from "../../../packages/common/result"; import { IBodyProcessor, ParseInput, ParseOutput } from "./IBodyProcessor"; +import { + getCacheTokenUsage, + getEffectivePromptTokens, +} from "@helicone-package/cost/usage/cacheTokenUtils"; export class GenericBodyProcessor implements IBodyProcessor { public async parse( - parseInput: ParseInput + parseInput: ParseInput, ): PromiseGenericResult { const parsedResponseBody = JSON.parse(parseInput.responseBody); @@ -54,6 +58,10 @@ export class GenericBodyProcessor implements IBodyProcessor { cached_tokens?: number; audio_tokens?: number; cache_write_tokens?: number; + cache_write_details?: { + write_5m_tokens?: number; + write_1h_tokens?: number; + }; }; completion_tokens_details?: { reasoning_tokens?: number; @@ -61,7 +69,7 @@ export class GenericBodyProcessor implements IBodyProcessor { accepted_prediction_tokens?: number; rejected_prediction_tokens?: number; }; - + // OpenAI Responses API input_tokens?: number; output_tokens?: number; @@ -74,32 +82,53 @@ export class GenericBodyProcessor implements IBodyProcessor { // OpenRouter cost?: number; + + // Anthropic and DeepSeek-compatible cache fields + cache_creation_input_tokens?: number; + cache_read_input_tokens?: number; + prompt_cache_hit_tokens?: number; + prompt_cache_miss_tokens?: number; }; }; // OpenAI charges for input, input cache read, output, output audio, input audio. - // Guard: if cached > prompt_tokens, data is already non-cached (Anthropic convention) const usage = response.usage; - const gPromptToks = usage?.prompt_tokens ?? usage?.input_tokens ?? 0; - const gCachedToks = usage?.prompt_tokens_details?.cached_tokens ?? usage?.input_tokens_details?.cached_tokens ?? 0; - const gAudioToks = usage?.prompt_tokens_details?.audio_tokens ?? 0; - const effectivePromptTokens = gCachedToks > gPromptToks - ? Math.max(0, gPromptToks - gAudioToks) - : Math.max(0, gPromptToks - gCachedToks - gAudioToks); - const effectiveCompletionTokens = usage?.completion_tokens !== undefined - ? Math.max(0, (usage.completion_tokens ?? 0) - (usage.completion_tokens_details?.reasoning_tokens ?? 0) - (usage.completion_tokens_details?.audio_tokens ?? 0)) - : Math.max(0, (usage.output_tokens ?? 0) - (usage.output_tokens_details?.reasoning_tokens ?? 0)); - + const { + promptTokens, + cachedTokens, + promptAudioTokens, + cacheWrite5mTokens, + cacheWrite1hTokens, + } = getCacheTokenUsage(usage); + const effectivePromptTokens = getEffectivePromptTokens({ + promptTokens, + cachedTokens, + promptAudioTokens, + }); + const effectiveCompletionTokens = + usage?.completion_tokens !== undefined + ? Math.max( + 0, + (usage.completion_tokens ?? 0) - + (usage.completion_tokens_details?.reasoning_tokens ?? 0) - + (usage.completion_tokens_details?.audio_tokens ?? 0), + ) + : Math.max( + 0, + (usage.output_tokens ?? 0) - + (usage.output_tokens_details?.reasoning_tokens ?? 0), + ); + return { promptTokens: effectivePromptTokens, - promptCacheReadTokens: usage?.prompt_tokens_details?.cached_tokens ?? usage?.input_tokens_details?.cached_tokens ?? 0, - promptCacheWriteTokens: usage?.prompt_tokens_details?.cache_write_tokens ?? 0, + promptCacheReadTokens: cachedTokens, + promptCacheWriteTokens: cacheWrite5mTokens + cacheWrite1hTokens, completionTokens: effectiveCompletionTokens, totalTokens: usage?.total_tokens, heliconeCalculated: false, // OpenRouter may contain these fields based on wallet/BYOK setup - cost: usage?.cost + cost: usage?.cost, }; } } diff --git a/valhalla/jawn/src/lib/shared/bodyProcessors/openAIStreamProcessor.ts b/valhalla/jawn/src/lib/shared/bodyProcessors/openAIStreamProcessor.ts index 1ba032eb48..c4612af075 100644 --- a/valhalla/jawn/src/lib/shared/bodyProcessors/openAIStreamProcessor.ts +++ b/valhalla/jawn/src/lib/shared/bodyProcessors/openAIStreamProcessor.ts @@ -5,6 +5,10 @@ import { import { PromiseGenericResult, err, ok } from "../../../packages/common/result"; import { IBodyProcessor, ParseInput, ParseOutput } from "./IBodyProcessor"; import { isParseInputJson } from "./helpers"; +import { + getCacheTokenUsage, + getEffectivePromptTokens, +} from "@helicone-package/cost/usage/cacheTokenUtils"; export const NON_DATA_LINES = [ "event: content_block_delta", @@ -91,7 +95,7 @@ export class OpenAIStreamProcessor implements IBodyProcessor { const isResponsesAPI = data.some( (item) => item?.type === "response.created" || - item?.type === "response.completed" + item?.type === "response.completed", ); if (isResponsesAPI) { @@ -101,26 +105,25 @@ export class OpenAIStreamProcessor implements IBodyProcessor { let usage; if (usageData) { - // Responses API uses input_tokens/output_tokens - // Guard: if cached > input_tokens, data is already non-cached (Anthropic convention) - const rInputToks = usageData.input_tokens ?? 0; - const rCachedToks = usageData.input_tokens_details?.cached_tokens ?? 0; - const effectivePromptTokens = rCachedToks > rInputToks - ? rInputToks - : Math.max(0, rInputToks - rCachedToks); + const { promptTokens, cachedTokens, promptAudioTokens } = + getCacheTokenUsage(usageData); + const effectivePromptTokens = getEffectivePromptTokens({ + promptTokens, + cachedTokens, + promptAudioTokens, + }); const effectiveCompletionTokens = Math.max( 0, (usageData.output_tokens ?? 0) - - (usageData.output_tokens_details?.reasoning_tokens ?? 0) + (usageData.output_tokens_details?.reasoning_tokens ?? 0), ); usage = { totalTokens: usageData.total_tokens, completionTokens: effectiveCompletionTokens, promptTokens: effectivePromptTokens, - promptCacheReadTokens: - usageData.input_tokens_details?.cached_tokens ?? 0, + promptCacheReadTokens: cachedTokens, heliconeCalculated: false, }; } else { @@ -152,14 +155,18 @@ export class OpenAIStreamProcessor implements IBodyProcessor { let usage; if (usageData) { - // Guard: if cached > prompt_tokens, data is already non-cached (Anthropic convention) - const promptToks = usageData.prompt_tokens ?? usageData.input_tokens ?? 0; - const cachedToks = usageData.prompt_tokens_details?.cached_tokens - ?? usageData.input_tokens_details?.cached_tokens ?? 0; - const audioToks = usageData.prompt_tokens_details?.audio_tokens ?? 0; - const effectivePromptTokens = cachedToks > promptToks - ? Math.max(0, promptToks - audioToks) - : Math.max(0, promptToks - cachedToks - audioToks); + const { + promptTokens, + cachedTokens, + promptAudioTokens, + cacheWrite5mTokens, + cacheWrite1hTokens, + } = getCacheTokenUsage(usageData); + const effectivePromptTokens = getEffectivePromptTokens({ + promptTokens, + cachedTokens, + promptAudioTokens, + }); const effectiveCompletionTokens = usageData?.completion_tokens !== undefined @@ -167,24 +174,20 @@ export class OpenAIStreamProcessor implements IBodyProcessor { 0, (usageData.completion_tokens ?? 0) - (usageData.completion_tokens_details?.reasoning_tokens ?? 0) - - (usageData.completion_tokens_details?.audio_tokens ?? 0) + (usageData.completion_tokens_details?.audio_tokens ?? 0), ) : Math.max( 0, (usageData.output_tokens ?? 0) - - (usageData.output_tokens_details?.reasoning_tokens ?? 0) + (usageData.output_tokens_details?.reasoning_tokens ?? 0), ); usage = { totalTokens: usageData?.total_tokens, completionTokens: effectiveCompletionTokens, promptTokens: effectivePromptTokens, - promptCacheReadTokens: - usageData?.prompt_tokens_details?.cached_tokens ?? - usageData?.input_tokens_details?.cached_tokens ?? - 0, - promptCacheWriteTokens: - usageData?.prompt_tokens_details?.cache_write_tokens ?? 0, + promptCacheReadTokens: cachedTokens, + promptCacheWriteTokens: cacheWrite5mTokens + cacheWrite1hTokens, heliconeCalculated: usageData?.helicone_calculated ?? false, // OpenRouter may contain these fields based on wallet/BYOK setup diff --git a/worker/src/lib/dbLogger/DBLoggable.ts b/worker/src/lib/dbLogger/DBLoggable.ts index 3ec9de002c..a138753d34 100644 --- a/worker/src/lib/dbLogger/DBLoggable.ts +++ b/worker/src/lib/dbLogger/DBLoggable.ts @@ -38,6 +38,10 @@ import { } from "../../RequestBodyBuffer/IRequestBodyBuffer"; import { ModelProviderName } from "@helicone-package/cost/models/providers"; import { BodyMappingType } from "@helicone-package/cost/models/types"; +import { + getCacheTokenUsage, + getEffectivePromptTokens, +} from "@helicone-package/cost/usage/cacheTokenUtils"; export interface DBLoggableProps { response: { @@ -441,7 +445,9 @@ export class DBLoggable { prompt_tokens: usage?.prompt_tokens ?? usage?.input_tokens ?? usage?.inputTokens, completion_tokens: - usage?.completion_tokens ?? usage?.output_tokens ?? usage?.outputTokens, + usage?.completion_tokens ?? + usage?.output_tokens ?? + usage?.outputTokens, }; } @@ -495,6 +501,10 @@ export class DBLoggable { cached_tokens?: number; audio_tokens?: number; }; + input_tokens_details?: { + cached_tokens?: number; + audio_tokens?: number; + }; completion_tokens_details?: { reasoning_tokens?: number; audio_tokens?: number; @@ -502,26 +512,33 @@ export class DBLoggable { // Anthropic cache usage cache_creation_input_tokens?: number; cache_read_input_tokens?: number; + // DeepSeek-compatible cache usage + prompt_cache_hit_tokens?: number; + prompt_cache_miss_tokens?: number; }; }; const usage = response.usage; + const { + promptTokens, + cachedTokens, + promptAudioTokens, + cacheWrite5mTokens, + cacheWrite1hTokens, + } = getCacheTokenUsage(usage); return { - prompt_tokens: - usage?.prompt_tokens ?? usage?.input_tokens, - completion_tokens: - usage?.completion_tokens ?? usage?.output_tokens, - prompt_cache_read_tokens: - usage?.prompt_tokens_details?.cached_tokens ?? - usage?.cache_read_input_tokens, + prompt_tokens: getEffectivePromptTokens({ + promptTokens, + cachedTokens, + promptAudioTokens, + }), + completion_tokens: usage?.completion_tokens ?? usage?.output_tokens, + prompt_cache_read_tokens: cachedTokens || undefined, prompt_cache_write_tokens: - usage?.cache_creation_input_tokens, - prompt_audio_tokens: - usage?.prompt_tokens_details?.audio_tokens, - completion_audio_tokens: - usage?.completion_tokens_details?.audio_tokens, - reasoning_tokens: - usage?.completion_tokens_details?.reasoning_tokens, + cacheWrite5mTokens + cacheWrite1hTokens || undefined, + prompt_audio_tokens: promptAudioTokens || undefined, + completion_audio_tokens: usage?.completion_tokens_details?.audio_tokens, + reasoning_tokens: usage?.completion_tokens_details?.reasoning_tokens, }; } @@ -821,7 +838,8 @@ export class DBLoggable { const parsedResponse = JSON.parse(responseText); extractedUsage = this.getDetailedUsage(parsedResponse); // Check if we actually got usage tokens - failedToGetUsage = !extractedUsage.prompt_tokens && !extractedUsage.completion_tokens; + failedToGetUsage = + !extractedUsage.prompt_tokens && !extractedUsage.completion_tokens; // Extract model from response (OpenAI format) if ( typeof parsedResponse === "object" && @@ -861,7 +879,8 @@ export class DBLoggable { const responseStatus = await this.response.status(); if (responseStatus < 400) { try { - const bodyMapping = this.request.attempt?.endpoint.userConfig?.gatewayMapping; + const bodyMapping = + this.request.attempt?.endpoint.userConfig?.gatewayMapping; // Normalize response and convert to user's requested format (OPENAI or RESPONSES) openAIResponse = await normalizeAIGatewayResponse({ @@ -925,7 +944,8 @@ export class DBLoggable { gatewayProvider = provider as ModelProviderName; gatewayModel = model as string; - aiGatewayBodyMapping = this.request.attempt?.endpoint.userConfig?.gatewayMapping ?? "OPENAI"; + aiGatewayBodyMapping = + this.request.attempt?.endpoint.userConfig?.gatewayMapping ?? "OPENAI"; } const kafkaMessage: MessageData = {