feat(opencode): surface turn-level cache and spend telemetry

src/admin/metrics-sink.ts

@@ -36,6 +36,8 @@ export interface TurnMetricSample {
   cacheRead?: number;
   cacheWrite?: number;
   uncachedInput?: number;
+  outputTokens?: number;
+  costUsd?: number;
 }
 
 export function cacheHitRatio(s: { cacheRead?: number; cacheWrite?: number; uncachedInput?: number }): number | null {

src/admin/postgres-metrics-sink.ts

@@ -37,6 +37,8 @@ const COLUMNS: readonly EventColumn<keyof TurnMetricSample & string>[] = [
   ["cache_read", "cacheRead", "BIGINT", "number"],
   ["cache_write", "cacheWrite", "BIGINT", "number"],
   ["uncached_input", "uncachedInput", "BIGINT", "number"],
+  ["output_tokens", "outputTokens", "BIGINT", "number"],
+  ["cost_usd", "costUsd", "DOUBLE PRECISION", "number"],
 ];
 
 const EXTRA_SCHEMA_STATEMENTS = [

src/api/routes/admin/observability.ts

@@ -150,6 +150,18 @@ export async function metrics(ctx: ApiCtx): Promise<void> {
     uncachedInputTotal,
   };
 
+  // Turn-level spend, straight off the usage the harnesses already meter:
+  // output tokens and provider-computed cost land on turn_metrics next to
+  // the cache telemetry, so an operator no longer joins session_llm_requests
+  // by hand to know what a scope or model spends.
+  const costSamples = samples.filter((s) => s.costUsd !== undefined || s.outputTokens !== undefined);
+  const spend = {
+    samples: costSamples.length,
+    turnsWithKnownCost: samples.filter((s) => s.costUsd !== undefined).length,
+    costUsdTotal: samples.reduce((n, s) => n + (s.costUsd ?? 0), 0),
+    outputTokensTotal: samples.reduce((n, s) => n + (s.outputTokens ?? 0), 0),
+  };
+
   const phaseFields: [string, (s: TurnMetricSample) => number | undefined][] = [
     ["total", (s) => s.totalMs],
     ["ttft", (s) => s.ttftMs],
@@ -225,6 +237,7 @@ export async function metrics(ctx: ApiCtx): Promise<void> {
     series,
     anatomy,
     cache,
+    spend,
     phases,
   });
 }

src/core/orchestrator.ts

@@ -2796,6 +2796,9 @@ export function createOrchestrator(deps: OrchestratorDeps): Orchestrator {
                 uncachedInput: result.cacheUsage.uncachedInput,
               }
             : {}),
+          ...(result.costUsage
+            ? { outputTokens: result.costUsage.outputTokens, costUsd: result.costUsage.costUsd }
+            : {}),
         });
         const onTurnEnd = memoryStrategy.onTurnEnd?.bind(memoryStrategy);
         if (!pausing && useMemory && memoryPolicy.capture !== "off" && onTurnEnd) {

src/harness/claude-harness.ts

@@ -764,11 +764,12 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness {
       const usageTotals = [...callUsage.values()].reduce(
         (acc, usage) => {
           acc.input += usage.input;
+          acc.output += usage.output;
           acc.cacheRead += usage.cacheRead;
           acc.cacheWrite += usage.cacheWrite;
           return acc;
         },
-        { input: 0, cacheRead: 0, cacheWrite: 0 },
+        { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
       );
       return {
         reply,
@@ -793,6 +794,12 @@ export function createClaudeHarness(opts: ClaudeHarnessOptions = {}): Harness {
                   finalResult.usage.cache_creation_input_tokens,
               ),
             },
+        // The SDK result carries no cost total, so spend is surfaced only
+        // when the per-message accumulator saw cost deltas; the fallback
+        // branch keeps output tokens but reports no known cost.
+        costUsage: callUsage.size
+          ? { outputTokens: usageTotals.output, costUsd: lastTotalCostUsd }
+          : { outputTokens: finalResult.usage.output_tokens ?? 0, costUsd: 0 },
         ...(tapeWriteFailed ? { tapeWriteFailed: true } : {}),
       };
     } finally {

src/harness/codex-harness.ts

@@ -9,6 +9,7 @@ import { DEFAULT_CODEX_MODEL_ID, modelSupportedByHarness } from "../model/pi-mod
 import { startSignalPoll, type RunSignalStore } from "../runs/run-signal-store.ts";
 import type { TaskStatus, TaskStore } from "../tasks/task-store.ts";
 import type { LlmCallUsage } from "../sessions/session-store.ts";
+import { usageToTurnTelemetry } from "./harness.ts";
 import type { ScopeId, SessionEntry } from "../types.ts";
 import { swallow } from "../util/errors.ts";
 import { countTokens } from "../util/tokens.ts";
@@ -832,13 +833,20 @@ export function createCodexHarness(opts: CodexHarnessOptions = {}): Harness {
           payload: { text: reply, stopped: state.stopped || undefined },
           scopeLabel: turn.scopeLabel,
         });
+      // Turn-level usage straight off the same live thread totals the
+      // recordLlmRequest flush reports, so codex turns land in turn_metrics
+      // with cache and spend telemetry like pi and claude. The codex SDK
+      // reports no cost, so costUsd stays 0 (the value already persisted to
+      // session_llm_requests for the same calls).
+      const telemetry = usageToTurnTelemetry(sumUsage(state.usageByThread));
       return {
         reply,
         ...(state.stopped ? { stopped: true as const } : {}),
         ...(ref.silentRequested ? { silent: true } : {}),
         ...(ref.pendingApprovals?.length ? { pendingApprovals: ref.pendingApprovals } : {}),
         ...(ref.pausedOnApproval ? { pausedOnApproval: true } : {}),
         modelCalls: state.modelCalls,
+        ...(telemetry ?? {}),
         ...(state.tapeWriteFailed ? { tapeWriteFailed: true } : {}),
       };
     } finally {

src/harness/harness.ts

@@ -109,10 +109,22 @@ export interface HarnessTurnResult {
   pausedOnApproval?: boolean;
   modelCalls?: number;
   cacheUsage?: { cacheRead: number; cacheWrite: number; uncachedInput: number };
+  costUsage?: { outputTokens: number; costUsd: number };
   compileMs?: number;
   tapeWriteFailed?: boolean;
 }
 
+export function usageToTurnTelemetry(usage: LlmCallUsage | null): {
+  cacheUsage: { cacheRead: number; cacheWrite: number; uncachedInput: number };
+  costUsage: { outputTokens: number; costUsd: number };
+} | null {
+  if (!usage) return null;
+  return {
+    cacheUsage: { cacheRead: usage.cacheRead, cacheWrite: usage.cacheWrite, uncachedInput: usage.input },
+    costUsage: { outputTokens: usage.output, costUsd: usage.costUsd },
+  };
+}
+
 export interface HarnessDetectInput {
   session: Session;
   message: string;

src/harness/mock-harness.ts

@@ -682,6 +682,10 @@ export function createMockHarness(): Harness {
           }),
           { cacheRead: 0, cacheWrite: 0, uncachedInput: 0 },
         );
+        const costUsage = steps.reduce(
+          (acc, u) => ({ outputTokens: acc.outputTokens + u.output, costUsd: acc.costUsd + u.costUsd }),
+          { outputTokens: 0, costUsd: 0 },
+        );
         const base = collected.length
           ? {
               reply,
@@ -691,7 +695,7 @@ export function createMockHarness(): Harness {
             }
           : { reply, modelCalls };
         if (muteReply) base.reply = "";
-        return { ...base, ...(silent ? { silent: true as const } : {}), cacheUsage };
+        return { ...base, ...(silent ? { silent: true as const } : {}), cacheUsage, costUsage };
       },
 
       shouldRespond(detect: HarnessDetectInput): Promise<HarnessDetectResult> {

src/harness/opencode-harness.ts

@@ -13,6 +13,7 @@ import type { CustomProviderSpec } from "../model/custom-providers.ts";
 import { DEFAULT_AGENT_MODEL_ID, resolveModel } from "../model/pi-models.ts";
 import { startSignalPoll, type RunSignalStore } from "../runs/run-signal-store.ts";
 import type { LlmCallUsage } from "../sessions/session-store.ts";
+import { usageToTurnTelemetry } from "./harness.ts";
 import type { ScopeId, SessionEntry } from "../types.ts";
 import type { TaskStore } from "../tasks/task-store.ts";
 import { errMessage, swallow } from "../util/errors.ts";
@@ -84,6 +85,7 @@ type LlmCapture = { sessionId: string; step: number; model: string; request: unk
 type ActiveTurn = {
   turn: HarnessTurnInput;
   ref: ToolContextRef;
+  turnUsage: LlmCallUsage | null;
   tools: Map<string, BridgedTool>;
   system: string;
   history: unknown[];
@@ -484,6 +486,7 @@ export function createOpenCodeHarness(opts: OpenCodeHarnessOptions = {}): Harnes
     seenText: new Map(),
     seenTasks: new Map(),
     eventTail: Promise.resolve(),
+    turnUsage: null,
   });
 
   const processEvent = async (event: unknown): Promise<void> => {
@@ -884,6 +887,7 @@ export function createOpenCodeHarness(opts: OpenCodeHarnessOptions = {}): Harnes
       eventTail: Promise.resolve(),
       stopped: false,
       child: false,
+      turnUsage: null,
     };
     active.set(sessionId, state);
     const abort = async (stopped: boolean) => {
@@ -984,6 +988,20 @@ export function createOpenCodeHarness(opts: OpenCodeHarnessOptions = {}): Harnes
         } catch (error) {
           swallow("opencode: llm request record", error);
         }
+        const usage = usageFromInfo(info);
+        if (usage) {
+          const prior = state.turnUsage;
+          state.turnUsage = prior
+            ? {
+                input: prior.input + usage.input,
+                output: prior.output + usage.output,
+                cacheRead: prior.cacheRead + usage.cacheRead,
+                cacheWrite: prior.cacheWrite + usage.cacheWrite,
+                totalTokens: prior.totalTokens + usage.totalTokens,
+                costUsd: prior.costUsd + usage.costUsd,
+              }
+            : usage;
+        }
       }
     };
     const prompt = [turn.input, turn.environment].filter((item) => item?.trim()).join("\n\n");
@@ -1072,13 +1090,20 @@ export function createOpenCodeHarness(opts: OpenCodeHarnessOptions = {}): Harnes
           payload: { text: reply, stopped: state.stopped || undefined },
           scopeLabel: turn.scopeLabel,
         });
+      // The finally block already awaits the flush before this promise
+      // resolves, so running it here changes no user-visible timing -- it
+      // only lets the turn result carry the usage totals the flush just
+      // recorded. The finally call stays: the splice makes it a no-op here
+      // and it still covers every non-happy path.
+      await flushLlmRequests();
       return {
         reply,
         ...(state.stopped ? { stopped: true as const } : {}),
         ...(ref.silentRequested ? { silent: true } : {}),
         ...(ref.pendingApprovals?.length ? { pendingApprovals: ref.pendingApprovals } : {}),
         ...(ref.pausedOnApproval ? { pausedOnApproval: true } : {}),
         modelCalls: state.captures.length,
+        ...(usageToTurnTelemetry(state.turnUsage) ?? {}),
         ...(tapeWriteFailed ? { tapeWriteFailed: true } : {}),
       };
     } finally {

src/harness/pi-harness.ts

@@ -574,6 +574,21 @@ function piUsageToCallUsage(u: PiUsageShape | undefined): LlmCallUsage | null {
   };
 }
 
+function sumCostUsage(
+  stats: ReadonlyArray<{ usage: LlmCallUsage | null }>,
+): { outputTokens: number; costUsd: number } | null {
+  let saw = false;
+  let outputTokens = 0;
+  let costUsd = 0;
+  for (const s of stats) {
+    if (!s.usage) continue;
+    saw = true;
+    outputTokens += s.usage.output;
+    costUsd += s.usage.costUsd;
+  }
+  return saw ? { outputTokens, costUsd } : null;
+}
+
 function sumCacheUsage(
   stats: ReadonlyArray<{ usage: LlmCallUsage | null }>,
 ): { cacheRead: number; cacheWrite: number; uncachedInput: number } | null {
@@ -1989,14 +2004,15 @@ export function createPiHarness(opts?: PiHarnessOptions): Harness {
             });
             await checkpointSubturn(finalEntry.seq);
             const cacheUsage = sumCacheUsage(callStats);
+            const costUsage = sumCostUsage(callStats);
             const base = {
               reply,
               stopped: true as const,
               modelCalls: entry.ref.modelCalls ?? 0,
               compileMs,
               ...(tapeWriteFailed ? { tapeWriteFailed: true } : {}),
             };
-            return cacheUsage ? { ...base, cacheUsage } : base;
+            return { ...base, ...(cacheUsage ? { cacheUsage } : {}), ...(costUsage ? { costUsage } : {}) };
           }
           const closingText = recoveryDead ? "" : (piLastAssistantTextOrThrow(entry.agentSession) ?? "");
           const closingTextWithWaiver = [closingText, grindWaiverNote].filter(Boolean).join("\n\n");
@@ -2011,6 +2027,7 @@ export function createPiHarness(opts?: PiHarnessOptions): Harness {
           const pendingApprovals = entry.ref.pendingApprovals ?? [];
           const modelCalls = entry.ref.modelCalls ?? 0;
           const cacheUsage = sumCacheUsage(callStats);
+          const costUsage = sumCostUsage(callStats);
           const silent = entry.ref.silentRequested ? { silent: true as const } : {};
           const base = pendingApprovals.length
             ? {
@@ -2023,7 +2040,7 @@ export function createPiHarness(opts?: PiHarnessOptions): Harness {
                 compileMs,
               }
             : { reply, ...(tapeWriteFailed ? { tapeWriteFailed: true } : {}), modelCalls, ...silent, compileMs };
-          return cacheUsage ? { ...base, cacheUsage } : base;
+          return { ...base, ...(cacheUsage ? { cacheUsage } : {}), ...(costUsage ? { costUsage } : {}) };
         } finally {
           removeIsolatedDirs(entry);
         }

test/codex-harness.test.ts

@@ -9,6 +9,7 @@ import {
   codexNonRetryable,
   codexProviderFailure,
   codexUsageTotals,
+
   codexChildToolAllowed,
   codexReasoningEffort,
   codexReplayCallId,
@@ -19,6 +20,7 @@ import {
   createCodexHarness,
   prepareCodexHome,
 } from "../src/harness/codex-harness.ts";
+import { usageToTurnTelemetry } from "../src/harness/harness.ts";
 import type { HarnessLlmRequestRecord, HarnessTurnInput } from "../src/harness/harness.ts";
 import { NonRetryableTurnError } from "../src/core/turn-error.ts";
 import type { ScopeId, Session, SessionEntry } from "../src/types.ts";
@@ -71,7 +73,7 @@ rl.on("line", (line) => {
     send({ method: "item/started", params: { threadId: "thread-1", turnId: "turn-1", item: { type: "collabAgentToolCall", id: "collab-1", tool: "spawnAgent", status: "inProgress", senderThreadId: "thread-1", receiverThreadIds: ["child-1"], prompt: "return ALPHA", agentsStates: { "child-1": { status: "running", message: null } } } } });
     send({ method: "thread/tokenUsage/updated", params: { threadId: "child-1", tokenUsage: { total: { inputTokens: 70 }, last: { inputTokens: 70 } } } });
     send({ method: "item/completed", params: { threadId: "thread-1", turnId: "turn-1", item: { type: "collabAgentToolCall", id: "collab-1", tool: "spawnAgent", status: "completed", senderThreadId: "thread-1", receiverThreadIds: ["child-1"], prompt: "return ALPHA", agentsStates: { "child-1": { status: "completed", message: "ALPHA" } } } } });
-    send({ method: "thread/tokenUsage/updated", params: { threadId: "thread-1", tokenUsage: { total: { inputTokens: 250 }, last: { inputTokens: 150 } } } });
+    send({ method: "thread/tokenUsage/updated", params: { threadId: "thread-1", tokenUsage: { total: { inputTokens: 250, outputTokens: 60, cachedInputTokens: 120 }, last: { inputTokens: 150 } } } });
     send({ method: "item/agentMessage/delta", params: { threadId: "thread-1", turnId: "turn-1", itemId: "item-1", delta: "hello" } });
     send({ method: "item/completed", params: { threadId: "thread-1", turnId: "turn-1", item: { type: "agentMessage", id: "item-1", text: "hello", phase: "final_answer", memoryCitation: null } } });
     return send({ method: "turn/completed", params: { threadId: "thread-1", turn: { id: "turn-1", status: "completed", items: [], itemsView: "notLoaded" } } });
@@ -205,6 +207,16 @@ test("Codex harness drives app-server JSON-RPC with a read-only jail", async (t)
   assert.equal(result.reply, "hello");
   assert.deepEqual(deltas, ["hello"]);
   assert.deepEqual(modelCalls, [100, 70, 150]);
+  assert.deepEqual(
+    result.cacheUsage,
+    { cacheRead: 120, cacheWrite: 0, uncachedInput: 250 },
+    "turn-level cache telemetry comes off the live thread totals",
+  );
+  assert.deepEqual(
+    result.costUsage,
+    { outputTokens: 60, costUsd: 0 },
+    "turn-level spend is surfaced (codex reports no cost, so it stays 0)",
+  );
   assert.deepEqual(
     entries.map((entry) => entry.type),
     ["user", "tool_call", "tool_result", "assistant"],
@@ -498,6 +510,17 @@ test("Codex never classifies its own infrastructure failures as terminal", () =>
   assert.ok(!(codexProviderFailure("socket hang up") instanceof NonRetryableTurnError));
 });
 
+test("Codex turn telemetry maps thread usage totals onto cache and spend", () => {
+  assert.deepEqual(
+    usageToTurnTelemetry({ input: 250, output: 60, cacheRead: 120, cacheWrite: 0, totalTokens: 430, costUsd: 0 }),
+    {
+      cacheUsage: { cacheRead: 120, cacheWrite: 0, uncachedInput: 250 },
+      costUsage: { outputTokens: 60, costUsd: 0 },
+    },
+  );
+  assert.equal(usageToTurnTelemetry(null), null, "a turn with no usage reports no telemetry");
+});
+
 test("Codex reads cumulative usage totals off the app-server's token notification", () => {
   assert.deepEqual(
     codexUsageTotals({

test/opencode-harness.test.ts

@@ -150,6 +150,16 @@ test("OpenCode records real usage, cost, and timings for each captured model cal
     totalTokens: 183,
     costUsd: 0.0353,
   });
+  assert.deepEqual(
+    result.cacheUsage,
+    { cacheRead: 50, cacheWrite: 10, uncachedInput: 100 },
+    "turn-level cache telemetry sums the flushed per-capture usage",
+  );
+  assert.deepEqual(
+    result.costUsage,
+    { outputTokens: 20, costUsd: 0.0353 },
+    "turn-level spend carries the provider-computed cost",
+  );
 });
 
 test("OpenCode startup failure reports the sidecar's real output and honors the configured timeout", async (t) => {