fix(claude): report confirmed input usage in message_start when upstream has already sent it

src/claude/outbound.ts

@@ -189,7 +189,7 @@ function webSearchPairFromItem(item: Rec): { id: string; input: Rec; resultConte
   return { id, input, resultContent, completed };
 }
 
-function messageSnapshot(model: string): Rec {
+function messageSnapshot(model: string, confirmedUsage?: Rec): Rec {
   return {
     id: `msg_${uuid()}`,
     type: "message",
@@ -198,7 +198,7 @@ function messageSnapshot(model: string): Rec {
     model,
     stop_reason: null,
     stop_sequence: null,
-    usage: { input_tokens: 0, output_tokens: 0 },
+    usage: confirmedUsage ?? { input_tokens: 0, output_tokens: 0 },
   };
 }
 
@@ -246,6 +246,7 @@ export function responsesSseToAnthropicSse(
   let open: OpenBlock | null = null;
   let sawToolUse = false;
   let webSearchRequests = 0;
+  let earlyAnthropicUsage: Rec | undefined;
   let pingTimer: ReturnType<typeof setInterval> | undefined;
   let reader: ReadableStreamDefaultReader<Uint8Array> | undefined;
   const utf8SliceBytes = (value: string, start: number, end: number): number => {
@@ -282,7 +283,7 @@ export function responsesSseToAnthropicSse(
       const ensureStarted = () => {
         if (started) return;
         started = true;
-        emit("message_start", { type: "message_start", message: messageSnapshot(model) });
+        emit("message_start", { type: "message_start", message: messageSnapshot(model, earlyAnthropicUsage) });
         emit("ping", { type: "ping" });
       };
       // Keepalive pings protect remote deployments behind LB/NAT idle timeouts even
@@ -405,8 +406,17 @@ export function responsesSseToAnthropicSse(
       const handleFrame = (eventName: string, data: Rec) => {
         switch (eventName) {
           case "response.created":
-            // Transport prelude only. Start Anthropic framing on semantic output or completion.
+          case "response.in_progress": {
+            // Lifecycle preludes do not start Anthropic framing, but some upstreams attach
+            // confirmed input usage before semantic output. Retain only its bounded Anthropic
+            // projection so message_start can report measurements that already arrived.
+            const response = isRec(data.response) ? data.response : {};
+            const usage = isRec(response.usage) ? response.usage : undefined;
+            if (!started && usage && typeof usage.input_tokens === "number") {
+              earlyAnthropicUsage = anthropicUsage(usage);
+            }
             break;
+          }
           case "response.heartbeat":
             if ((controller.desiredSize ?? 0) > 0) emit("ping", { type: "ping" });
             break;

structure/clients/claude-desktop.md

@@ -10,6 +10,7 @@ Codex-native model discovery follows the [shared retirement policy](../catalog.m
 That projection does not migrate existing user-selected Desktop configuration or usage history.
 
 Shared parsing and streaming follow the [request-copy](../transports/byte-accounting.md#request-copy-accounting) and [stream-buffer accounting](../transports/byte-accounting.md#stream-buffer-accounting) contracts. Response-attached WebSocket telemetry follows the [stage record identity contract](../transports/responses.md#passthrough-sse-stream-shapes-314).
+Translated Anthropic first-frame usage follows the [runtime snapshot contract](../runtime.md#anthropic-streaming-usage-snapshots); Desktop profile state and usage-ledger ownership are unchanged.
 
 Claude-only connections keep their existing non-failing readiness policy; displayed catalog reasons follow the [terminal rendering contract](../runtime.md#cli-readiness-diagnostics) whether they surface at connect time or on a later refresh.
 

structure/runtime.md

@@ -16,6 +16,15 @@ it requires no runtime lifecycle change or new configuration option.
 
 Shared parsing and streaming follow the [request-copy](transports/byte-accounting.md#request-copy-accounting) and [stream-buffer accounting](transports/byte-accounting.md#stream-buffer-accounting) contracts. Response-attached WebSocket telemetry follows the [stage record identity contract](transports/responses.md#passthrough-sse-stream-shapes-314).
 
+## Anthropic streaming usage snapshots
+
+`src/claude/outbound.ts` starts Anthropic semantic framing lazily. When `response.created` or
+`response.in_progress` reports numeric input usage before the first content event, `message_start`
+uses that confirmed value through the normal Anthropic cache-token transform. Without an early
+measurement it emits the required zero snapshot without estimating or delaying content. The terminal
+`message_delta.usage` remains cumulative and is always derived from the terminal response usage;
+this wire projection does not change the usage ledger.
+
 ## CLI readiness diagnostics
 
 Catalog-derived reasoning-level diagnostics are escaped only at the human-output boundary, which `src/cli/runtime-api.ts` owns alongside the human/JSON print split. Every CLI path that prints a hub-supplied catalog value renders it there: the first-time refusal in `src/cli/connect.ts` and the connected `ocx sync` refusal in `src/cli/dispatch.ts`. C0/C1 controls, DEL, and Unicode line/paragraph separators print as visible hexadecimal escapes; structured status retains the exact reason, and a rendered failure keeps the domain error as its `cause`. The ready/unverified/incompatible classification and exit policy are unchanged.

tests/claude-integration/claude-outbound.test.ts

@@ -212,6 +212,66 @@ describe("claude outbound SSE", () => {
     expect(events.find(e => e.name === "content_block_start")?.data.content_block).toMatchObject({ type: "tool_use", name: "Bash", input: {} });
   });
 
+  test("message_start uses confirmed pre-content usage without changing cumulative terminal usage", async () => {
+    const earlyUsage = {
+      input_tokens: 120,
+      output_tokens: 0,
+      input_tokens_details: { cached_tokens: 100, cache_write_tokens: 5 },
+    };
+    const terminalUsage = { ...earlyUsage, output_tokens: 30 };
+    const upstream = [
+      sse("response.created", { response: { id: "resp_early_usage", status: "in_progress", usage: null } }),
+      sse("response.in_progress", { response: { id: "resp_early_usage", status: "in_progress", usage: earlyUsage } }),
+      sse("response.output_text.delta", { delta: "ready" }),
+      sse("response.completed", { response: { status: "completed", usage: terminalUsage } }),
+    ].join("");
+
+    const events = await collectEvents(responsesSseToAnthropicSse(streamFrom(upstream), "claude-ocx-test"));
+    expect(events.find(event => event.name === "message_start")!.data.message.usage).toEqual({
+      input_tokens: 15,
+      output_tokens: 0,
+      cache_read_input_tokens: 100,
+      cache_creation_input_tokens: 5,
+    });
+    expect(events.find(event => event.name === "message_delta")!.data.usage).toEqual({
+      input_tokens: 15,
+      output_tokens: 30,
+      cache_read_input_tokens: 100,
+      cache_creation_input_tokens: 5,
+    });
+  });
+
+  test("message_start documents unknown pre-content usage as zero while terminal usage stays authoritative", async () => {
+    const upstream = [
+      sse("response.created", { response: { id: "resp_terminal_usage", status: "in_progress", usage: null } }),
+      sse("response.output_text.delta", { delta: "ready" }),
+      sse("response.completed", {
+        response: {
+          status: "completed",
+          usage: {
+            input_tokens: 120,
+            output_tokens: 30,
+            input_tokens_details: { cached_tokens: 100, cache_write_tokens: 5 },
+          },
+        },
+      }),
+    ].join("");
+
+    const events = await collectEvents(responsesSseToAnthropicSse(streamFrom(upstream), "claude-ocx-test"));
+    // Zero is the documented honest placeholder when no input measurement has arrived. Do not
+    // replace it with an estimate or delay the first content frame to await terminal usage.
+    expect(events.find(event => event.name === "message_start")!.data.message.usage).toEqual({
+      input_tokens: 0,
+      output_tokens: 0,
+    });
+    expect(events.find(event => event.name === "message_delta")!.data.usage).toEqual({
+      input_tokens: 15,
+      output_tokens: 30,
+      cache_read_input_tokens: 100,
+      cache_creation_input_tokens: 5,
+    });
+  });
+
   test("text + thinking + tool call + completed w/ usage -> exact Anthropic sequence", async () => {
     const upstream = [
       sse("response.created", { response: { id: "resp_1", status: "in_progress" } }),