fix(usage): validate token values and escape human output

src/adapters/anthropic.ts

@@ -566,16 +566,28 @@ function defaultReasoningEffort(provider: OcxProviderConfig, modelId: string): s
   return trimmed;
 }
 
-function usageFromAnthropic(usage: Record<string, number> | 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<string, number> | undefined): OcxUsage
   };
 }
 
-function mergeAnthropicUsage(
-  base: Record<string, number> | undefined,
-  next: Record<string, number> | undefined,
-): Record<string, number> | undefined {
-  if (!next) return base;
-  if (!base) return { ...next };
+type PendingAnthropicUsage = Record<string, unknown> | 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<string, number> | 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<string, number> } | 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<string, number> | 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;

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);
 }

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

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");

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 }));