diff --git a/src/adapters/anthropic.ts b/src/adapters/anthropic.ts index a9a8279198..cc3dfebf3f 100644 --- a/src/adapters/anthropic.ts +++ b/src/adapters/anthropic.ts @@ -566,16 +566,28 @@ function defaultReasoningEffort(provider: OcxProviderConfig, modelId: string): s return trimmed; } -function usageFromAnthropic(usage: Record | undefined): OcxUsage | undefined { - if (!usage) return undefined; +function usageFromAnthropic(usage: unknown): OcxUsage | undefined { + if (!isAnthropicRecord(usage)) return undefined; + const tokens = (key: string): number | undefined => { + const value = usage[key]; + if (value === undefined) return 0; + return typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : undefined; + }; + const input = tokens("input_tokens"); + const output = tokens("output_tokens"); + const read = tokens("cache_read_input_tokens"); + const write = tokens("cache_creation_input_tokens"); + // Invalid upstream usage is unreported, not a measured zero or a string that + // can pass through aggregation into a human-readable usage report. + if (input === undefined || output === undefined || read === undefined || write === undefined) return undefined; const hasCache = usage.cache_read_input_tokens !== undefined || usage.cache_creation_input_tokens !== undefined; - const read = usage.cache_read_input_tokens ?? 0; - const write = usage.cache_creation_input_tokens ?? 0; // Anthropic reports input_tokens EXCLUSIVE of cache read/write; normalize to the // canonical inclusive convention (types.ts OcxUsage / devlog 070). + const inputTokens = input + read + write; + if (!Number.isFinite(inputTokens)) return undefined; return { - inputTokens: (usage.input_tokens ?? 0) + read + write, - outputTokens: usage.output_tokens ?? 0, + inputTokens, + outputTokens: output, ...(hasCache ? { cachedInputTokens: read, cacheReadInputTokens: read, @@ -584,15 +596,18 @@ function usageFromAnthropic(usage: Record | undefined): OcxUsage }; } -function mergeAnthropicUsage( - base: Record | undefined, - next: Record | undefined, -): Record | undefined { - if (!next) return base; - if (!base) return { ...next }; +type PendingAnthropicUsage = Record | null | undefined; + +function mergeAnthropicUsage(base: PendingAnthropicUsage, next: unknown): PendingAnthropicUsage { + // null remembers an invalid observation. A later partial cumulative frame + // cannot re-establish the missing totals, while an absent update changes nothing. + if (base === null) return null; + if (next === undefined) return base; + if (!isAnthropicRecord(next)) return null; // Anthropic `message_delta.usage` values are CUMULATIVE; adding them to the // message_start snapshot double-counted output tokens. Later frames win per key. - return { ...base, ...next }; + const merged = { ...base, ...next }; + return usageFromAnthropic(merged) === undefined ? null : merged; } function buildToolNameTransforms(provider: OcxProviderConfig): { toWire: (name: string) => string; fromWire: (name: string) => string } { @@ -1059,7 +1074,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti let currentToolCallId = ""; let currentToolCallName = ""; let currentToolCallJson = ""; - let pendingUsage: Record | undefined; + let pendingUsage: PendingAnthropicUsage; let pendingStopReason: string | undefined; let emittedDone = false; let sawVisibleText = false; @@ -1113,7 +1128,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti switch (record.event || data.type) { case "message_start": { - const message = data.message as { usage?: Record } | undefined; + const message = data.message as { usage?: unknown } | undefined; pendingUsage = mergeAnthropicUsage(pendingUsage, message?.usage); break; } @@ -1202,7 +1217,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti break; } case "message_delta": { - const usage = data.usage as Record | undefined; + const usage = data.usage; pendingUsage = mergeAnthropicUsage(pendingUsage, usage); const delta = data.delta as { stop_reason?: unknown } | undefined; if (typeof delta?.stop_reason === "string") pendingStopReason = delta.stop_reason; diff --git a/src/cli/usage-report.ts b/src/cli/usage-report.ts index 6b684277c0..3311a781a1 100644 --- a/src/cli/usage-report.ts +++ b/src/cli/usage-report.ts @@ -59,8 +59,11 @@ interface UsageReportInput { const MAX_MODEL_ROWS = 10; -function terminalText(value: string): string { - return value.replace(/[\x00-\x1f\x7f-\x9f]/g, character => { +function terminalText(value: unknown): string { + const text = typeof value === "string" ? value + : value === null || value === undefined ? "" + : typeof value === "number" || typeof value === "boolean" ? String(value) : "[invalid]"; + return text.replace(/[\x00-\x1f\x7f-\x9f\u2028\u2029]/g, character => { const code = character.charCodeAt(0); return code <= 0x7f ? `\\x${code.toString(16).padStart(2, "0")}` @@ -69,7 +72,8 @@ function terminalText(value: string): string { } function count(value: number | undefined): string { - return (value ?? 0).toLocaleString("en-US"); + if (value === undefined || value === null) return "0"; + return typeof value === "number" && Number.isFinite(value) ? value.toLocaleString("en-US") : "—"; } /** @@ -111,7 +115,7 @@ export function formatUsageReport(data: UsageReportInput): string[] { .filter(Boolean).join(" and "); lines.push(`No usage recorded for ${terminalText(what)} in this range.`); lines.push("Check the spelling against `ocx usage --json`, or widen --range."); - return lines; + return lines.map(terminalText); } const tokenSplit = [ @@ -185,5 +189,5 @@ export function formatUsageReport(data: UsageReportInput): string[] { lines.push(""); lines.push("Not a billing receipt. Subscription usage or provider credits may apply instead."); - return lines; + return lines.map(terminalText); } diff --git a/tests/adapters/anthropic/anthropic-error-stop-reason.test.ts b/tests/adapters/anthropic/anthropic-error-stop-reason.test.ts index b30cb9641d..52a0a8309b 100644 --- a/tests/adapters/anthropic/anthropic-error-stop-reason.test.ts +++ b/tests/adapters/anthropic/anthropic-error-stop-reason.test.ts @@ -13,6 +13,70 @@ const provider: OcxProviderConfig = { apiKey: "test-key", }; +describe("Anthropic usage numeric boundary", () => { + test("preserves empty usage and absent usage as distinct states", async () => { + for (const usage of [{}, undefined]) { + const events = await createAnthropicAdapter(provider).parseResponse!(Response.json({ + content: [{ type: "text", text: "ok" }], stop_reason: "end_turn", ...(usage ? { usage } : {}), + })) as AdapterEvent[]; + const done = events.find(event => event.type === "done"); + expect(done && "usage" in done ? done.usage : undefined) + .toEqual(usage ? { inputTokens: 0, outputTokens: 0 } : undefined); + } + }); + + test("preserves inclusive cache input and cumulative streaming output", async () => { + const frames = [ + { type: "message_start", message: { usage: { input_tokens: 10, cache_read_input_tokens: 3, cache_creation_input_tokens: 2 } } }, + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 4 } }, + { type: "message_stop" }, + ].map(frame => `event: ${frame.type}\ndata: ${JSON.stringify(frame)}\n\n`).join(""); + const events: AdapterEvent[] = []; + for await (const event of createAnthropicAdapter(provider).parseStream(new Response(frames))) events.push(event); + const done = events.find(event => event.type === "done"); + expect(done && "usage" in done ? done.usage : undefined).toEqual({ + inputTokens: 15, outputTokens: 4, cachedInputTokens: 3, cacheReadInputTokens: 3, cacheCreationInputTokens: 2, + }); + }); + + test.each(["input_tokens", "output_tokens", "cache_read_input_tokens", "cache_creation_input_tokens"])( + "does not emit malformed %s as reported usage", async key => { + for (const invalid of ["\x1b[2J", "42", null, -1, true, {}, []]) { + const response = Response.json({ content: [{ type: "text", text: "ok" }], stop_reason: "end_turn", + usage: { input_tokens: 10, output_tokens: 4, [key]: invalid } }); + const events = await createAnthropicAdapter(provider).parseResponse!(response) as AdapterEvent[]; + const done = events.find(event => event.type === "done"); + expect(done).toBeDefined(); + expect(done && "usage" in done ? done.usage : undefined).toBeUndefined(); + } + }, + ); + + test("rejects an overflowing inclusive input total", async () => { + const response = Response.json({ content: [{ type: "text", text: "ok" }], stop_reason: "end_turn", + usage: { input_tokens: Number.MAX_VALUE, cache_read_input_tokens: Number.MAX_VALUE, output_tokens: 4 } }); + const events = await createAnthropicAdapter(provider).parseResponse!(response) as AdapterEvent[]; + const done = events.find(event => event.type === "done"); + expect(done && "usage" in done ? done.usage : undefined).toBeUndefined(); + }); + + test("streaming rejects a malformed cumulative update without changing content", async () => { + const frames = [ + { type: "message_start", message: { usage: { input_tokens: 10 } } }, + { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "ok" } }, + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: "\x1b[2J" } }, + { type: "message_stop" }, + ].map(frame => `event: ${frame.type}\ndata: ${JSON.stringify(frame)}\n\n`).join(""); + const events: AdapterEvent[] = []; + for await (const event of createAnthropicAdapter(provider).parseStream(new Response(frames))) events.push(event); + const done = events.find(event => event.type === "done"); + expect(done).toBeDefined(); + expect(done && "usage" in done ? done.usage : undefined).toBeUndefined(); + expect(JSON.stringify(buildResponseJSON(events, "anthropic/claude-test"))).toContain("ok"); + }); +}); + /** * These drive the REAL adapter parsers. An earlier version of this suite constructed the * downstream error event by hand, so it stayed green while the adapter itself still emitted a diff --git a/tests/cli/cli-usage-report.test.ts b/tests/cli/cli-usage-report.test.ts index 3342aab2ea..5399137a57 100644 --- a/tests/cli/cli-usage-report.test.ts +++ b/tests/cli/cli-usage-report.test.ts @@ -52,6 +52,38 @@ async function run(argv: string[], body: unknown): Promise<{ code: number; out: } describe("formatUsageReport", () => { + test("keeps malformed token counts and every human line inert", () => { + const control = "before\x1b[2J\x07\u2028after\u2029"; + const body = payload({ + range: control, + summary: { requests: 1, totalTokens: control, inputTokens: Infinity, outputTokens: control }, + providers: [{ provider: null, requests: 1, totalTokens: control }], + accounts: [{ accountLogLabel: control, requests: 1, totalTokens: control }], + }); + const lines = formatUsageReport(body as never); + expect(lines.every(line => !/[\x00-\x1f\x7f-\x9f\u2028\u2029]/.test(line))).toBe(true); + expect(lines.join("\n")).toContain("before\\x1b[2J\\x07\\u2028after\\u2029"); + expect(lines.find(line => line.startsWith("Tokens"))).toBe("Tokens — (in — / out —)"); + expect(body.summary).toEqual({ requests: 1, totalTokens: control, inputTokens: Infinity, outputTokens: control }); + }); + + test("escapes Unicode line separators on the no-match return too", () => { + const lines = formatUsageReport(payload({ + filter: { provider: "before\u2028after", model: null, matched: false, comboOverlap: false }, + }) as never); + expect(lines.every(line => !/[\x00-\x1f\x7f-\x9f\u2028\u2029]/.test(line))).toBe(true); + expect(lines.join("\n")).toContain("before\\u2028after"); + }); + + test("preserves ordinary per-account totals and keeps JSON unchanged", async () => { + const body = payload({ accounts: [{ accountLogLabel: "account-1", requests: 12, totalTokens: 345, estimatedCostUsd: 0.125 }] }); + expect(formatUsageReport(body as never).join("\n")).toMatch(/account-1\s+12\s+345\s+~\$0\.1250/); + const malformed = payload({ summary: { requests: 1, outputTokens: "\x1b[2J" } }); + const { code, out } = await run(["usage", "--json"], malformed); + expect(code).toBe(0); + expect(JSON.parse(out)).toEqual(malformed); + }); + test("prints per-provider and per-model cost, not an item count", () => { const out = formatUsageReport(payload() as never).join("\n"); expect(out).toContain("~$12.3456"); diff --git a/tests/usage/usage-aggregate-cache.test.ts b/tests/usage/usage-aggregate-cache.test.ts index 3efb5615e4..cdaea2c23b 100644 --- a/tests/usage/usage-aggregate-cache.test.ts +++ b/tests/usage/usage-aggregate-cache.test.ts @@ -23,6 +23,12 @@ import { resetUsageReadCacheForTests, type PersistedUsageEntry } from "../../src import * as usageLedgerScannerModule from "../../src/usage/ledger-scanner"; import { refreshUserCostOverlays } from "../../src/usage/user-cost-overlays"; import { buildRouteDecisionTrace } from "../../src/routing/trace"; +import { createAnthropicAdapter } from "../../src/adapters/anthropic"; +import { buildResponseJSON } from "../../src/bridge"; +import { formatUsageReport } from "../../src/cli/usage-report"; +import { addFinalRequestLog, clearRequestLogsForTests, type RequestLogContext } from "../../src/server/request-log"; +import type { AdapterEvent } from "../../src/types"; +import { withTestTranslatorBudget } from "../helpers/translator-budget"; const NOW = Date.parse("2026-09-01T10:00:00.000Z"); @@ -62,6 +68,7 @@ beforeEach(() => { }); afterEach(() => { + clearRequestLogsForTests(); resetUsageAggregateCacheForTests(); resetUsageReadCacheForTests(); resetAppOwnedMemoryForTests(); @@ -72,6 +79,56 @@ afterEach(() => { }); describe("retained usage aggregate cache", () => { + test.each(["message_start", "message_delta"].flatMap(phase => + ["bad", [], null, false, 7, { output_tokens: "bad" }].map(usage => ({ phase, usage })), + ))("malformed streamed usage at $phase stays unreported after a valid update: $usage", async ({ phase, usage }) => { + const adapter = withTestTranslatorBudget(createAnthropicAdapter({ + adapter: "anthropic", baseUrl: "https://api.anthropic.com", apiKey: "test-key", + })); + const frames = [ + { type: "message_start", message: { usage: phase === "message_start" ? usage : { input_tokens: 10 } } }, + { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "ok" } }, + ...(phase === "message_delta" ? [{ type: "message_delta", delta: {}, usage }] : []), + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 4 } }, + { type: "message_stop" }, + ].map(frame => `event: ${frame.type}\ndata: ${JSON.stringify(frame)}\n\n`).join(""); + const events: AdapterEvent[] = []; + for await (const event of adapter.parseStream(new Response(frames))) events.push(event); + const logCtx: RequestLogContext = { provider: "anthropic", model: "claude-test" }; + const result = buildResponseJSON(events, "anthropic/claude-test", { onUsage: observed => { logCtx.usage = observed; } }); + expect(result.status).toBe("completed"); + expect(JSON.stringify(result.output)).toContain("ok"); + addFinalRequestLog("malformed-stream-usage", Date.now(), logCtx, 200, { closeReason: "non_stream" }); + const persisted = JSON.parse(readFileSync(join(testDir, "usage.jsonl"), "utf8").trim()); + expect(persisted.usageStatus).toBe("unreported"); + expect(persisted.usage).toBeUndefined(); + const report = (await getUsageAggregate()).accumulator.summarize("all", Date.now()); + expect(report.summary.requests).toBe(1); + expect(report.summary.unmeteredRequests).toBe(1); + }); + + test("malformed Anthropic usage stays unmetered through the real ledger and human report", async () => { + const adapter = withTestTranslatorBudget(createAnthropicAdapter({ + adapter: "anthropic", baseUrl: "https://api.anthropic.com", apiKey: "test-key", + })); + const response = Response.json({ + content: [{ type: "text", text: "ok" }], stop_reason: "end_turn", + usage: { input_tokens: 10, output_tokens: "\x1b[2J" }, + }); + const events = await adapter.parseResponse!(response) as AdapterEvent[]; + const logCtx: RequestLogContext = { provider: "anthropic", model: "claude-test" }; + buildResponseJSON(events, "anthropic/claude-test", { onUsage: usage => { logCtx.usage = usage; } }); + addFinalRequestLog("malformed-usage", Date.now(), logCtx, 200, { closeReason: "non_stream" }); + const persisted = JSON.parse(readFileSync(join(testDir, "usage.jsonl"), "utf8").trim()); + const report = (await getUsageAggregate()).accumulator.summarize("all", Date.now()); + expect(report.summary.requests).toBe(1); + expect(formatUsageReport(report).every(line => !/[\x00-\x1f\x7f-\x9f]/.test(line))).toBe(true); + expect(persisted.usageStatus).toBe("unreported"); + expect(persisted.usage).toBeUndefined(); + expect(report.summary.unmeteredRequests).toBe(1); + }); + test("custom cache keys isolate both endpoints and never poison preset aggregates", async () => { const path = join(testDir, "usage.jsonl"); const rows = [NOW - 2_000, NOW - 1_000, NOW].map((timestamp, index) => ({ ...entry(String(index)), timestamp }));