diff --git a/src/admin/metrics-sink.ts b/src/admin/metrics-sink.ts index df9cb1743..8dfba6411 100644 --- a/src/admin/metrics-sink.ts +++ b/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 { diff --git a/src/admin/postgres-metrics-sink.ts b/src/admin/postgres-metrics-sink.ts index 29187f828..57bb48161 100644 --- a/src/admin/postgres-metrics-sink.ts +++ b/src/admin/postgres-metrics-sink.ts @@ -37,6 +37,8 @@ const COLUMNS: readonly EventColumn[] = [ ["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 = [ diff --git a/src/api/routes/admin/observability.ts b/src/api/routes/admin/observability.ts index aee13d633..e856cec04 100644 --- a/src/api/routes/admin/observability.ts +++ b/src/api/routes/admin/observability.ts @@ -150,6 +150,18 @@ export async function metrics(ctx: ApiCtx): Promise { 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 { series, anatomy, cache, + spend, phases, }); } diff --git a/src/core/orchestrator.ts b/src/core/orchestrator.ts index 3a17db7ab..a5e86ba4f 100644 --- a/src/core/orchestrator.ts +++ b/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) { diff --git a/src/harness/claude-harness.ts b/src/harness/claude-harness.ts index 2edbb26eb..8d09c77b3 100644 --- a/src/harness/claude-harness.ts +++ b/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 { diff --git a/src/harness/codex-harness.ts b/src/harness/codex-harness.ts index 46d79a12b..a89137bdb 100644 --- a/src/harness/codex-harness.ts +++ b/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,6 +833,12 @@ 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 } : {}), @@ -839,6 +846,7 @@ export function createCodexHarness(opts: CodexHarnessOptions = {}): Harness { ...(ref.pendingApprovals?.length ? { pendingApprovals: ref.pendingApprovals } : {}), ...(ref.pausedOnApproval ? { pausedOnApproval: true } : {}), modelCalls: state.modelCalls, + ...(telemetry ?? {}), ...(state.tapeWriteFailed ? { tapeWriteFailed: true } : {}), }; } finally { diff --git a/src/harness/harness.ts b/src/harness/harness.ts index 77940e80a..14d5c1491 100644 --- a/src/harness/harness.ts +++ b/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; diff --git a/src/harness/mock-harness.ts b/src/harness/mock-harness.ts index 319732612..36ca80574 100644 --- a/src/harness/mock-harness.ts +++ b/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 { diff --git a/src/harness/opencode-harness.ts b/src/harness/opencode-harness.ts index 71fa163bd..aa35dc675 100644 --- a/src/harness/opencode-harness.ts +++ b/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; 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 => { @@ -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,6 +1090,12 @@ 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 } : {}), @@ -1079,6 +1103,7 @@ export function createOpenCodeHarness(opts: OpenCodeHarnessOptions = {}): Harnes ...(ref.pendingApprovals?.length ? { pendingApprovals: ref.pendingApprovals } : {}), ...(ref.pausedOnApproval ? { pausedOnApproval: true } : {}), modelCalls: state.captures.length, + ...(usageToTurnTelemetry(state.turnUsage) ?? {}), ...(tapeWriteFailed ? { tapeWriteFailed: true } : {}), }; } finally { diff --git a/src/harness/pi-harness.ts b/src/harness/pi-harness.ts index 73ff2ba74..6d86e5758 100644 --- a/src/harness/pi-harness.ts +++ b/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,6 +2004,7 @@ 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, @@ -1996,7 +2012,7 @@ export function createPiHarness(opts?: PiHarnessOptions): Harness { 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); } diff --git a/test/codex-harness.test.ts b/test/codex-harness.test.ts index 9b9cf957e..0b1586369 100644 --- a/test/codex-harness.test.ts +++ b/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({ diff --git a/test/opencode-harness.test.ts b/test/opencode-harness.test.ts index 091039139..8168c291b 100644 --- a/test/opencode-harness.test.ts +++ b/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) => { diff --git a/test/turn-metrics-cost.test.ts b/test/turn-metrics-cost.test.ts new file mode 100644 index 000000000..ed15a9035 --- /dev/null +++ b/test/turn-metrics-cost.test.ts @@ -0,0 +1,66 @@ +import "./support/auto-fake-sprites.ts"; + +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { AddressInfo } from "node:net"; +import { createInsecureTestServer } from "../src/api/server.ts"; +import { buildApp } from "../src/wiring.ts"; +import type { TurnRequest } from "../src/types.ts"; +import { testConfig } from "./support/test-config.ts"; + +function start() { + const built = buildApp(testConfig({ dataDir: mkdtempSync(join(tmpdir(), "cost-obs-")) })); + const server = createInsecureTestServer(built.app, { + admin: built.admin, + sessions: built.sessions, + auditLog: built.auditLog, + metrics: built.metrics, + runs: built.runs, + }); + server.listen(0); + const base = `http://localhost:${(server.address() as AddressInfo).port}`; + return { base, built, close: () => new Promise((r) => server.close(() => r())) }; +} + +const ALICE = { "x-admin-actor": "admin-alice@default-org" }; +const getJson = async (base: string, path: string, headers: Record = ALICE): Promise => + (await fetch(base + path, { headers })).json(); + +test("metrics: turn spend (output tokens + cost) is surfaced on the metrics endpoint", async () => { + const s = start(); + try { + const turn: TurnRequest = { + surface: "test", + actor: { externalId: "U1" }, + conversation: { kind: "dm", threadRef: "dm:U1:c1" }, + text: "hello there", + }; + assert.equal((await s.built.app.turn(turn)).status, "ok"); + + const m = await getJson(s.base, "/v1/admin/metrics?scope=org:default-org"); + assert.ok(m.spend, "the metrics response carries a spend block"); + assert.ok(m.spend.samples >= 1, "the turn carried spend telemetry"); + assert.ok(m.spend.turnsWithKnownCost >= 1, "the mock harness reports known (if zero) cost"); + assert.ok(m.spend.outputTokensTotal > 0, "output tokens are summed per turn"); + assert.equal(typeof m.spend.costUsdTotal, "number", "cost total is numeric even when the mock reports 0"); + } finally { + await s.close(); + } +}); + +test("metrics: an empty scope yields a present-but-empty spend block", async () => { + const s = start(); + try { + const m = await getJson(s.base, "/v1/admin/metrics?scope=personal:nobody"); + assert.ok(m.spend, "spend block is always present"); + assert.equal(m.spend.samples, 0); + assert.equal(m.spend.turnsWithKnownCost, 0); + assert.equal(m.spend.outputTokensTotal, 0); + assert.equal(m.spend.costUsdTotal, 0); + } finally { + await s.close(); + } +});