diff --git a/src/claude/outbound.ts b/src/claude/outbound.ts index f281b632b7..8b98a8ea95 100644 --- a/src/claude/outbound.ts +++ b/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 | undefined; let reader: ReadableStreamDefaultReader | 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; diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index 305c123854..a2025bdb16 100644 --- a/structure/clients/claude-desktop.md +++ b/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. diff --git a/structure/runtime.md b/structure/runtime.md index 17a3ccbf95..7302c5fd91 100644 --- a/structure/runtime.md +++ b/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. diff --git a/tests/claude-integration/claude-outbound.test.ts b/tests/claude-integration/claude-outbound.test.ts index 375fac9155..769cae779d 100644 --- a/tests/claude-integration/claude-outbound.test.ts +++ b/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" } }),