fix(gateway): meter OpenAI usage from the shapes the wire actually sends

services/gateway/src/usage.ts

@@ -28,10 +28,23 @@ export function usageFromJson(provider: Provider, body: unknown): Partial<Usage>
       cacheWriteTokens: numberValue(row.cache_creation_input_tokens),
     };
   }
+  // OpenAI's two APIs name usage differently: Responses reports
+  // input_tokens/output_tokens (details under input_tokens_details), Chat
+  // Completions prompt_tokens/completion_tokens (prompt_tokens_details).
+  // Both report input INCLUSIVE of cached tokens, while costCents sums the
+  // buckets additively in the Anthropic convention - so cached tokens are
+  // subtracted from input here, or they would be billed twice.
+  const inputInclusive = numberValue(row.input_tokens) ?? numberValue(row.prompt_tokens);
+  const cacheRead =
+    nestedNumber(row.input_tokens_details, "cached_tokens") ??
+    nestedNumber(row.prompt_tokens_details, "cached_tokens");
   return {
-    inputTokens: numberValue(row.input_tokens),
-    outputTokens: numberValue(row.output_tokens),
-    cacheReadTokens: nestedNumber(row.input_tokens_details, "cached_tokens"),
+    inputTokens:
+      inputInclusive !== undefined && cacheRead !== undefined
+        ? Math.max(0, inputInclusive - cacheRead)
+        : inputInclusive,
+    outputTokens: numberValue(row.output_tokens) ?? numberValue(row.completion_tokens),
+    cacheReadTokens: cacheRead,
     cacheWriteTokens: nestedNumber(row.input_tokens_details, "cache_creation_tokens"),
   };
 }
@@ -125,7 +138,14 @@ export class UsageTee extends Transform {
           mergeUsage(this.usage, usageFromJson("anthropic", source));
         }
       } else {
-        mergeUsage(this.usage, usageFromJson("openai", parsed));
+        // OpenAI streamed shapes: Chat Completions puts usage on the final
+        // chunk's top level (requested via stream_options.include_usage),
+        // while the Responses API - the wire Codex speaks - nests it inside
+        // the response.completed envelope ({"type":"response.completed",
+        // "response":{"usage":{...}}}) and never at the frame's top level.
+        for (const source of [parsed, (parsed as { response?: unknown })?.response]) {
+          mergeUsage(this.usage, usageFromJson("openai", source));
+        }
       }
     } catch {
       // Provider bytes are still passed through; malformed telemetry frames only affect metering.

services/gateway/test/gateway.test.ts

@@ -502,7 +502,13 @@ describe("gateway", async () => {
     const row = (
       await db.select().from(llmRequests).where(eq(llmRequests.virtualKeyId, setup.keyId))
     )[0];
-    expect(row?.costCents).toBe(225);
+    // Chat reports prompt/completion names, with prompt_tokens INCLUSIVE of
+    // cached: 1M prompt (400k cached) + 1M completion on gpt-5.5-mini is
+    // 0.6*0.25 + 1*2 + 0.4*0.025 = $2.16 - not input-rate on the full million.
+    expect(row?.inputTokens).toBe(600_000);
+    expect(row?.cacheRead).toBe(400_000);
+    expect(row?.outputTokens).toBe(1_000_000);
+    expect(row?.costCents).toBe(216);
   });
 
   it("3b. OpenAI Responses streaming preserves the native request shape", async () => {
@@ -513,7 +519,7 @@ describe("gateway", async () => {
       input: "hello",
     });
     expect(response.status).toBe(200);
-    await response.json();
+    expect(await response.text()).toContain("response.completed");
     await waitForRequestCount(1);
     expect(stubState.lastOpenAiRequest).toMatchObject({
       model: "gpt-5.6-sol",
@@ -523,6 +529,17 @@ describe("gateway", async () => {
     expect(
       (stubState.lastOpenAiRequest as { stream_options?: unknown }).stream_options,
     ).toBeUndefined();
+    // The Codex path: usage lives inside the response.completed envelope.
+    // 1M input (250k cached) + 200k output on gpt-5.6-sol:
+    // 0.75*5 + 0.2*30 + 0.25*0.5 = $9.875. Metering this at zero is #232's
+    // headline - hard budgets never accumulate for Codex runs.
+    const row = (
+      await db.select().from(llmRequests).where(eq(llmRequests.virtualKeyId, setup.keyId))
+    )[0];
+    expect(row?.inputTokens).toBe(750_000);
+    expect(row?.cacheRead).toBe(250_000);
+    expect(row?.outputTokens).toBe(200_000);
+    expect(row?.costCents).toBe(987.5);
   });
 
   it("4. Hard budget exceeded returns 402 and skips upstream", async () => {
@@ -1346,7 +1363,7 @@ async function buildStub(state: StubState) {
       reply.raw.end(
         [
           'data: {"choices":[{"delta":{"content":"ok"}}]}\n\n',
-          'data: {"choices":[],"usage":{"input_tokens":1000000,"output_tokens":1000000}}\n\n',
+          'data: {"choices":[],"usage":{"prompt_tokens":1000000,"completion_tokens":1000000,"prompt_tokens_details":{"cached_tokens":400000}}}\n\n',
           "data: [DONE]\n\n",
         ].join(""),
       );
@@ -1355,14 +1372,26 @@ async function buildStub(state: StubState) {
     return {
       id: "chatcmpl_stub",
       choices: [{ message: { role: "assistant", content: "ok" } }],
-      usage: { input_tokens: 1_000_000, output_tokens: 1_000_000 },
+      usage: { prompt_tokens: 1_000_000, completion_tokens: 1_000_000 },
     };
   });
 
-  app.post("/openai/v1/responses", async (request) => {
+  app.post("/openai/v1/responses", async (request, reply) => {
     state.openaiCalls += 1;
     expect(request.headers.authorization).toBe("Bearer real-openai");
     state.lastOpenAiRequest = request.body;
+    if ((request.body as { stream?: boolean }).stream) {
+      // The wire shape Codex consumes: usage arrives once, nested inside the
+      // response.completed envelope - never at the frame's top level.
+      reply.raw.writeHead(200, { "content-type": "text/event-stream" });
+      reply.raw.end(
+        [
+          'data: {"type":"response.created","response":{"id":"resp_stub"}}\n\n',
+          'data: {"type":"response.completed","response":{"id":"resp_stub","usage":{"input_tokens":1000000,"output_tokens":200000,"input_tokens_details":{"cached_tokens":250000}}}}\n\n',
+        ].join(""),
+      );
+      return reply;
+    }
     return {
       id: "resp_stub",
       output: [],

services/gateway/test/usage.test.ts

@@ -8,6 +8,20 @@ const CONTENT_DELTA =
 const MESSAGE_DELTA =
   'event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":700}}\n\n';
 
+// OpenAI speaks two dialects. Chat Completions names the buckets
+// prompt/completion and reports usage on a final chunk requested through
+// stream_options.include_usage; the Responses API - the wire Codex speaks -
+// names them input/output and nests usage inside the response.completed
+// envelope, never at the frame's top level. Both report input INCLUSIVE of
+// cached tokens, while costCents sums the buckets additively.
+const CHAT_CONTENT_DELTA = 'data: {"choices":[{"delta":{"content":"ok"}}]}\n\n';
+const CHAT_USAGE_FINAL =
+  'data: {"choices":[],"usage":{"prompt_tokens":1000,"completion_tokens":700,"prompt_tokens_details":{"cached_tokens":400}}}\n\n';
+const CHAT_DONE = "data: [DONE]\n\n";
+const RESPONSES_CREATED = 'data: {"type":"response.created","response":{"id":"resp_1"}}\n\n';
+const RESPONSES_COMPLETED =
+  'data: {"type":"response.completed","response":{"id":"resp_1","usage":{"input_tokens":1000,"output_tokens":700,"input_tokens_details":{"cached_tokens":400}}}}\n\n';
+
 describe("UsageTee anthropic streaming usage", () => {
   it("meters input and cache tokens reported by message_start", async () => {
     const tee = new UsageTee("anthropic");
@@ -66,6 +80,51 @@ describe("UsageTee anthropic streaming usage", () => {
   });
 });
 
+describe("UsageTee openai streaming usage", () => {
+  it("meters the chat completions naming and bills cached tokens once", async () => {
+    const tee = new UsageTee("openai");
+    await feed(tee, [CHAT_CONTENT_DELTA, CHAT_USAGE_FINAL, CHAT_DONE]);
+
+    // Reading only input_tokens/output_tokens meters this stream at zero, which
+    // is a hard-budget bypass: nothing accumulates against the reservation.
+    // prompt_tokens is inclusive of the 400 cached, so input is the 600 that
+    // were actually processed - leaving 1000 bills those 400 at the input rate
+    // on top of the cache-read rate.
+    expect(tee.usage).toEqual({
+      inputTokens: 600,
+      outputTokens: 700,
+      cacheReadTokens: 400,
+      cacheWriteTokens: 0,
+    });
+  });
+
+  it("meters usage nested in the response.completed envelope", async () => {
+    const tee = new UsageTee("openai");
+    await feed(tee, [RESPONSES_CREATED, RESPONSES_COMPLETED]);
+
+    // Codex streams here. Usage never appears at the frame's top level, so
+    // reading only the frame meters the whole run at zero. The earlier
+    // response.created frame carries no usage and must not erase the total.
+    expect(tee.usage).toEqual({
+      inputTokens: 600,
+      outputTokens: 700,
+      cacheReadTokens: 400,
+      cacheWriteTokens: 0,
+    });
+  });
+
+  it("still meters a non-streamed body using the chat completions naming", async () => {
+    const tee = new UsageTee("openai");
+    await feed(tee, [JSON.stringify({ usage: { prompt_tokens: 12, completion_tokens: 34 } })]);
+
+    // No cached details here: input stays as reported rather than becoming
+    // undefined, which is what a naive subtraction would produce.
+    expect(tee.usage.inputTokens).toBe(12);
+    expect(tee.usage.outputTokens).toBe(34);
+    expect(tee.usage.cacheReadTokens).toBe(0);
+  });
+});
+
 async function feed(tee: UsageTee, chunks: string[]): Promise<string> {
   const seen: Buffer[] = [];
   tee.on("data", (chunk: Buffer) => seen.push(chunk));