Merge pull request #1681 from songkuan-zheng/fix/usage-cache-convention-by-source

Normalize usage token conventions per source

packages/core/src/observability/request-log-store.ts

@@ -533,11 +533,19 @@ export class RequestLogStore {
     const responseError = normalizeFilterValue(input.error) ??
       detectSseError(responseBodyText, headerValue(responseHeaders, "content-type"));
     const bodyUsage = extractUsageFromBody(responseBodyText);
-    const usage: UsageSnapshot = normalizeUsageInputTokens(mergeUsageSnapshots(extractUsageFromBillingHeaders(input.responseHeaders), bodyUsage), {
-      path: input.path,
-      providerProtocol: input.providerProtocol,
-      usageHint: bodyUsage
-    }) ?? {};
+    // Each source carries its own cache-inclusion convention; normalize before
+    // merging (see UsageConventionSource).
+    const usage: UsageSnapshot = mergeUsageSnapshots(
+      normalizeUsageInputTokens(extractUsageFromBillingHeaders(input.responseHeaders), {
+        path: input.path,
+        providerProtocol: input.providerProtocol,
+        source: "providerBilling"
+      }),
+      normalizeUsageInputTokens(bodyUsage, {
+        path: input.path,
+        source: "responseBody"
+      })
+    ) ?? {};
     const route = splitRequestLogRouteSelector(input.fallbackModel);
     const bodyModel = requestLogRequestedModel(input.requestBody, input.path);
     const requestModel = normalizeFilterValue(input.model) ?? bodyModel;
@@ -868,10 +876,20 @@ export class RequestLogStore {
       const bodyUsage = input.responseBodyText === undefined
         ? undefined
         : extractUsageFromBody(input.responseBodyText);
-      const usage: UsageSnapshot = normalizeUsageInputTokens<UsageSnapshot>(mergeUsageSnapshots(extractUsageFromBillingHeaders(responseHeaders), bodyUsage), {
-        path: usagePath,
-        usageHint: bodyUsage
-      }) ?? {};
+      // As in record(), each source is normalized under its own convention.
+      // Raw-trace updates carry no provider protocol — it is not part of the
+      // gateway's raw-trace sync contract — so the billing headers fall back to
+      // the request path, which is only a proxy for the upstream's convention.
+      const usage: UsageSnapshot = mergeUsageSnapshots(
+        normalizeUsageInputTokens(extractUsageFromBillingHeaders(responseHeaders), {
+          path: usagePath,
+          source: "providerBilling"
+        }),
+        normalizeUsageInputTokens<UsageSnapshot>(bodyUsage, {
+          path: usagePath,
+          source: "responseBody"
+        })
+      ) ?? {};
       if (hasUsageNumbers(usage)) {
         const inputTokens = normalizeCount(usage.inputTokens);
         const outputTokens = normalizeCount(usage.outputTokens);

packages/core/src/usage/billing-sync.ts

@@ -166,8 +166,11 @@ export class GatewayBillingSynchronizer {
       outputTokens: numberValue(usage.output_tokens),
       totalTokens: numberValue(usage.total_tokens)
     }, {
+      // The gateway's own billing event, so these are the upstream provider's
+      // counters in the upstream's convention.
       path,
-      providerProtocol
+      providerProtocol,
+      source: "providerBilling"
     });
     const reportedCost = finiteNumber(cost.total);
     const input: UsageEventInput = {

packages/core/src/usage/normalization.ts

@@ -7,13 +7,25 @@ export type UsageTokenAccounting = {
   inputTokens?: number;
 };
 
+/**
+ * Which artefact the counts were read from. The two disagree whenever the
+ * gateway translates between formats — an Anthropic response served from an
+ * OpenAI upstream carries billing headers in the upstream's cache-inclusive
+ * convention next to a body in Anthropic's cache-exclusive one — so each has
+ * to be normalized under its own rule before the two are merged.
+ */
+export type UsageConventionSource = "providerBilling" | "responseBody";
+
+export type UsageNormalizationOptions = {
+  path?: string;
+  providerProtocol?: GatewayProviderProtocol;
+  source?: UsageConventionSource;
+  usageHint?: UsageTokenAccounting;
+};
+
 export function normalizeUsageInputTokens<T extends UsageTokenAccounting>(
   usage: T | undefined,
-  options: {
-    path?: string;
-    providerProtocol?: GatewayProviderProtocol;
-    usageHint?: UsageTokenAccounting;
-  } = {}
+  options: UsageNormalizationOptions = {}
 ): T | undefined {
   if (!usage) {
     return undefined;
@@ -37,12 +49,26 @@ export function normalizeUsageInputTokens<T extends UsageTokenAccounting>(
 
 function inputIncludesCacheTokens(
   usage: UsageTokenAccounting,
-  options: {
-    path?: string;
-    providerProtocol?: GatewayProviderProtocol;
-    usageHint?: UsageTokenAccounting;
-  }
+  options: UsageNormalizationOptions
 ): boolean | undefined {
+  if (options.source === "responseBody") {
+    // A body describes its own wire format, and the field names we parsed are
+    // the direct evidence of it. The upstream protocol is deliberately not
+    // consulted: it describes what the gateway talked to, not what it emitted,
+    // and reading a translated body under the upstream's rule subtracts the
+    // cached prefix a second time — clamping input tokens to zero on any turn
+    // where the cache hit exceeds the new input, which is most of them.
+    if (usage.inputIncludesCacheTokens !== undefined) {
+      return usage.inputIncludesCacheTokens;
+    }
+    if (options.usageHint?.inputIncludesCacheTokens !== undefined) {
+      return options.usageHint.inputIncludesCacheTokens;
+    }
+    return inputIncludesCacheTokensForPath(options.path);
+  }
+
+  // Billing headers and billing events restate the upstream provider's own
+  // counters verbatim, so the upstream protocol governs them.
   const protocolValue = inputIncludesCacheTokensForProtocol(options.providerProtocol);
   if (protocolValue !== undefined) {
     return protocolValue;

packages/core/src/usage/store.ts

@@ -220,11 +220,20 @@ export class UsageStore {
   async recordCapture(input: UsageCaptureInput): Promise<void> {
     const headersUsage = extractUsageFromBillingHeaders(input.responseHeaders);
     const bodyUsage = extractUsageFromBody(input.bodyText);
-    const usage = normalizeUsageInputTokens(mergeUsageSnapshots(headersUsage, bodyUsage), {
-      path: input.path,
-      providerProtocol: input.providerProtocol,
-      usageHint: bodyUsage
-    });
+    // Normalize each source under its own convention before merging them: on a
+    // translated response the billing headers and the body state input tokens
+    // differently, and one shared rule is wrong for one of them.
+    const usage = mergeUsageSnapshots(
+      normalizeUsageInputTokens(headersUsage, {
+        path: input.path,
+        providerProtocol: input.providerProtocol,
+        source: "providerBilling"
+      }),
+      normalizeUsageInputTokens(bodyUsage, {
+        path: input.path,
+        source: "responseBody"
+      })
+    );
     const fallbackAttribution = resolveUsageModelAttribution(input.config, input.fallbackModel);
     const responseAttribution = resolveUsageResponseModelAttribution(input.config, bodyUsage?.model);
     const route = splitRouteSelector(input.fallbackModel);

packages/core/test/unit/usage/usage-normalization.test.mjs

@@ -38,6 +38,96 @@ test("normalizeUsageInputTokens keeps Anthropic input tokens unchanged", () => {
   });
 });
 
+test("normalizeUsageInputTokens ignores the upstream protocol for a translated body", () => {
+  // Anthropic response served from an OpenAI upstream: the body already excludes
+  // the cached prefix, so the upstream's cache-inclusive rule must not apply.
+  const usage = normalizeUsageInputTokens(
+    {
+      cacheReadTokens: 3584,
+      inputIncludesCacheTokens: false,
+      inputTokens: 250,
+      outputTokens: 40
+    },
+    {
+      path: "/v1/messages",
+      providerProtocol: "openai_chat_completions",
+      source: "responseBody"
+    }
+  );
+
+  assert.equal(usage?.inputTokens, 250);
+});
+
+test("normalizeUsageInputTokens does not clamp a translated body to zero", () => {
+  // The failure this guards against: subtracting an already-excluded cached
+  // prefix drives input tokens negative, and the clamp reports it as 0.
+  const usage = normalizeUsageInputTokens(
+    {
+      cacheReadTokens: 260608,
+      inputIncludesCacheTokens: false,
+      inputTokens: 1888
+    },
+    {
+      path: "/v1/messages",
+      providerProtocol: "openai_chat_completions",
+      source: "responseBody"
+    }
+  );
+
+  assert.equal(usage?.inputTokens, 1888);
+});
+
+test("normalizeUsageInputTokens subtracts on billing headers from the same response", () => {
+  // Same response, other source: the gateway restates the upstream's own
+  // counters, which do include the cached prefix.
+  const usage = normalizeUsageInputTokens(
+    {
+      cacheReadTokens: 3584,
+      inputTokens: 3834,
+      outputTokens: 40
+    },
+    {
+      path: "/v1/messages",
+      providerProtocol: "openai_chat_completions",
+      source: "providerBilling"
+    }
+  );
+
+  assert.equal(usage?.inputTokens, 250);
+});
+
+test("normalizeUsageInputTokens still subtracts for a same-format body", () => {
+  // No translation: an OpenAI body from an OpenAI upstream is cache-inclusive,
+  // and the body's own fields say so.
+  const usage = normalizeUsageInputTokens(
+    {
+      cacheReadTokens: 20,
+      inputIncludesCacheTokens: true,
+      inputTokens: 100
+    },
+    { path: "/v1/chat/completions", source: "responseBody" }
+  );
+
+  assert.equal(usage?.inputTokens, 80);
+});
+
+test("normalizeUsageInputTokens uses the path when a body declares no convention", () => {
+  assert.equal(
+    normalizeUsageInputTokens(
+      { cacheReadTokens: 8, inputTokens: 50 },
+      { path: "/v1/messages", source: "responseBody" }
+    )?.inputTokens,
+    50
+  );
+  assert.equal(
+    normalizeUsageInputTokens(
+      { cacheReadTokens: 8, inputTokens: 50 },
+      { path: "/v1/chat/completions", source: "responseBody" }
+    )?.inputTokens,
+    42
+  );
+});
+
 test("normalizeUsageInputTokens falls back to path and usage hints", () => {
   assert.equal(
     normalizeUsageInputTokens(