diff --git a/packages/core/src/__tests__/model-call-attempt.test.ts b/packages/core/src/__tests__/model-call-attempt.test.ts index 11a6490c0f..f4293b8d8c 100644 --- a/packages/core/src/__tests__/model-call-attempt.test.ts +++ b/packages/core/src/__tests__/model-call-attempt.test.ts @@ -25,10 +25,8 @@ import { MODEL_CALL_ATTEMPT_SCHEMA_VERSION, PROMPT_COMPOSITION_MAX_TOOLS, decodeModelCallAttempt, + dedupeModelCallAttempts, groupModelCallAttempts, - settledAttempt, - sumModelCallCostUsd, - summarizeModelCallCoverage, type ModelCallAttempt, } from '../model-call-attempt.js'; @@ -323,72 +321,27 @@ describe('ModelCallAttempt codec', () => { }); describe('ModelCallAttempt projections', () => { - test('groups retries under one logical call and derives the settled attempt', () => { + test('groups retries under one logical call', () => { const groups = groupModelCallAttempts([ attempt({ attemptId: 'a-0', attempt: 0, status: 'failed', costUsd: 0.001 }), attempt({ attemptId: 'a-1', attempt: 1, status: 'completed', costUsd: 0.004 }), attempt({ attemptId: 'b-0', logicalCallId: 'call-2' }), ]); assert.equal(groups.length, 2); - const retried = groups.find((g) => g.logicalCallId === 'call-1'); - assert.equal(retried?.attempts.length, 2); - assert.equal(settledAttempt(retried!)?.attemptId, 'a-1'); + assert.equal(groups.find((g) => g.logicalCallId === 'call-1')?.attempts.length, 2); }); - test('coverage counts priced, unpriced, and usage bases separately', () => { - const coverage = summarizeModelCallCoverage([ - attempt({ attemptId: 'a' }), - attempt({ attemptId: 'b', costBasis: 'unpriced', costUsd: undefined }), - attempt({ - attemptId: 'c', - costBasis: 'unpriced', - costUsd: undefined, - usageBasis: 'missing', - inputTokens: undefined, - outputTokens: undefined, - }), - attempt({ attemptId: 'd', usageBasis: 'partial', outputTokens: undefined }), - ]); - assert.deepEqual(coverage, { - attempts: 4, - pricedAttempts: 2, - unpricedAttempts: 2, - usageReportedAttempts: 2, - usagePartialAttempts: 1, - usageMissingAttempts: 1, - }); - }); - - test('cost sum reports the qualifying coverage alongside the total', () => { - const { costUsd, coverage } = sumModelCallCostUsd([ - attempt({ attemptId: 'a', costUsd: 0.004 }), - attempt({ attemptId: 'b', costUsd: 0.006 }), - attempt({ attemptId: 'c', costBasis: 'unpriced', costUsd: undefined }), - ]); - assert.equal(Math.round(costUsd * 1000) / 1000, 0.01); - assert.equal(coverage.unpricedAttempts, 1); - }); - - test('a replayed attemptId is counted once through sum and coverage', () => { + test('a replayed attemptId is the same call, kept at its last value', () => { const stream = [ attempt({ attemptId: 'a', logicalCallId: 'call-1', costUsd: 0.004 }), attempt({ attemptId: 'b', logicalCallId: 'call-2', costUsd: 0.006 }), attempt({ attemptId: 'a', logicalCallId: 'call-1', costUsd: 0.005 }), ]; - const { costUsd, coverage } = sumModelCallCostUsd(stream); - assert.equal(Math.round(costUsd * 1000) / 1000, 0.011); - assert.equal(coverage.attempts, 2); - assert.equal(coverage.pricedAttempts, 2); - assert.equal(summarizeModelCallCoverage(stream).attempts, 2); + const unique = dedupeModelCallAttempts(stream); + assert.deepEqual( + unique.map((a) => a.costUsd), + [0.005, 0.006], + ); assert.equal(groupModelCallAttempts(stream).length, 2); }); - - test('a genuinely free priced call is distinguishable from an unpriced one', () => { - const free = attempt({ attemptId: 'free', costBasis: 'priced', costUsd: 0 }); - const unknown = attempt({ attemptId: 'unknown', costBasis: 'unpriced', costUsd: undefined }); - const { costUsd, coverage } = sumModelCallCostUsd([free, unknown]); - assert.equal(costUsd, 0); - assert.equal(coverage.pricedAttempts, 1); - assert.equal(coverage.unpricedAttempts, 1); - }); }); diff --git a/packages/core/src/__tests__/model-call-usage-projection.test.ts b/packages/core/src/__tests__/model-call-usage-projection.test.ts deleted file mode 100644 index c8e57ec581..0000000000 --- a/packages/core/src/__tests__/model-call-usage-projection.test.ts +++ /dev/null @@ -1,303 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -import { describe, test } from 'node:test'; -import assert from 'node:assert/strict'; - -import { MODEL_CALL_ATTEMPT_SCHEMA_VERSION, type ModelCallAttempt } from '../model-call-attempt.js'; -import { - projectModelCallUsageBuckets, - projectModelCallUsageLogs, - projectModelCallUsageSummary, - selectModelCallAttempts, - usageStatusForAttempt, -} from '../model-call-usage-projection.js'; - -// A realistic epoch-ms clock: a small NOW would push "40 days ago" negative and -// silently fall outside the `all` range, which starts at 0. -const NOW = 1_750_000_000_000; - -function attempt(overrides: Partial = {}): ModelCallAttempt { - return { - schemaVersion: MODEL_CALL_ATTEMPT_SCHEMA_VERSION, - logicalCallId: 'call-1', - attemptId: 'attempt-1', - traceId: 'trace-1', - sessionId: 'session-1', - runId: 'run-1', - turnId: 'turn-1', - step: 0, - attempt: 0, - callKind: 'main', - providerId: 'anthropic', - modelId: 'claude-opus-5', - startedAt: NOW - 1_000, - completedAt: NOW - 500, - latencyMs: 500, - status: 'completed', - usageBasis: 'reported', - inputTokens: 100, - outputTokens: 20, - costBasis: 'priced', - costUsd: 0.004, - ...overrides, - }; -} - -describe('model-call usage projection', () => { - test('a log row keeps its cost basis, so free and unpriced stay distinguishable', () => { - // The page-level coverage says how many rows were unpriced but not which - // ones. Without a per-row basis a genuinely free call and a call whose - // price could not be resolved both render as $0. - const { rows } = projectModelCallUsageLogs( - [ - attempt({ attemptId: 'free', logicalCallId: 'free', costBasis: 'priced', costUsd: 0 }), - attempt({ - attemptId: 'unknown', - logicalCallId: 'unknown', - costBasis: 'unpriced', - costUsd: undefined, - }), - ], - { range: 'all' }, - NOW, - ); - - const free = rows.find((row) => row.id === 'free'); - const unknown = rows.find((row) => row.id === 'unknown'); - assert.equal(free?.costBasis, 'priced'); - assert.equal(free?.costUsd, 0); - assert.equal(unknown?.costBasis, 'unpriced'); - assert.equal(unknown?.costUsd, undefined); - assert.equal(Object.hasOwn(unknown ?? {}, 'costUsd'), false); - }); - - test('a total never counts unpriced spend as zero, and says so in coverage', () => { - // The old per-send table had nowhere to record "we could not price this", - // so it wrote 0 and unpriced spend looked free. The total here excludes it - // and the coverage reports it instead. - const summary = projectModelCallUsageSummary( - [ - attempt({ attemptId: 'a', costUsd: 0.004 }), - attempt({ attemptId: 'b', costBasis: 'unpriced', costUsd: undefined }), - ], - { range: 'all' }, - NOW, - ); - assert.equal(Math.round(summary.totalCostUsd * 1000) / 1000, 0.004); - assert.equal(summary.totalRequests, 2); - assert.equal(summary.coverage.pricedAttempts, 1); - assert.equal(summary.coverage.unpricedAttempts, 1); - }); - - test('the summary sums recorded call time over the rows it counts', () => { - const summary = projectModelCallUsageSummary( - [ - attempt({ attemptId: 'a', logicalCallId: 'a', latencyMs: 1_200 }), - attempt({ attemptId: 'b', logicalCallId: 'b', latencyMs: 300 }), - ], - { range: 'all' }, - NOW, - ); - assert.equal(summary.totalDurationMs, 1_500); - assert.equal(summary.totalRequests, 2); - }); - - test('a genuinely free call is still counted as priced', () => { - const summary = projectModelCallUsageSummary( - [attempt({ attemptId: 'free', costUsd: 0 })], - { range: 'all' }, - NOW, - ); - assert.equal(summary.totalCostUsd, 0); - assert.equal(summary.coverage.pricedAttempts, 1); - assert.equal(summary.coverage.unpricedAttempts, 0); - }); - - test('usage-missing records are reported separately from unpriced ones', () => { - const summary = projectModelCallUsageSummary( - [ - attempt({ - attemptId: 'no-usage', - status: 'failed', - usageBasis: 'missing', - inputTokens: undefined, - outputTokens: undefined, - costBasis: 'unpriced', - costUsd: undefined, - }), - ], - { range: 'all' }, - NOW, - ); - assert.equal(summary.coverage.usageMissingAttempts, 1); - assert.equal(summary.coverage.unpricedAttempts, 1); - assert.equal(summary.totalTokens.total, 0); - }); - - test('does not let one malformed cache reading inflate the Session cache total', () => { - const summary = projectModelCallUsageSummary( - [ - attempt({ - attemptId: 'malformed-cache', - inputTokens: 100, - cacheReadInputTokens: 200, - }), - attempt({ - attemptId: 'cache-miss', - inputTokens: 100, - cacheReadInputTokens: 0, - }), - ], - { range: 'all' }, - NOW, - ); - - assert.equal(summary.totalTokens.input, 200); - assert.equal(summary.totalTokens.cacheRead, 100); - }); - - test('preserves provider cache-only evidence without inventing an input total', () => { - const cacheOnly = attempt({ - attemptId: 'cache-only', - usageBasis: 'partial', - inputTokens: undefined, - outputTokens: undefined, - cacheReadInputTokens: 10, - }); - const summary = projectModelCallUsageSummary([cacheOnly], { range: 'all' }, NOW); - - assert.equal(summary.totalTokens.input, 0); - assert.equal(summary.totalTokens.cacheRead, 10); - assert.equal(summary.cacheHitRequests, 1); - assert.equal(summary.coverage.usagePartialAttempts, 1); - - const bucket = projectModelCallUsageBuckets([cacheOnly], { range: 'all' }, 'provider', NOW)[0]; - assert.equal(bucket?.inputTokens, 0); - assert.equal(bucket?.cacheReadTokens, 10); - - const log = projectModelCallUsageLogs([cacheOnly], { range: 'all' }, NOW).rows[0]; - assert.equal(log?.inputTokens, 0); - assert.equal(log?.cacheReadTokens, 10); - }); - - test('a replayed attemptId is counted once', () => { - const row = attempt({ attemptId: 'dup' }); - const summary = projectModelCallUsageSummary([row, row], { range: 'all' }, NOW); - assert.equal(summary.totalRequests, 1); - assert.equal(Math.round(summary.totalCostUsd * 1000) / 1000, 0.004); - }); - - test('filters by Session, range, provider, model, and status', () => { - const rows = [ - attempt({ attemptId: 'recent' }), - attempt({ attemptId: 'old', completedAt: NOW - 40 * 86_400_000 }), - attempt({ attemptId: 'other-provider', providerId: 'openai', modelId: 'gpt-x' }), - attempt({ attemptId: 'failed', status: 'failed' }), - attempt({ attemptId: 'other-session', sessionId: 'session-2' }), - ]; - assert.equal(selectModelCallAttempts(rows, { range: '24h' }, NOW).rows.length, 4); - assert.equal( - selectModelCallAttempts(rows, { range: 'all', sessionId: 'session-1' }, NOW).rows.length, - 4, - ); - assert.equal( - selectModelCallAttempts(rows, { range: 'all', providerId: 'openai' }, NOW).rows.length, - 1, - ); - assert.equal( - selectModelCallAttempts(rows, { range: 'all', modelId: 'claude-opus-5' }, NOW).rows.length, - 4, - ); - assert.equal( - selectModelCallAttempts(rows, { range: 'all', status: 'error' }, NOW).rows.length, - 1, - ); - assert.equal( - selectModelCallAttempts(rows, { range: 'all', status: 'all' }, NOW).rows.length, - 5, - ); - }); - - test('interrupted counts as aborted, not as an error', () => { - // Collapsing a cut-short call into `error` would inflate the error rate - // with user cancellations. - assert.equal(usageStatusForAttempt('completed'), 'success'); - assert.equal(usageStatusForAttempt('failed'), 'error'); - assert.equal(usageStatusForAttempt('aborted'), 'aborted'); - assert.equal(usageStatusForAttempt('interrupted'), 'aborted'); - - const summary = projectModelCallUsageSummary( - [attempt({ attemptId: 'cut', status: 'interrupted' })], - { range: 'all' }, - NOW, - ); - assert.equal(summary.errorRequests, 0); - }); - - test('buckets group by provider and model and exclude unpriced cost', () => { - const rows = [ - attempt({ attemptId: 'a', providerId: 'anthropic', costUsd: 0.004 }), - attempt({ attemptId: 'b', providerId: 'anthropic', costUsd: 0.006 }), - attempt({ - attemptId: 'c', - providerId: 'openai', - modelId: 'gpt-x', - costBasis: 'unpriced', - costUsd: undefined, - }), - ]; - const byProvider = projectModelCallUsageBuckets(rows, { range: 'all' }, 'provider', NOW); - assert.deepEqual( - byProvider.map((b) => [b.key, b.requests]), - [ - ['anthropic', 2], - ['openai', 1], - ], - ); - assert.equal(Math.round((byProvider[0]?.costUsd ?? 0) * 1000) / 1000, 0.01); - assert.equal(byProvider[1]?.costUsd, 0); - - const byModel = projectModelCallUsageBuckets(rows, { range: 'all' }, 'model', NOW); - assert.equal(byModel.length, 2); - assert.ok(byModel.some((b) => b.key === 'anthropic:claude-opus-5')); - }); - - test('logs page newest first and carry coverage for the whole match', () => { - const rows = [ - attempt({ attemptId: 'older', completedAt: NOW - 3_000 }), - attempt({ attemptId: 'newer', completedAt: NOW - 1_000 }), - attempt({ - attemptId: 'unpriced', - completedAt: NOW - 2_000, - costBasis: 'unpriced', - costUsd: undefined, - }), - ]; - const page = projectModelCallUsageLogs(rows, { range: 'all' }, NOW, 0, 2); - assert.deepEqual( - page.rows.map((r) => r.id), - ['newer', 'unpriced'], - ); - assert.equal(page.total, 3); - // Coverage describes every matching record, not just the returned page. - assert.equal(page.coverage.attempts, 3); - assert.equal(page.coverage.unpricedAttempts, 1); - }); -}); diff --git a/packages/core/src/__tests__/usage-ledger-merge.test.ts b/packages/core/src/__tests__/usage-ledger-merge.test.ts index 9f031d064f..e22e1954d5 100644 --- a/packages/core/src/__tests__/usage-ledger-merge.test.ts +++ b/packages/core/src/__tests__/usage-ledger-merge.test.ts @@ -20,7 +20,12 @@ import { describe, test } from 'node:test'; import assert from 'node:assert/strict'; -import { MODEL_CALL_ATTEMPT_SCHEMA_VERSION, type ModelCallAttempt } from '../model-call-attempt.js'; +import { EMPTY_MODEL_CALL_COVERAGE } from '../usage-ledger-merge.js'; +import type { + ModelCallUsageBuckets, + ModelCallUsageLogs, + ModelCallUsageSummary, +} from '../model-call-usage-projection.js'; import { EMPTY_USAGE_PROVENANCE, estimatedUsageCost, @@ -37,33 +42,63 @@ import type { UsageBucket, UsageLogRow, UsageSummaryV2 } from '../usage-stats/ty // and silently fall outside the `all` range, which starts at 0. const NOW = 1_750_000_000_000; -function attempt(overrides: Partial = {}): ModelCallAttempt { +/** + * A canonical answer as the ledger returns it: already aggregated, with what + * qualifies it. The merge's job is to add two answers and record where each + * half came from; how the canonical half was computed is the ledger's. + */ +function canonical( + projection: T, + overrides: { unreadableRecords?: number; pendingRepairs?: number } = {}, +): { projection: T; unreadableRecords: number; pendingRepairs: number } { + return { projection, unreadableRecords: 0, pendingRepairs: 0, ...overrides }; +} + +function canonicalSummary(overrides: Partial = {}): ModelCallUsageSummary { return { - schemaVersion: MODEL_CALL_ATTEMPT_SCHEMA_VERSION, - logicalCallId: 'call-1', - attemptId: 'attempt-1', - traceId: 'trace-1', - sessionId: 'session-1', - runId: 'run-1', - turnId: 'turn-1', - step: 0, - attempt: 0, - callKind: 'main', - providerId: 'anthropic', - modelId: 'claude-opus-5', - startedAt: NOW - 1_000, - completedAt: NOW - 500, - latencyMs: 500, - status: 'completed', - usageBasis: 'reported', - inputTokens: 100, - outputTokens: 20, - costBasis: 'priced', - costUsd: 0.004, + range: { from: 0, to: NOW }, + totalRequests: 1, + totalCostUsd: 0.004, + totalDurationMs: 500, + totalTokens: { + input: 100, + output: 20, + cacheMiss: 0, + cacheRead: 0, + cacheWrite: 0, + reasoning: 0, + total: 120, + }, + cacheHitRequests: 0, + cacheCreateRequests: 0, + errorRequests: 0, + coverage: { ...EMPTY_MODEL_CALL_COVERAGE, attempts: 1, pricedAttempts: 1 }, ...overrides, }; } +function canonicalBuckets(buckets: UsageBucket[]): ModelCallUsageBuckets { + return { + buckets, + coverage: { + ...EMPTY_MODEL_CALL_COVERAGE, + attempts: buckets.reduce((total, bucket) => total + bucket.requests, 0), + }, + }; +} + +function canonicalLogs(rows: UsageLogRow[]): ModelCallUsageLogs { + return { + rows, + total: rows.length, + coverage: { ...EMPTY_MODEL_CALL_COVERAGE, attempts: rows.length }, + }; +} + +function canonicalLog(id: string, ts: number, overrides: Partial = {}): UsageLogRow { + return { ...legacyLog(id, ts), costBasis: 'priced', costUsd: 0.004, ...overrides }; +} + function legacySummary(overrides: Partial = {}): UsageSummaryV2 { return { range: { from: 0, to: NOW }, @@ -126,12 +161,7 @@ function legacyBucket(overrides: Partial = {}): UsageBucket { describe('usage ledger merge', () => { test('sums both sources and reports how much came from the frozen table', () => { - const merged = mergeUsageSummary( - legacySummary(), - { attempts: [attempt({ attemptId: 'a' })], unreadableRecords: 0, pendingRepairs: 0 }, - { range: 'all' }, - NOW, - ); + const merged = mergeUsageSummary(legacySummary(), canonical(canonicalSummary())); assert.equal(merged.totalRequests, 3); assert.equal(merged.totalCostUsd, 0.01 + 0.004); @@ -148,13 +178,7 @@ describe('usage ledger merge', () => { test('merges recorded call time from both ledgers', () => { const merged = mergeUsageSummary( legacySummary({ totalDurationMs: 700 }), - { - attempts: [attempt({ attemptId: 'a', latencyMs: 500 })], - unreadableRecords: 0, - pendingRepairs: 0, - }, - { range: 'all' }, - NOW, + canonical(canonicalSummary({ totalDurationMs: 500 })), ); assert.equal(merged.totalDurationMs, 1_200); @@ -162,13 +186,7 @@ describe('usage ledger merge', () => { // with no recorded time simply contributes a zero to the sum. const canonicalOnly = mergeUsageSummary( legacySummary(), - { - attempts: [attempt({ attemptId: 'a', latencyMs: 500 })], - unreadableRecords: 0, - pendingRepairs: 0, - }, - { range: 'all' }, - NOW, + canonical(canonicalSummary({ totalDurationMs: 500 })), ); assert.equal(canonicalOnly.totalDurationMs, 500); }); @@ -176,13 +194,12 @@ describe('usage ledger merge', () => { test('unpriced canonical spend stays out of the total and is reported instead', () => { const merged = mergeUsageSummary( legacySummary({ totalRequests: 0, totalCostUsd: 0, errorRequests: 0 }), - { - attempts: [attempt({ attemptId: 'a', costBasis: 'unpriced', costUsd: undefined })], - unreadableRecords: 0, - pendingRepairs: 0, - }, - { range: 'all' }, - NOW, + canonical( + canonicalSummary({ + totalCostUsd: 0, + coverage: { ...EMPTY_MODEL_CALL_COVERAGE, attempts: 1, unpricedAttempts: 1 }, + }), + ), ); assert.equal(merged.totalCostUsd, 0); @@ -194,9 +211,7 @@ describe('usage ledger merge', () => { test('records that could not be decoded are reported, not silently dropped', () => { const merged = mergeUsageSummary( legacySummary(), - { attempts: [], unreadableRecords: 3, pendingRepairs: 0 }, - { range: 'all' }, - NOW, + canonical(canonicalSummary(), { unreadableRecords: 3 }), ); assert.equal(merged.provenance.unreadableRecords, 3); @@ -205,17 +220,19 @@ describe('usage ledger merge', () => { test('buckets sharing a key combine, re-weighting the per-request means', () => { const merged = mergeUsageBuckets( [legacyBucket()], - { - attempts: [ - attempt({ attemptId: 'a', latencyMs: 800, status: 'completed' }), - attempt({ attemptId: 'b', latencyMs: 800, status: 'failed' }), - ], - unreadableRecords: 0, - pendingRepairs: 0, - }, - { range: 'all' }, - 'model', - NOW, + canonical( + canonicalBuckets([ + legacyBucket({ + requests: 2, + costUsd: 0.008, + avgLatencyMs: 800, + errorRate: 0.5, + inputTokens: 200, + outputTokens: 40, + totalTokens: 240, + }), + ]), + ), ); assert.equal(merged.buckets.length, 1); @@ -232,10 +249,7 @@ describe('usage ledger merge', () => { test('buckets with no counterpart in the other source pass through intact', () => { const merged = mergeUsageBuckets( [legacyBucket({ key: 'openai:gpt-5', label: 'openai:gpt-5' })], - { attempts: [attempt({ attemptId: 'a' })], unreadableRecords: 0, pendingRepairs: 0 }, - { range: 'all' }, - 'model', - NOW, + canonical(canonicalBuckets([legacyBucket({ requests: 1 })])), ); assert.deepEqual(merged.buckets.map((bucket) => bucket.key).sort(), [ @@ -246,41 +260,21 @@ describe('usage ledger merge', () => { test('log pages interleave both sources newest first and page across the boundary', () => { const legacyRows = [legacyLog('legacy-new', NOW - 100), legacyLog('legacy-old', NOW - 900)]; - const canonical = { - attempts: [ - attempt({ attemptId: 'canonical-mid', logicalCallId: 'c-mid', completedAt: NOW - 400 }), - attempt({ - attemptId: 'canonical-oldest', - logicalCallId: 'c-old', - completedAt: NOW - 1_200, - }), - ], - unreadableRecords: 0, - pendingRepairs: 0, - }; - - const first = mergeUsageLogs( - { rows: legacyRows, total: 2 }, - canonical, - { range: 'all' }, - NOW, - 0, - 2, + const canonicalPage = canonical( + canonicalLogs([ + canonicalLog('canonical-mid', NOW - 400), + canonicalLog('canonical-oldest', NOW - 1_200), + ]), ); + + const first = mergeUsageLogs({ rows: legacyRows, total: 2 }, canonicalPage, 0, 2); assert.deepEqual( first.rows.map((row) => row.id), ['legacy-new', 'canonical-mid'], ); assert.equal(first.total, 4); - const second = mergeUsageLogs( - { rows: legacyRows, total: 2 }, - canonical, - { range: 'all' }, - NOW, - 2, - 2, - ); + const second = mergeUsageLogs({ rows: legacyRows, total: 2 }, canonicalPage, 2, 2); assert.deepEqual( second.rows.map((row) => row.id), ['legacy-old', 'canonical-oldest'], @@ -289,14 +283,7 @@ describe('usage ledger merge', () => { }); test('a page beyond both sources is empty rather than throwing', () => { - const merged = mergeUsageLogs( - { rows: [], total: 0 }, - { attempts: [], unreadableRecords: 0, pendingRepairs: 0 }, - { range: 'all' }, - NOW, - 0, - 10, - ); + const merged = mergeUsageLogs({ rows: [], total: 0 }, canonical(canonicalLogs([])), 0, 10); assert.deepEqual(merged.rows, []); assert.equal(merged.total, 0); diff --git a/packages/core/src/model-call-attempt.ts b/packages/core/src/model-call-attempt.ts index 844218cdbd..d017abf2ab 100644 --- a/packages/core/src/model-call-attempt.ts +++ b/packages/core/src/model-call-attempt.ts @@ -609,15 +609,6 @@ export function decodeModelCallAttempt(value: unknown): ModelCallAttempt { return value as unknown as ModelCallAttempt; } -export function isModelCallAttempt(value: unknown): value is ModelCallAttempt { - try { - decodeModelCallAttempt(value); - return true; - } catch { - return false; - } -} - /** * Collapses re-appended records by `attemptId`, keeping the last occurrence. * @@ -655,42 +646,6 @@ export function groupModelCallAttempts(attempts: readonly ModelCallAttempt[]): M return [...groups.values()]; } -/** - * The attempt that settled a logical call: the highest `attempt` ordinal that - * reached a provider outcome. Terminality is a projection concern, not a stored - * field, so it is derived rather than recorded. - */ -export function settledAttempt(group: ModelCallGroup): ModelCallAttempt | undefined { - let settled: ModelCallAttempt | undefined; - for (const attempt of group.attempts) { - if (!settled || attempt.attempt > settled.attempt) settled = attempt; - } - return settled; -} - -/** - * Extracts the canonical attempts a run committed, from that run's AgentRun - * events. This is the projection the Usage read model is rebuilt through, so it - * has to be total: an event that cannot be decoded is counted, not thrown, or - * one bad record would block every later one in the same run from ever being - * projected. - */ -export function modelCallAttemptsFromRunEvents( - events: readonly { readonly type: string; readonly data?: Record }[], -): { attempts: ModelCallAttempt[]; unreadableEvents: number } { - const attempts: ModelCallAttempt[] = []; - let unreadableEvents = 0; - for (const event of events) { - if (event.type !== MODEL_CALL_ATTEMPT_EVENT_TYPE) continue; - try { - attempts.push(decodeModelCallAttempt(event.data)); - } catch { - unreadableEvents += 1; - } - } - return { attempts, unreadableEvents }; -} - /** * Classification of the records present in a set. * @@ -709,42 +664,3 @@ export interface ModelCallCoverage { /** Dispatched calls the provider never reported usage for. */ usageMissingAttempts: number; } - -export function summarizeModelCallCoverage( - attempts: readonly ModelCallAttempt[], -): ModelCallCoverage { - const unique = dedupeModelCallAttempts(attempts); - const coverage: ModelCallCoverage = { - attempts: unique.length, - pricedAttempts: 0, - unpricedAttempts: 0, - usageReportedAttempts: 0, - usagePartialAttempts: 0, - usageMissingAttempts: 0, - }; - for (const attempt of unique) { - if (attempt.costBasis === 'priced') coverage.pricedAttempts += 1; - else coverage.unpricedAttempts += 1; - if (attempt.usageBasis === 'reported') coverage.usageReportedAttempts += 1; - else if (attempt.usageBasis === 'partial') coverage.usagePartialAttempts += 1; - else coverage.usageMissingAttempts += 1; - } - return coverage; -} - -/** - * Sums cost across attempts. Returns the total alongside the coverage that - * qualifies it, because a bare number cannot express "plus an unknown amount - * from unpriced calls". - */ -export function sumModelCallCostUsd(attempts: readonly ModelCallAttempt[]): { - costUsd: number; - coverage: ModelCallCoverage; -} { - const unique = dedupeModelCallAttempts(attempts); - let costUsd = 0; - for (const attempt of unique) { - if (attempt.costBasis === 'priced' && attempt.costUsd !== undefined) costUsd += attempt.costUsd; - } - return { costUsd, coverage: summarizeModelCallCoverage(unique) }; -} diff --git a/packages/core/src/model-call-usage-projection.ts b/packages/core/src/model-call-usage-projection.ts index 3c394edbe8..f6cf170c2b 100644 --- a/packages/core/src/model-call-usage-projection.ts +++ b/packages/core/src/model-call-usage-projection.ts @@ -17,35 +17,23 @@ * under the License. */ -import { - dedupeModelCallAttempts, - summarizeModelCallCoverage, - type ModelCallAttempt, - type ModelCallCoverage, -} from './model-call-attempt.js'; -import { usageBucketKey } from './usage-stats/bucket-key.js'; -import type { - TimeRange, - UsageBucket, - UsageGroupBy, - UsageLogRow, - UsageQuery, - UsageSummaryV2, -} from './usage-stats/types.js'; +import type { ModelCallAttemptStatus, ModelCallCoverage } from './model-call-attempt.js'; +import type { TimeRange, UsageBucket, UsageLogRow, UsageSummaryV2 } from './usage-stats/types.js'; /** - * Usage projections over the canonical `ModelCallAttempt` ledger. + * What a Usage answer over the canonical model-call ledger looks like. * - * Pure: the caller supplies the records and owns their materialization. This is - * the aggregation the Usage surface reads once the read path moves off the - * per-send `LlmCallRecord` table. + * The aggregation itself belongs to the ledger, which holds these fields as + * columns and can sum them without materializing a workspace's history. What + * lives here is the vocabulary both Usage sources share: the shape of an + * answer, the window a query resolves to, and the rules a `SUM` has to mirror. * - * The one behavioural difference from that table is deliberate. `totalCostUsd` - * sums only records whose price was resolvable, and every result carries the - * {@link ModelCallCoverage} that qualifies it. The old schema had nowhere to say - * "this call cost something we could not price", so it wrote zero — making - * unpriced spend indistinguishable from a free call. A total presented without - * its coverage repeats that claim. + * One behavioural rule runs through all of it. `totalCostUsd` sums only records + * whose price was resolvable, and every result carries the + * {@link ModelCallCoverage} that qualifies it. The frozen pre-cutover table had + * nowhere to say "this call cost something we could not price", so it wrote + * zero — making unpriced spend indistinguishable from a free call. A total + * presented without its coverage repeats that claim. */ export interface ModelCallUsageSummary extends UsageSummaryV2 { /** Always present: the projection measures every attempt it counts. */ @@ -53,6 +41,11 @@ export interface ModelCallUsageSummary extends UsageSummaryV2 { coverage: ModelCallCoverage; } +export interface ModelCallUsageBuckets { + buckets: UsageBucket[]; + coverage: ModelCallCoverage; +} + export interface ModelCallUsageLogs { rows: UsageLogRow[]; total: number; @@ -79,228 +72,13 @@ export function resolveUsageRange(range: TimeRange, now: number): { from: number * would inflate the error rate with user cancellations. */ export function usageStatusForAttempt( - status: ModelCallAttempt['status'], + status: ModelCallAttemptStatus, ): 'success' | 'error' | 'aborted' { if (status === 'completed') return 'success'; if (status === 'failed') return 'error'; return 'aborted'; } -function matchesQuery( - attempt: ModelCallAttempt, - query: UsageQuery, - range: { from: number; to: number }, -): boolean { - if (attempt.completedAt < range.from || attempt.completedAt > range.to) return false; - if (query.sessionId !== undefined && attempt.sessionId !== query.sessionId) return false; - if (query.providerId !== undefined && attempt.providerId !== query.providerId) return false; - if (query.modelId !== undefined && attempt.modelId !== query.modelId) return false; - if (query.connectionSlug !== undefined && attempt.connectionSlug !== query.connectionSlug) { - return false; - } - if (query.status !== undefined && query.status !== 'all') { - if (usageStatusForAttempt(attempt.status) !== query.status) return false; - } - return true; -} - -/** - * Selects the attempts a query addresses, in append order, deduped by - * `attemptId` so a re-appended settlement counts once. - */ -export function selectModelCallAttempts( - attempts: readonly ModelCallAttempt[], - query: UsageQuery, - now: number, -): { rows: ModelCallAttempt[]; range: { from: number; to: number } } { - const range = resolveUsageRange(query.range, now); - const rows = dedupeModelCallAttempts(attempts).filter((a) => matchesQuery(a, query, range)); - return { rows, range }; -} - -function tokens(attempt: ModelCallAttempt): { - input: number; - output: number; - cacheMiss: number; - cacheRead: number; - cacheWrite: number; - reasoning: number; - total: number; -} { - const reportedCacheRead = attempt.cacheReadInputTokens ?? 0; - const input = attempt.inputTokens ?? 0; - const output = attempt.outputTokens ?? 0; - const cacheMiss = attempt.cacheMissInputTokens ?? 0; - // With no reported prompt total there is no denominator to validate against, - // but the provider's cache evidence remains authoritative. Presentation code - // must leave ratios unavailable while Usage coverage is partial. - const cacheRead = - attempt.inputTokens === undefined - ? reportedCacheRead - : clampCacheReadTokens(input, reportedCacheRead); - const cacheWrite = attempt.cacheWriteInputTokens ?? 0; - const reasoning = attempt.reasoningTokens ?? 0; - return { input, output, cacheMiss, cacheRead, cacheWrite, reasoning, total: input + output }; -} - export function clampCacheReadTokens(inputTokens: number, cacheReadTokens: number): number { return Math.min(cacheReadTokens, inputTokens); } - -/** Cost contributed by an attempt. Unpriced records contribute nothing to the - * sum and are surfaced through coverage instead of being counted as zero. */ -function pricedCost(attempt: ModelCallAttempt): number { - return attempt.costBasis === 'priced' ? (attempt.costUsd ?? 0) : 0; -} - -export function projectModelCallUsageSummary( - attempts: readonly ModelCallAttempt[], - query: UsageQuery, - now: number, -): ModelCallUsageSummary { - const { rows, range } = selectModelCallAttempts(attempts, query, now); - const totals = { - input: 0, - output: 0, - cacheMiss: 0, - cacheRead: 0, - cacheWrite: 0, - reasoning: 0, - total: 0, - }; - let totalCostUsd = 0; - let totalDurationMs = 0; - let cacheHitRequests = 0; - let cacheCreateRequests = 0; - let errorRequests = 0; - for (const attempt of rows) { - const t = tokens(attempt); - totals.input += t.input; - totals.output += t.output; - totals.cacheMiss += t.cacheMiss; - totals.cacheRead += t.cacheRead; - totals.cacheWrite += t.cacheWrite; - totals.reasoning += t.reasoning; - totals.total += t.total; - totalCostUsd += pricedCost(attempt); - totalDurationMs += attempt.latencyMs; - if (t.cacheRead > 0) cacheHitRequests += 1; - if (t.cacheWrite > 0) cacheCreateRequests += 1; - if (usageStatusForAttempt(attempt.status) === 'error') errorRequests += 1; - } - return { - range, - totalRequests: rows.length, - totalCostUsd, - totalDurationMs, - totalTokens: totals, - cacheHitRequests, - cacheCreateRequests, - errorRequests, - coverage: summarizeModelCallCoverage(rows), - }; -} - -export function projectModelCallUsageBuckets( - attempts: readonly ModelCallAttempt[], - query: UsageQuery, - groupBy: UsageGroupBy, - now: number, -): UsageBucket[] { - const { rows } = selectModelCallAttempts(attempts, query, now); - const groups = new Map(); - for (const attempt of rows) { - const key = usageBucketKey( - { providerId: attempt.providerId, modelId: attempt.modelId, ts: attempt.completedAt }, - groupBy, - ); - const group = groups.get(key); - if (group) group.push(attempt); - else groups.set(key, [attempt]); - } - return [...groups.entries()] - .map(([key, group]) => { - const agg = { - input: 0, - output: 0, - cacheMiss: 0, - cacheRead: 0, - cacheWrite: 0, - reasoning: 0, - total: 0, - }; - let costUsd = 0; - let latency = 0; - let errors = 0; - for (const attempt of group) { - const t = tokens(attempt); - agg.input += t.input; - agg.output += t.output; - agg.cacheMiss += t.cacheMiss; - agg.cacheRead += t.cacheRead; - agg.cacheWrite += t.cacheWrite; - agg.reasoning += t.reasoning; - agg.total += t.total; - costUsd += pricedCost(attempt); - latency += attempt.latencyMs; - if (usageStatusForAttempt(attempt.status) === 'error') errors += 1; - } - return { - key, - label: key, - requests: group.length, - inputTokens: agg.input, - outputTokens: agg.output, - cacheMissTokens: agg.cacheMiss, - cacheReadTokens: agg.cacheRead, - cacheWriteTokens: agg.cacheWrite, - reasoningTokens: agg.reasoning, - totalTokens: agg.total, - costUsd, - avgLatencyMs: group.length === 0 ? 0 : latency / group.length, - errorRate: group.length === 0 ? 0 : errors / group.length, - } satisfies UsageBucket; - }) - .sort((left, right) => right.requests - left.requests); -} - -export function projectModelCallUsageLogs( - attempts: readonly ModelCallAttempt[], - query: UsageQuery, - now: number, - offset = 0, - limit = 100, -): ModelCallUsageLogs { - const { rows } = selectModelCallAttempts(attempts, query, now); - const ordered = [...rows].sort((left, right) => right.completedAt - left.completedAt); - const page = ordered.slice(offset, offset + limit).map((attempt) => { - const t = tokens(attempt); - return { - id: attempt.attemptId, - ts: attempt.completedAt, - callKind: attempt.callKind, - callId: attempt.logicalCallId, - ...(attempt.connectionSlug !== undefined ? { connectionSlug: attempt.connectionSlug } : {}), - providerId: attempt.providerId, - modelId: attempt.modelId, - inputTokens: t.input, - outputTokens: t.output, - cacheMissTokens: t.cacheMiss, - cacheReadTokens: t.cacheRead, - cacheWriteTokens: t.cacheWrite, - reasoningTokens: t.reasoning, - totalTokens: t.total, - // A row keeps its basis, not just its number. Collapsing an unpriced call - // to 0 here would reproduce, per row, exactly the ambiguity the coverage - // breakdown removes from the totals. - ...(attempt.costBasis === 'priced' ? { costUsd: attempt.costUsd ?? 0 } : {}), - costBasis: attempt.costBasis, - latencyMs: attempt.latencyMs, - status: usageStatusForAttempt(attempt.status), - ...(attempt.errorClass !== undefined ? { errorClass: attempt.errorClass } : {}), - sessionId: attempt.sessionId, - turnId: attempt.turnId, - } satisfies UsageLogRow; - }); - return { rows: page, total: ordered.length, coverage: summarizeModelCallCoverage(rows) }; -} diff --git a/packages/core/src/record-schema.ts b/packages/core/src/record-schema.ts index a0489d5e19..c451ab52b3 100644 --- a/packages/core/src/record-schema.ts +++ b/packages/core/src/record-schema.ts @@ -68,6 +68,19 @@ export function hasExactShape(value: Record, shape: ExactObject ); } +/** + * Narrows a record to the keys a shape allows. `undefined` entries are dropped + * so the result serializes the way {@link hasExactShape} reads it back. + */ +export function pickShape(value: T, shape: ExactObjectShape): T { + const picked: Record = {}; + for (const key of shape.allowed) { + const entry = (value as Record)[key]; + if (entry !== undefined) picked[key] = entry; + } + return picked as T; +} + export function isFiniteNumber(value: unknown): value is number { return typeof value === 'number' && Number.isFinite(value); } diff --git a/packages/core/src/session-trace.ts b/packages/core/src/session-trace.ts index f3f67160c6..71a624d810 100644 --- a/packages/core/src/session-trace.ts +++ b/packages/core/src/session-trace.ts @@ -330,7 +330,11 @@ const MODEL_CALL_STEP_SHAPE = defineObjectShape()( ], ['connectionSlug', 'historyCompactRoute', 'costUsd'], ); -const MODEL_ATTEMPT_SHAPE = defineObjectShape()( +/** + * The Inspector's own view of an attempt. Exported so the projection narrows an + * authority record through this list rather than restating it by hand. + */ +export const MODEL_ATTEMPT_SHAPE = defineObjectShape()( [ 'attemptId', 'attempt', diff --git a/packages/core/src/usage-ledger-merge.ts b/packages/core/src/usage-ledger-merge.ts index f176ea0e5f..7bf7e007fe 100644 --- a/packages/core/src/usage-ledger-merge.ts +++ b/packages/core/src/usage-ledger-merge.ts @@ -17,19 +17,13 @@ * under the License. */ -import type { ModelCallAttempt, ModelCallCoverage } from './model-call-attempt.js'; -import { - projectModelCallUsageBuckets, - projectModelCallUsageLogs, - projectModelCallUsageSummary, -} from './model-call-usage-projection.js'; +import type { ModelCallCoverage } from './model-call-attempt.js'; import type { - UsageBucket, - UsageGroupBy, - UsageLogRow, - UsageQuery, - UsageSummaryV2, -} from './usage-stats/types.js'; + ModelCallUsageBuckets, + ModelCallUsageLogs, + ModelCallUsageSummary, +} from './model-call-usage-projection.js'; +import type { UsageBucket, UsageLogRow, UsageSummaryV2 } from './usage-stats/types.js'; /** * Merges the canonical `ModelCallAttempt` ledger with the frozen pre-cutover @@ -141,19 +135,22 @@ export interface MergedUsageLogs { provenance: UsageProvenance; } -export interface CanonicalUsageSource { - attempts: readonly ModelCallAttempt[]; +/** + * One canonical answer, already aggregated by the ledger, with what qualifies + * it: rows in the window whose pricing was lost, and runs the projection has + * not folded in yet. + */ +export interface CanonicalUsageSource { + projection: T; unreadableRecords: number; pendingRepairs: number; } export function mergeUsageSummary( legacy: UsageSummaryV2, - canonical: CanonicalUsageSource, - query: UsageQuery, - now: number, + canonical: CanonicalUsageSource, ): MergedUsageSummary { - const projected = projectModelCallUsageSummary(canonical.attempts, query, now); + const projected = canonical.projection; return { range: projected.range, totalRequests: legacy.totalRequests + projected.totalRequests, @@ -184,21 +181,17 @@ export function mergeUsageSummary( export function mergeUsageBuckets( legacy: readonly UsageBucket[], - canonical: CanonicalUsageSource, - query: UsageQuery, - groupBy: UsageGroupBy, - now: number, + canonical: CanonicalUsageSource, ): MergedUsageBuckets { - const projected = projectModelCallUsageBuckets(canonical.attempts, query, groupBy, now); const merged = new Map(); - for (const bucket of [...legacy, ...projected]) { + for (const bucket of [...legacy, ...canonical.projection.buckets]) { const existing = merged.get(bucket.key); merged.set(bucket.key, existing ? combineBuckets(existing, bucket) : { ...bucket }); } return { buckets: [...merged.values()].sort((left, right) => right.requests - left.requests), provenance: { - coverage: projectModelCallUsageSummary(canonical.attempts, query, now).coverage, + coverage: canonical.projection.coverage, legacyRecords: legacy.reduce((total, bucket) => total + bucket.requests, 0), unreadableRecords: canonical.unreadableRecords, pendingRepairs: canonical.pendingRepairs, @@ -213,13 +206,11 @@ export function mergeUsageBuckets( */ export function mergeUsageLogs( legacy: { rows: readonly UsageLogRow[]; total: number }, - canonical: CanonicalUsageSource, - query: UsageQuery, - now: number, + canonical: CanonicalUsageSource, offset: number, limit: number, ): MergedUsageLogs { - const projected = projectModelCallUsageLogs(canonical.attempts, query, now, 0, offset + limit); + const projected = canonical.projection; const rows: UsageLogRow[] = []; let left = 0; let right = 0; diff --git a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts index 9b18f5b261..bfae661190 100644 --- a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts @@ -2025,9 +2025,10 @@ test('production Host executes a canonical ai-sdk Session against a real provide assert.equal(compactUsage.inputTokens, 7); assert.equal(compactUsage.outputTokens, 3); const capturedRequestCount = mainRequests.length + compactRequests.length; - const attempts = await waitForCanonicalAttempts(usageStores, session.id, capturedRequestCount); - assert.equal(attempts.length, capturedRequestCount); - assert.ok(attempts.every((attempt) => attempt.promptComposition)); + assert.equal( + await waitForCanonicalRequests(usageStores, session.id, capturedRequestCount), + capturedRequestCount, + ); const contextDiagnostics = await composition.handlers['context.diagnostics.query']( { sessionId: session.id }, connectionContext, @@ -3856,28 +3857,23 @@ async function waitForUsage( throw new Error('Hosted real-model usage attribution was not persisted'); } -async function waitForCanonicalAttempts( +async function waitForCanonicalRequests( usage: InteractiveUsageStoresWriter, sessionId: string, expectedRequests: number, -): Promise { +): Promise { + const ask = () => usage.modelCalls.modelCallSummary({ range: 'all', sessionId }, Date.now()); for (let attempt = 0; attempt < 100; attempt += 1) { - const page = await usage.modelCalls.modelCallAttempts( - { from: 0, to: Number.MAX_SAFE_INTEGER }, - sessionId, - ); - if (page.attempts.length >= expectedRequests) return page.attempts; + const { projection } = await ask(); + if (projection.totalRequests >= expectedRequests) return projection.totalRequests; await new Promise((resolve) => setTimeout(resolve, 10)); } - const page = await usage.modelCalls.modelCallAttempts( - { from: 0, to: Number.MAX_SAFE_INTEGER }, - sessionId, - ); + const { projection, unreadableRecords } = await ask(); throw new Error( `Hosted canonical model-call attempts were not persisted: ${JSON.stringify({ expectedRequests, - attempts: page.attempts.length, - unreadableRecords: page.unreadableRecords, + totalRequests: projection.totalRequests, + unreadableRecords, })}`, ); } diff --git a/packages/runtime-host/src/server/canonical-usage-reader.ts b/packages/runtime-host/src/server/canonical-usage-reader.ts index 4f0c315999..9d7d5a1d9a 100644 --- a/packages/runtime-host/src/server/canonical-usage-reader.ts +++ b/packages/runtime-host/src/server/canonical-usage-reader.ts @@ -17,10 +17,17 @@ * under the License. */ -import { resolveUsageRange } from '@maka/core/model-call-usage-projection'; -import type { UsageQuery } from '@maka/core/usage-stats/types'; +import type { + ModelCallUsageBuckets, + ModelCallUsageLogs, + ModelCallUsageSummary, +} from '@maka/core/model-call-usage-projection'; +import type { UsageGroupBy, UsageQuery } from '@maka/core/usage-stats/types'; import type { CanonicalUsageSource } from '@maka/core/usage-ledger-merge'; -import type { InteractiveUsageStoresWriter } from '@maka/storage/usage-stores'; +import type { + InteractiveUsageStoresWriter, + ModelCallLedgerResult, +} from '@maka/storage/usage-stores'; export class CanonicalUsageProjectionIncompleteError extends Error { constructor() { super('Canonical Usage projection is incomplete'); @@ -28,16 +35,19 @@ export class CanonicalUsageProjectionIncompleteError extends Error { } } -/** Reads and repairs the canonical usage source shared by Host-owned projections. */ -export async function readCanonicalUsage( +/** + * Repairs the projection, then asks the ledger for one answer. + * + * `catchUpModelCallProjection` is a write. Paged log reads call this once per + * page, so a caller that already repaired on its first page passes + * `repair: false` on later pages to avoid a redundant repair write per page. + */ +async function readCanonical( stores: InteractiveUsageStoresWriter, query: UsageQuery, - now: number, - repair = true, -): Promise { - // `catchUpModelCallProjection` is a write. Paged log reads call this once per - // page, so a caller that already repaired on its first page can pass - // `repair: false` on later pages to avoid a redundant repair write per page. + repair: boolean, + ask: () => Promise>, +): Promise> { const repairOutcome = repair ? await stores.modelCalls .catchUpModelCallProjection( @@ -45,24 +55,56 @@ export async function readCanonicalUsage( ) .catch(() => ({ pendingRuns: 1, unreadableEvents: 0 })) : { pendingRuns: 0, unreadableEvents: 0 }; - const page = await stores.modelCalls.modelCallAttempts( - resolveUsageRange(query.range, now), - query.sessionId, - ); + const answer = await ask(); return { - attempts: page.attempts, - unreadableRecords: page.unreadableRecords + repairOutcome.unreadableEvents, + projection: answer.projection, + unreadableRecords: answer.unreadableRecords + repairOutcome.unreadableEvents, pendingRepairs: repairOutcome.pendingRuns, }; } +export function readCanonicalUsageSummary( + stores: InteractiveUsageStoresWriter, + query: UsageQuery, + now: number, + repair = true, +): Promise> { + return readCanonical(stores, query, repair, () => stores.modelCalls.modelCallSummary(query, now)); +} + +export function readCanonicalUsageBuckets( + stores: InteractiveUsageStoresWriter, + query: UsageQuery, + groupBy: UsageGroupBy, + now: number, + repair = true, +): Promise> { + return readCanonical(stores, query, repair, () => + stores.modelCalls.modelCallBuckets(query, groupBy, now), + ); +} + +export function readCanonicalUsageLogs( + stores: InteractiveUsageStoresWriter, + query: UsageQuery, + now: number, + limit: number, + repair = true, +): Promise> { + // Both sources are newest-first, so a merged page can only be drawn from each + // source's own first `offset + limit` rows. + return readCanonical(stores, query, repair, () => + stores.modelCalls.modelCallLogs(query, now, 0, limit), + ); +} + /** Runs one bounded repair pass and rejects data still unsafe for durable derivatives. */ -export async function readCompleteCanonicalUsage( +export async function readCompleteCanonicalUsageSummary( stores: InteractiveUsageStoresWriter, query: UsageQuery, now: number, -): Promise { - const source = await readCanonicalUsage(stores, query, now); +): Promise> { + const source = await readCanonicalUsageSummary(stores, query, now); if (source.unreadableRecords > 0 || source.pendingRepairs > 0) { throw new CanonicalUsageProjectionIncompleteError(); } diff --git a/packages/runtime-host/src/server/daily-review-coordinator.ts b/packages/runtime-host/src/server/daily-review-coordinator.ts index 60f0564664..2d62df383f 100644 --- a/packages/runtime-host/src/server/daily-review-coordinator.ts +++ b/packages/runtime-host/src/server/daily-review-coordinator.ts @@ -54,7 +54,8 @@ import type { HostDailyReviewModel } from './execution-model-authority.js'; import type { RuntimeHostResidency } from './host-kernel.js'; import { CanonicalUsageProjectionIncompleteError, - readCompleteCanonicalUsage, + readCanonicalUsageBuckets, + readCompleteCanonicalUsageSummary, } from './canonical-usage-reader.js'; const ARCHIVE_LIMIT = 180; @@ -253,7 +254,16 @@ export class HostDailyReviewCoordinator { async #buildSummary(range: DayRangeMs, now: number): Promise { const query = dailyUsageQuery(range); - const canonical = await readCompleteCanonicalUsage(this.#usage, query, now); + // The summary read repairs the projection and refuses an incomplete one; the + // bucket read that follows reuses that pass rather than repairing again. + const canonical = await readCompleteCanonicalUsageSummary(this.#usage, query, now); + const canonicalModels = await readCanonicalUsageBuckets( + this.#usage, + query, + 'model', + now, + false, + ); const [usageSummary, toolBuckets, modelBuckets, sessions] = await Promise.all([ this.#usage.telemetry.summary(query), this.#usage.telemetry.buckets(query, 'tool'), @@ -262,7 +272,7 @@ export class HostDailyReviewCoordinator { ]); return buildDailyReviewSummary({ day: range, - usageSummary: mergeUsageSummary(usageSummary, canonical, query, now), + usageSummary: mergeUsageSummary(usageSummary, canonical), sessions: pickDailyReviewSessions( collapseSessionRevisions(sessions), range, @@ -270,7 +280,7 @@ export class HostDailyReviewCoordinator { ), topTools: pickDailyReviewTopEntries(toolBuckets, DAILY_REVIEW_LIST_LIMIT), topModels: pickDailyReviewTopEntries( - mergeUsageBuckets(modelBuckets, canonical, query, 'model', now).buckets, + mergeUsageBuckets(modelBuckets, canonicalModels).buckets, DAILY_REVIEW_LIST_LIMIT, ), }); diff --git a/packages/runtime-host/src/server/usage-pricing-coordinator.ts b/packages/runtime-host/src/server/usage-pricing-coordinator.ts index ab53dd0e3b..149546a24d 100644 --- a/packages/runtime-host/src/server/usage-pricing-coordinator.ts +++ b/packages/runtime-host/src/server/usage-pricing-coordinator.ts @@ -62,7 +62,11 @@ import { } from '../protocol/index.js'; import type { UsagePricingOperationHandlerMap } from './operation-dispatcher.js'; import { RuntimePolicyActivationGate } from './runtime-policy-activation-gate.js'; -import { readCanonicalUsage } from './canonical-usage-reader.js'; +import { + readCanonicalUsageBuckets, + readCanonicalUsageLogs, + readCanonicalUsageSummary, +} from './canonical-usage-reader.js'; /** Root-scoped projection over the authentic lease-bound usage stores. */ export class HostUsagePricingCoordinator { @@ -125,27 +129,13 @@ export class HostUsagePricingCoordinator { return titles; } - /** - * Reads the canonical ledger for the window a query addresses (#1679). The - * range is resolved once here so both sources answer the same window. - */ - async #canonicalUsage( - query: UsageQuery, - now: number, - repair = true, - ): Promise { - return readCanonicalUsage(this.#stores, query, now, repair); - } - async #queryUsage(input: UsageQueryInput): Promise> { try { const now = Date.now(); if (input.kind === 'summary') { const merged = mergeUsageSummary( await this.#stores.telemetry.summary(input.query), - await this.#canonicalUsage(input.query, now), - input.query, - now, + await readCanonicalUsageSummary(this.#stores, input.query, now), ); // Tool executions are in their own ledger, not the model-call one, so // their totals ride beside the merged summary rather than inside it — @@ -180,10 +170,13 @@ export class HostUsagePricingCoordinator { : mergeUsageBuckets( legacy, // Only the first page repairs; later pages reuse it. - await this.#canonicalUsage(input.query, now, offset === 0), - input.query, - input.groupBy, - now, + await readCanonicalUsageBuckets( + this.#stores, + input.query, + input.groupBy, + now, + offset === 0, + ), ); if (offset > merged.buckets.length) return invalidUsageOffset(); return { @@ -225,9 +218,7 @@ export class HostUsagePricingCoordinator { const merged = mergeUsageLogs( legacy, // Only the first page repairs; later pages reuse it. - await this.#canonicalUsage(input.query, now, offset === 0), - input.query, - now, + await readCanonicalUsageLogs(this.#stores, input.query, now, offset + limit, offset === 0), offset, limit, ); diff --git a/packages/runtime/src/session-trace-projection.ts b/packages/runtime/src/session-trace-projection.ts index e7c68c3897..62fac82ad0 100644 --- a/packages/runtime/src/session-trace-projection.ts +++ b/packages/runtime/src/session-trace-projection.ts @@ -25,6 +25,7 @@ import { import { TERMINAL_RUNTIME_EVENT_STATUSES, type RuntimeEvent } from '@maka/core/runtime-event'; import { SESSION_TRACE_SCHEMA_VERSION, + MODEL_ATTEMPT_SHAPE, traceTurnIdentityKey, type SessionTrace, type SessionTraceCoverage, @@ -35,6 +36,7 @@ import { type TraceStep, type TurnTrace, } from '@maka/core/session-trace'; +import { pickShape } from '@maka/core/record-schema'; /** * Builds the per-session causal trace the Inspector renders (#1625). @@ -270,35 +272,7 @@ function projectModelCallSteps(attempts: readonly ModelCallAttempt[]): TraceMode } function toTraceAttempt(attempt: ModelCallAttempt): TraceModelAttempt { - return { - attemptId: attempt.attemptId, - attempt: attempt.attempt, - status: attempt.status, - startedAt: attempt.startedAt, - completedAt: attempt.completedAt, - latencyMs: attempt.latencyMs, - ...(attempt.timeToFirstTokenMs !== undefined - ? { timeToFirstTokenMs: attempt.timeToFirstTokenMs } - : {}), - ...(attempt.finishReason !== undefined ? { finishReason: attempt.finishReason } : {}), - ...(attempt.errorClass !== undefined ? { errorClass: attempt.errorClass } : {}), - ...(attempt.httpStatus !== undefined ? { httpStatus: attempt.httpStatus } : {}), - ...(attempt.providerCode !== undefined ? { providerCode: attempt.providerCode } : {}), - ...(attempt.providerRequestId !== undefined - ? { providerRequestId: attempt.providerRequestId } - : {}), - ...(attempt.retryable !== undefined ? { retryable: attempt.retryable } : {}), - ...(attempt.inputTokens !== undefined ? { inputTokens: attempt.inputTokens } : {}), - ...(attempt.outputTokens !== undefined ? { outputTokens: attempt.outputTokens } : {}), - ...(attempt.cacheReadInputTokens !== undefined - ? { cacheReadInputTokens: attempt.cacheReadInputTokens } - : {}), - ...(attempt.reasoningTokens !== undefined ? { reasoningTokens: attempt.reasoningTokens } : {}), - ...(attempt.contextWindow !== undefined ? { contextWindow: attempt.contextWindow } : {}), - ...(attempt.costUsd !== undefined ? { costUsd: attempt.costUsd } : {}), - costBasis: attempt.costBasis, - usageBasis: attempt.usageBasis, - }; + return pickShape(attempt, MODEL_ATTEMPT_SHAPE); } /** Prefix the runtime gives a written history-compaction boundary. */ diff --git a/packages/storage/src/__tests__/fixtures/model-call-attempt.ts b/packages/storage/src/__tests__/fixtures/model-call-attempt.ts new file mode 100644 index 0000000000..0f5876a8f9 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/model-call-attempt.ts @@ -0,0 +1,171 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { + MODEL_CALL_ATTEMPT_EVENT_TYPE, + MODEL_CALL_ATTEMPT_SCHEMA_VERSION, + type ModelCallAttempt, +} from '@maka/core/model-call-attempt'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { createSqliteModelCallLedger, type ModelCallLedger } from '../../model-call-ledger.js'; +import { acquireOperationalStateDatabase } from '../../operational-state-store.js'; + +/** A realistic epoch-ms clock: a small value pushes "40 days ago" below zero. */ +export const MODEL_CALL_NOW = 1_750_000_000_000; + +export function modelCallAttempt(overrides: Partial = {}): ModelCallAttempt { + return { + schemaVersion: MODEL_CALL_ATTEMPT_SCHEMA_VERSION, + logicalCallId: 'call-1', + attemptId: 'attempt-1', + traceId: 'trace-1', + sessionId: 'session-1', + runId: 'run-1', + turnId: 'turn-1', + step: 0, + attempt: 0, + callKind: 'main', + providerId: 'anthropic', + modelId: 'claude-opus-5', + startedAt: MODEL_CALL_NOW - 1_000, + completedAt: MODEL_CALL_NOW - 500, + latencyMs: 500, + status: 'completed', + usageBasis: 'reported', + inputTokens: 100, + outputTokens: 20, + costBasis: 'priced', + costUsd: 0.004, + ...overrides, + }; +} + +/** An attempt carrying the request evidence and diagnostics the ledger drops. */ +export function wideModelCallAttempt(overrides: Partial = {}): ModelCallAttempt { + return modelCallAttempt({ + promptComposition: { segments: [{ kind: 'messages', bytes: 4_096 }] }, + requestObservation: { + schemaVersion: 1, + digest: `sha256:${'a'.repeat(64)}`, + bytes: 27_817, + segments: [ + { + kind: 'tool_schema', + index: 0, + cacheable: true, + comparison: 'exact', + digest: `sha256:${'0'.repeat(64)}`, + bytes: 434, + label: 'tool-0', + }, + ], + }, + providerRequestId: 'req-1', + httpStatus: 200, + pricingRevision: 3, + ...overrides, + }); +} + +export async function withLedger( + run: (ledger: ModelCallLedger, root: string) => Promise, +): Promise { + const root = await mkdtemp(join(tmpdir(), 'maka-model-call-ledger-')); + const ledger = createSqliteModelCallLedger(root); + try { + await run(ledger, root); + } finally { + await ledger.close().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + } +} + +/** + * Commits one attempt to the AgentRun authority the projection reads from. + * + * Tests seed through the authority rather than the ledger's table on purpose: + * nothing in production writes a row any other way. + */ +export function appendAuthorityEvent( + root: string, + sequence: number, + value: ModelCallAttempt | { readonly schemaVersion: number }, + sessionId = 'session-1', + runId = 'run-1', +): void { + const lease = acquireOperationalStateDatabase(root); + try { + lease.transaction('write', () => { + lease.database + .prepare(` + INSERT OR IGNORE INTO core_agent_runs(session_id, run_id, created_at) + VALUES (?, ?, ?) + `) + .run(sessionId, runId, MODEL_CALL_NOW - 1_000); + lease.database + .prepare(` + INSERT INTO core_agent_run_events( + session_id, run_id, sequence, event_id, event_type, event_ts, record_json + ) VALUES (?, ?, ?, ?, ?, ?, ?) + `) + .run( + sessionId, + runId, + sequence, + `event-${sessionId}-${runId}-${sequence}`, + MODEL_CALL_ATTEMPT_EVENT_TYPE, + MODEL_CALL_NOW - 500 + sequence, + JSON.stringify({ + id: `event-${sessionId}-${runId}-${sequence}`, + type: MODEL_CALL_ATTEMPT_EVENT_TYPE, + ts: MODEL_CALL_NOW - 500 + sequence, + sessionId, + runId, + turnId: 'turn-1', + data: value, + }), + ); + lease.database + .prepare(` + UPDATE core_agent_runs + SET latest_model_call_sequence = ? + WHERE session_id = ? AND run_id = ? + `) + .run(sequence, sessionId, runId); + }); + } finally { + lease.close(); + } +} + +/** Projects a whole set of attempts and returns the ledger holding them. */ +export async function withProjectedAttempts( + attempts: readonly ModelCallAttempt[], + run: (ledger: ModelCallLedger, root: string) => Promise, +): Promise { + await withLedger(async (ledger, root) => { + attempts.forEach((value, index) => { + appendAuthorityEvent(root, index, value, value.sessionId, value.runId); + }); + await ledger.catchUpProjection(); + await run(ledger, root); + }); +} diff --git a/packages/storage/src/__tests__/model-call-ledger.test.ts b/packages/storage/src/__tests__/model-call-ledger.test.ts index 0fe026cf79..3d272ea7fa 100644 --- a/packages/storage/src/__tests__/model-call-ledger.test.ts +++ b/packages/storage/src/__tests__/model-call-ledger.test.ts @@ -25,7 +25,6 @@ import { DatabaseSync } from 'node:sqlite'; import { describe, test } from 'node:test'; import { MODEL_CALL_ATTEMPT_EVENT_TYPE, - MODEL_CALL_ATTEMPT_SCHEMA_VERSION, type ModelCallAttempt, } from '@maka/core/model-call-attempt'; import { @@ -33,97 +32,45 @@ import { ModelCallLedgerClosedError, ModelCallLedgerPublicationError, type ModelCallLedger, + type ModelCallLedgerReader, } from '../model-call-ledger.js'; import { acquireOperationalStateDatabase } from '../operational-state-store.js'; +import { MODEL_CALL_COLUMNS } from '../sqlite-usage-schema.js'; import { createSqliteAgentRunStore } from '../agent-run-store.js'; import { openInvocation } from './fixtures/invocation-opening.js'; - -const NOW = 1_750_000_000_000; - -function attempt(overrides: Partial = {}): ModelCallAttempt { - return { - schemaVersion: MODEL_CALL_ATTEMPT_SCHEMA_VERSION, - logicalCallId: 'call-1', - attemptId: 'attempt-1', - traceId: 'trace-1', - sessionId: 'session-1', - runId: 'run-1', - turnId: 'turn-1', - step: 0, - attempt: 0, - callKind: 'main', - providerId: 'anthropic', - modelId: 'claude-opus-5', - startedAt: NOW - 1_000, - completedAt: NOW - 500, - latencyMs: 500, - status: 'completed', - usageBasis: 'reported', - inputTokens: 100, - outputTokens: 20, - costBasis: 'priced', - costUsd: 0.004, - ...overrides, - }; +import { + appendAuthorityEvent, + modelCallAttempt as attempt, + MODEL_CALL_NOW as NOW, + withLedger, +} from './fixtures/model-call-attempt.js'; + +/** The calls a window holds, newest first, as the Usage log surface sees them. */ +function ids(ledger: ModelCallLedgerReader, from = 0, sessionId?: string): string[] { + return ledger + .logs({ range: { from, to: NOW }, ...(sessionId ? { sessionId } : {}) }, NOW, 0, 100) + .projection.rows.map((row) => row.id); } -async function withLedger(run: (ledger: ModelCallLedger, root: string) => Promise) { - const root = await mkdtemp(join(tmpdir(), 'maka-model-call-ledger-')); - const ledger = createSqliteModelCallLedger(root); - try { - await run(ledger, root); - } finally { - await ledger.close().catch(() => undefined); - await rm(root, { recursive: true, force: true }); - } +function unreadable(ledger: ModelCallLedgerReader, sessionId?: string): number { + return ledger.logs( + { range: { from: 0, to: NOW }, ...(sessionId ? { sessionId } : {}) }, + NOW, + 0, + 1, + ).unreadableRecords; } -function appendAuthorityEvent( - root: string, - sequence: number, - value: ModelCallAttempt | { readonly schemaVersion: number }, - sessionId = 'session-1', - runId = 'run-1', -): void { +/** Records one call whose pricing was lost before the ledger held columns. */ +function insertTombstone(root: string, attemptId: string, sessionId?: string): void { const lease = acquireOperationalStateDatabase(root); try { lease.transaction('write', () => { lease.database - .prepare(` - INSERT OR IGNORE INTO core_agent_runs(session_id, run_id, created_at) - VALUES (?, ?, ?) - `) - .run(sessionId, runId, NOW - 1_000); - lease.database - .prepare(` - INSERT INTO core_agent_run_events( - session_id, run_id, sequence, event_id, event_type, event_ts, record_json - ) VALUES (?, ?, ?, ?, ?, ?, ?) - `) - .run( - sessionId, - runId, - sequence, - `event-${sessionId}-${runId}-${sequence}`, - MODEL_CALL_ATTEMPT_EVENT_TYPE, - NOW - 500 + sequence, - JSON.stringify({ - id: `event-${sessionId}-${runId}-${sequence}`, - type: MODEL_CALL_ATTEMPT_EVENT_TYPE, - ts: NOW - 500 + sequence, - sessionId, - runId, - turnId: 'turn-1', - data: value, - }), - ); - lease.database - .prepare(` - UPDATE core_agent_runs - SET latest_model_call_sequence = ? - WHERE session_id = ? AND run_id = ? - `) - .run(sequence, sessionId, runId); + .prepare( + 'INSERT INTO usage_model_call_attempts(attempt_id, completed_at, session_id) VALUES (?, ?, ?)', + ) + .run(attemptId, NOW - 400, sessionId ?? null); }); } finally { lease.close(); @@ -141,16 +88,12 @@ describe('canonical model call ledger', () => { ); await ledger.catchUpProjection(); - const page = ledger.read({ from: NOW - 1_000, to: NOW }); - assert.deepEqual( - page.attempts.map((row) => row.attemptId), - ['inside'], - ); - assert.equal(page.unreadableRecords, 0); + assert.deepEqual(ids(ledger, NOW - 1_000), ['inside']); + assert.equal(unreadable(ledger), 0); }); }); - test('provider failure diagnostics survive closing and reopening the ledger', async () => { + test('a failed call keeps its pricing basis and drops what pricing cannot use', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-model-call-ledger-reopen-')); const first = createSqliteModelCallLedger(root); try { @@ -179,13 +122,30 @@ describe('canonical model call ledger', () => { const reopened = createSqliteModelCallLedger(root); try { - const restored = reopened.read({ from: 0, to: NOW }).attempts[0]; - assert.equal(restored?.historyCompactRoute, 'provider_native'); + const restored = reopened.logs({ range: { from: 0, to: NOW } }, NOW, 0, 10).projection + .rows[0]; + assert.equal(restored?.callKind, 'history_compact'); + assert.equal(restored?.status, 'error'); + assert.equal(restored?.costBasis, 'unpriced'); + assert.equal(Object.hasOwn(restored ?? {}, 'costUsd'), false); + // The Usage log row shows this one; the rest of the provider diagnostics + // have nowhere to land here and are answered from the AgentRun authority. assert.equal(restored?.errorClass, 'RequestRejected'); - assert.equal(restored?.httpStatus, 400); - assert.equal(restored?.providerCode, 'invalid_request_error'); - assert.equal(restored?.providerRequestId, 'req-reopen-1'); - assert.equal(restored?.retryable, false); + const lease = acquireOperationalStateDatabase(root); + try { + assert.deepEqual( + ( + lease.database + .prepare('PRAGMA table_info(usage_model_call_attempts)') + .all() as Array<{ + name: string; + }> + ).map((column) => column.name), + [...MODEL_CALL_COLUMNS], + ); + } finally { + lease.close(); + } } finally { await reopened.close(); } @@ -218,10 +178,7 @@ describe('canonical model call ledger', () => { const migrated = createSqliteModelCallLedger(root); try { await migrated.catchUpProjection(); - assert.deepEqual( - migrated.read({ from: 0, to: NOW }).attempts.map((row) => row.attemptId), - ['pre-checkpoint'], - ); + assert.deepEqual(ids(migrated), ['pre-checkpoint']); const lease = acquireOperationalStateDatabase(root); try { assert.equal( @@ -241,6 +198,48 @@ describe('canonical model call ledger', () => { } }); + test('a row converted in place keeps spend the authority can no longer replay', async () => { + // Deleting a Session drops its runs and cascades their events, but leaves + // its ledger rows. Converging those rows by wiping and re-projecting would + // erase that spend from the all-time totals, so they are converted in place. + const root = await mkdtemp(join(tmpdir(), 'maka-model-call-ledger-convert-')); + const first = createSqliteModelCallLedger(root); + await first.close(); + + const database = new DatabaseSync(join(root, 'runtime.sqlite')); + database.exec(` + PRAGMA foreign_keys = ON; + DROP TABLE usage_model_call_attempts; + CREATE TABLE usage_model_call_attempts ( + attempt_id TEXT PRIMARY KEY, + completed_at INTEGER NOT NULL, + record_json TEXT NOT NULL, + session_id TEXT + ); + UPDATE operational_schema_migrations SET version = 6 WHERE scope = 'usage'; + `); + database + .prepare('INSERT INTO usage_model_call_attempts VALUES (?, ?, ?, ?)') + .run( + 'deleted-session-call', + NOW - 500, + JSON.stringify(attempt({ attemptId: 'deleted-session-call' })), + 'session-1', + ); + database.close(); + + const migrated = createSqliteModelCallLedger(root); + try { + const page = migrated.logs({ range: { from: 0, to: NOW } }, NOW, 0, 10); + assert.equal(page.unreadableRecords, 0); + assert.equal(page.projection.rows[0]?.id, 'deleted-session-call'); + assert.equal(page.projection.rows[0]?.costUsd, 0.004); + } finally { + await migrated.close(); + await rm(root, { recursive: true, force: true }); + } + }); + test('a late settlement replaces the provisional record under the same attempt id', async () => { // The abort path records provisionally without usage; a `finish` arriving // afterwards settles the same attempt. Two rows would double-count it. @@ -258,10 +257,10 @@ describe('canonical model call ledger', () => { appendAuthorityEvent(root, 1, attempt({ status: 'completed', usageBasis: 'reported' })); await ledger.catchUpProjection(); - const page = ledger.read({ from: 0, to: NOW }); - assert.equal(page.attempts.length, 1); - assert.equal(page.attempts[0]?.status, 'completed'); - assert.equal(page.attempts[0]?.inputTokens, 100); + const rows = ledger.logs({ range: { from: 0, to: NOW } }, NOW, 0, 10).projection.rows; + assert.equal(rows.length, 1); + assert.equal(rows[0]?.status, 'success'); + assert.equal(rows[0]?.inputTokens, 100); }); }); @@ -270,7 +269,7 @@ describe('canonical model call ledger', () => { appendAuthorityEvent(root, 0, attempt({ costBasis: 'unpriced', costUsd: 0.004 })); const result = await ledger.catchUpProjection(); assert.equal(result.unreadableEvents, 1); - assert.equal(ledger.read({ from: 0, to: NOW }).attempts.length, 0); + assert.equal(ids(ledger).length, 0); }); }); @@ -278,25 +277,10 @@ describe('canonical model call ledger', () => { await withLedger(async (ledger, root) => { appendAuthorityEvent(root, 0, attempt({ attemptId: 'good' })); await ledger.catchUpProjection(); - const lease = acquireOperationalStateDatabase(root); - try { - lease.transaction('write', () => { - lease.database - .prepare( - 'INSERT INTO usage_model_call_attempts(attempt_id, completed_at, record_json) VALUES (?, ?, ?)', - ) - .run('corrupt', NOW - 400, '{"schemaVersion":1,'); - }); - } finally { - lease.close(); - } + insertTombstone(root, 'lost'); - const page = ledger.read({ from: 0, to: NOW }); - assert.deepEqual( - page.attempts.map((row) => row.attemptId), - ['good'], - ); - assert.equal(page.unreadableRecords, 1); + assert.deepEqual(ids(ledger), ['good']); + assert.equal(unreadable(ledger), 1); }); }); @@ -317,27 +301,12 @@ describe('canonical model call ledger', () => { 'run-b', ); await ledger.catchUpProjection(); - const lease = acquireOperationalStateDatabase(root); - try { - lease.transaction('write', () => { - lease.database - .prepare( - `INSERT INTO usage_model_call_attempts( - attempt_id, completed_at, record_json, session_id - ) VALUES (?, ?, ?, ?)`, - ) - .run('session-b-corrupt', NOW - 400, '{', 'session-b'); - }); - } finally { - lease.close(); - } + insertTombstone(root, 'session-b-lost', 'session-b'); - const page = ledger.read({ from: 0, to: NOW }, 'session-a'); - assert.deepEqual( - page.attempts.map((row) => row.attemptId), - ['session-a-call'], - ); - assert.equal(page.unreadableRecords, 0); + assert.deepEqual(ids(ledger, 0, 'session-a'), ['session-a-call']); + assert.equal(unreadable(ledger, 'session-a'), 0); + // The other Session's lost row is still reported to whoever asks for it. + assert.equal(unreadable(ledger, 'session-b'), 1); }); }); @@ -348,7 +317,7 @@ describe('canonical model call ledger', () => { await ledger.close(); await assert.rejects(() => ledger.catchUpProjection(), ModelCallLedgerClosedError); - assert.throws(() => ledger.read({ from: 0, to: NOW }), ModelCallLedgerClosedError); + assert.throws(() => ledger.summary({ range: 'all' }, NOW), ModelCallLedgerClosedError); await rm(root, { recursive: true, force: true }); }); }); @@ -371,10 +340,7 @@ describe('catching the read model up from the AgentRun authority', () => { await ledger.catchUpProjection({ sessionId: 'session-1' }); - assert.deepEqual( - ledger.read({ from: 0, to: NOW }).attempts.map((row) => row.attemptId), - ['real-append'], - ); + assert.deepEqual(ids(ledger), ['real-append']); }); }); @@ -389,10 +355,7 @@ describe('catching the read model up from the AgentRun authority', () => { pendingRuns: 0, unreadableEvents: 0, }); - assert.deepEqual( - ledger.read({ from: 0, to: NOW }).attempts.map((row) => row.attemptId), - ['missed'], - ); + assert.deepEqual(ids(ledger), ['missed']); }); }); @@ -405,13 +368,7 @@ describe('catching the read model up from the AgentRun authority', () => { const result = await ledger.catchUpProjection({ sessionId: 'session-1' }); assert.equal(result.pendingRuns, 0); - assert.deepEqual( - ledger - .read({ from: 0, to: NOW }) - .attempts.map((row) => row.attemptId) - .sort(), - ['new', 'old'], - ); + assert.deepEqual(ids(ledger).sort(), ['new', 'old']); }); }); @@ -427,10 +384,7 @@ describe('catching the read model up from the AgentRun authority', () => { assert.equal(first.pendingRuns, 0); assert.equal(second.unreadableEvents, 1); assert.deepEqual(second.changedSessionIds, []); - assert.deepEqual( - ledger.read({ from: 0, to: NOW }).attempts.map((row) => row.attemptId), - ['good'], - ); + assert.deepEqual(ids(ledger), ['good']); }); }); @@ -450,7 +404,7 @@ describe('catching the read model up from the AgentRun authority', () => { assert.equal(first.pendingRuns, 1); assert.equal(second.pendingRuns, 0); - assert.equal(ledger.read({ from: 0, to: NOW }).attempts.length, 2); + assert.equal(ids(ledger).length, 2); }); }); @@ -478,10 +432,7 @@ describe('catching the read model up from the AgentRun authority', () => { const recovered = await ledger.catchUpProjection(); assert.equal(recovered.pendingRuns, 0); - assert.deepEqual( - ledger.read({ from: 0, to: NOW }).attempts.map((row) => row.attemptId), - ['retry-after-storage-failure'], - ); + assert.deepEqual(ids(ledger), ['retry-after-storage-failure']); }); }); }); diff --git a/packages/storage/src/__tests__/model-call-usage-query.test.ts b/packages/storage/src/__tests__/model-call-usage-query.test.ts new file mode 100644 index 0000000000..421650c210 --- /dev/null +++ b/packages/storage/src/__tests__/model-call-usage-query.test.ts @@ -0,0 +1,295 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; +import { + modelCallAttempt as attempt, + MODEL_CALL_NOW as NOW, + withProjectedAttempts, +} from './fixtures/model-call-attempt.js'; + +const ALL = { range: 'all' } as const; + +describe('Usage answers over the canonical ledger', () => { + test('a total never counts unpriced spend as zero, and says so in coverage', async () => { + // The frozen pre-cutover table had nowhere to record "we could not price + // this", so it wrote 0 and unpriced spend looked free. The total here + // excludes it and the coverage reports it instead. + await withProjectedAttempts( + [ + attempt({ attemptId: 'a', costUsd: 0.004 }), + attempt({ attemptId: 'b', costBasis: 'unpriced', costUsd: undefined }), + ], + async (ledger) => { + const { projection } = ledger.summary(ALL, NOW); + assert.equal(Math.round(projection.totalCostUsd * 1000) / 1000, 0.004); + assert.equal(projection.totalRequests, 2); + assert.equal(projection.coverage.pricedAttempts, 1); + assert.equal(projection.coverage.unpricedAttempts, 1); + }, + ); + }); + + test('a genuinely free call is counted as priced, and reads apart from an unpriced one', async () => { + await withProjectedAttempts( + [ + attempt({ attemptId: 'free', logicalCallId: 'free', costUsd: 0 }), + attempt({ + attemptId: 'unknown', + logicalCallId: 'unknown', + costBasis: 'unpriced', + costUsd: undefined, + }), + ], + async (ledger) => { + const { projection } = ledger.summary(ALL, NOW); + assert.equal(projection.totalCostUsd, 0); + assert.equal(projection.coverage.pricedAttempts, 1); + assert.equal(projection.coverage.unpricedAttempts, 1); + + // The page-level coverage says how many rows were unpriced but not which + // ones, so a log row has to carry its own basis. + const rows = ledger.logs(ALL, NOW, 0, 10).projection.rows; + const free = rows.find((row) => row.id === 'free'); + const unknown = rows.find((row) => row.id === 'unknown'); + assert.equal(free?.costBasis, 'priced'); + assert.equal(free?.costUsd, 0); + assert.equal(unknown?.costBasis, 'unpriced'); + assert.equal(Object.hasOwn(unknown ?? {}, 'costUsd'), false); + }, + ); + }); + + test('usage-missing records are reported separately from unpriced ones', async () => { + await withProjectedAttempts( + [ + attempt({ + attemptId: 'no-usage', + status: 'failed', + usageBasis: 'missing', + inputTokens: undefined, + outputTokens: undefined, + costBasis: 'unpriced', + costUsd: undefined, + }), + ], + async (ledger) => { + const { projection } = ledger.summary(ALL, NOW); + assert.equal(projection.coverage.usageMissingAttempts, 1); + assert.equal(projection.coverage.unpricedAttempts, 1); + assert.equal(projection.totalTokens.total, 0); + }, + ); + }); + + test('one malformed cache reading cannot inflate the cache total', async () => { + await withProjectedAttempts( + [ + attempt({ attemptId: 'malformed-cache', inputTokens: 100, cacheReadInputTokens: 200 }), + attempt({ attemptId: 'cache-miss', inputTokens: 100, cacheReadInputTokens: 0 }), + ], + async (ledger) => { + const { projection } = ledger.summary(ALL, NOW); + assert.equal(projection.totalTokens.input, 200); + assert.equal(projection.totalTokens.cacheRead, 100); + }, + ); + }); + + test('provider cache-only evidence survives without inventing an input total', async () => { + await withProjectedAttempts( + [ + attempt({ + attemptId: 'cache-only', + usageBasis: 'partial', + inputTokens: undefined, + outputTokens: undefined, + cacheReadInputTokens: 10, + }), + ], + async (ledger) => { + const summary = ledger.summary(ALL, NOW).projection; + assert.equal(summary.totalTokens.input, 0); + assert.equal(summary.totalTokens.cacheRead, 10); + assert.equal(summary.cacheHitRequests, 1); + assert.equal(summary.coverage.usagePartialAttempts, 1); + + const bucket = ledger.buckets(ALL, 'provider', NOW).projection.buckets[0]; + assert.equal(bucket?.inputTokens, 0); + assert.equal(bucket?.cacheReadTokens, 10); + + const log = ledger.logs(ALL, NOW, 0, 10).projection.rows[0]; + assert.equal(log?.inputTokens, 0); + assert.equal(log?.cacheReadTokens, 10); + }, + ); + }); + + test('the summary sums recorded call time over the rows it counts', async () => { + await withProjectedAttempts( + [ + attempt({ attemptId: 'a', logicalCallId: 'a', latencyMs: 1_200 }), + attempt({ attemptId: 'b', logicalCallId: 'b', latencyMs: 300 }), + ], + async (ledger) => { + const { projection } = ledger.summary(ALL, NOW); + assert.equal(projection.totalDurationMs, 1_500); + assert.equal(projection.totalRequests, 2); + }, + ); + }); + + test('filters by Session, window, provider, model, and status', async () => { + await withProjectedAttempts( + [ + attempt({ attemptId: 'recent' }), + attempt({ + attemptId: 'old', + startedAt: NOW - 40 * 86_400_000 - 1, + completedAt: NOW - 40 * 86_400_000, + }), + attempt({ attemptId: 'other-provider', providerId: 'openai', modelId: 'gpt-x' }), + attempt({ attemptId: 'failed', status: 'failed' }), + attempt({ attemptId: 'other-session', sessionId: 'session-2', runId: 'run-2' }), + ], + async (ledger) => { + const requests = (query: Parameters[0]) => + ledger.summary(query, NOW).projection.totalRequests; + assert.equal(requests({ range: '24h' }), 4); + assert.equal(requests({ range: 'all', sessionId: 'session-1' }), 4); + assert.equal(requests({ range: 'all', providerId: 'openai' }), 1); + assert.equal(requests({ range: 'all', modelId: 'claude-opus-5' }), 4); + assert.equal(requests({ range: 'all', status: 'error' }), 1); + assert.equal(requests({ range: 'all', status: 'all' }), 5); + }, + ); + }); + + test('interrupted counts as aborted, not as an error', async () => { + // Collapsing a cut-short call into `error` would inflate the error rate + // with user cancellations. + await withProjectedAttempts( + [attempt({ attemptId: 'cut', status: 'interrupted' })], + async (ledger) => { + assert.equal(ledger.summary(ALL, NOW).projection.errorRequests, 0); + assert.equal( + ledger.summary({ range: 'all', status: 'aborted' }, NOW).projection.totalRequests, + 1, + ); + }, + ); + }); + + test('buckets group by provider and by model, excluding unpriced cost', async () => { + await withProjectedAttempts( + [ + attempt({ attemptId: 'a', costUsd: 0.004 }), + attempt({ attemptId: 'b', costUsd: 0.006 }), + attempt({ + attemptId: 'c', + providerId: 'openai', + modelId: 'gpt-x', + costBasis: 'unpriced', + costUsd: undefined, + }), + ], + async (ledger) => { + const byProvider = ledger.buckets(ALL, 'provider', NOW).projection.buckets; + assert.deepEqual( + byProvider.map((bucket) => [bucket.key, bucket.requests]), + [ + ['anthropic', 2], + ['openai', 1], + ], + ); + assert.equal(Math.round((byProvider[0]?.costUsd ?? 0) * 1000) / 1000, 0.01); + assert.equal(byProvider[1]?.costUsd, 0); + + const byModel = ledger.buckets(ALL, 'model', NOW).projection.buckets; + assert.equal(byModel.length, 2); + assert.ok(byModel.some((bucket) => bucket.key === 'anthropic:claude-opus-5')); + }, + ); + }); + + test('time buckets are named by the same key both Usage sources derive', async () => { + // SQLite decides only which rows group together; the key is still built by + // `usageBucketKey`. If the two disagreed about where a day starts, one day + // would silently split into two buckets rather than fail. + const midnight = Date.parse('2025-03-04T00:00:00.000Z'); + await withProjectedAttempts( + [ + attempt({ attemptId: 'first', startedAt: midnight - 1, completedAt: midnight }), + attempt({ attemptId: 'last', startedAt: midnight, completedAt: midnight + 86_399_999 }), + attempt({ attemptId: 'next-day', startedAt: midnight, completedAt: midnight + 86_400_000 }), + ], + async (ledger) => { + const byDay = ledger.buckets({ range: 'all' }, 'day', NOW).projection.buckets; + assert.deepEqual( + [...byDay] + .sort((left, right) => left.key.localeCompare(right.key)) + .map((b) => [b.key, b.requests]), + [ + ['2025-03-04', 2], + ['2025-03-05', 1], + ], + ); + }, + ); + }); + + test('logs page newest first and carry coverage for the whole match', async () => { + await withProjectedAttempts( + [ + attempt({ attemptId: 'older', startedAt: NOW - 3_500, completedAt: NOW - 3_000 }), + attempt({ attemptId: 'newer', startedAt: NOW - 1_500, completedAt: NOW - 1_000 }), + attempt({ + attemptId: 'unpriced', + startedAt: NOW - 2_500, + completedAt: NOW - 2_000, + costBasis: 'unpriced', + costUsd: undefined, + }), + ], + async (ledger) => { + const page = ledger.logs(ALL, NOW, 0, 2).projection; + assert.deepEqual( + page.rows.map((row) => row.id), + ['newer', 'unpriced'], + ); + assert.equal(page.total, 3); + // Coverage describes every matching record, not just the returned page. + assert.equal(page.coverage.attempts, 3); + assert.equal(page.coverage.unpricedAttempts, 1); + }, + ); + }); + + test('a replayed attemptId is one call, not two', async () => { + await withProjectedAttempts( + [attempt({ attemptId: 'dup' }), attempt({ attemptId: 'dup', costUsd: 0.004 })], + async (ledger) => { + const { projection } = ledger.summary(ALL, NOW); + assert.equal(projection.totalRequests, 1); + assert.equal(Math.round(projection.totalCostUsd * 1000) / 1000, 0.004); + }, + ); + }); +}); diff --git a/packages/storage/src/__tests__/sqlite-usage-schema.test.ts b/packages/storage/src/__tests__/sqlite-usage-schema.test.ts index bc644548a6..b392c8870d 100644 --- a/packages/storage/src/__tests__/sqlite-usage-schema.test.ts +++ b/packages/storage/src/__tests__/sqlite-usage-schema.test.ts @@ -21,6 +21,46 @@ import assert from 'node:assert/strict'; import { DatabaseSync } from 'node:sqlite'; import { test } from 'node:test'; import { migrateSqliteUsageDatabase } from '../sqlite-usage-schema.js'; +import { + MODEL_CALL_NOW as NOW, + modelCallAttempt as attempt, + wideModelCallAttempt as wideAttempt, +} from './fixtures/model-call-attempt.js'; +import { MODEL_CALL_COLUMNS } from '../sqlite-usage-schema.js'; + +/** A ledger as it stood before the record was spread into columns. */ +function blobLedger(database: DatabaseSync): void { + database.exec(` + CREATE TABLE usage_model_call_attempts ( + attempt_id TEXT PRIMARY KEY, + completed_at INTEGER NOT NULL, + record_json TEXT NOT NULL, + session_id TEXT + ); + `); +} + +function insertBlob( + database: DatabaseSync, + attemptId: string, + record: unknown, + sessionId?: string, +) { + database + .prepare('INSERT INTO usage_model_call_attempts VALUES (?, ?, ?, ?)') + .run( + attemptId, + NOW - 500, + typeof record === 'string' ? record : JSON.stringify(record), + sessionId ?? null, + ); +} + +function storedRow(database: DatabaseSync, attemptId: string): Record { + return database + .prepare('SELECT * FROM usage_model_call_attempts WHERE attempt_id = ?') + .get(attemptId) as Record; +} test('usage migration backfills Session identity for existing ledger rows', () => { const database = new DatabaseSync(':memory:'); @@ -74,3 +114,147 @@ test('usage migration backfills Session identity for existing ledger rows', () = database.close(); } }); + +test('the migration spreads a stored record into the columns a cost answer sums', () => { + const database = new DatabaseSync(':memory:'); + try { + blobLedger(database); + const wide = wideAttempt(); + insertBlob(database, wide.attemptId, wide, wide.sessionId); + + migrateSqliteUsageDatabase(database); + + const row = storedRow(database, wide.attemptId); + assert.deepEqual(Object.keys(row), [...MODEL_CALL_COLUMNS]); + // Every number a Usage total is built from reads the same after the spread. + assert.equal(row.cost_usd, 0.004); + assert.equal(row.cost_basis, 'priced'); + assert.equal(row.input_tokens, 100); + assert.equal(row.output_tokens, 20); + assert.equal(row.provider_id, 'anthropic'); + assert.equal(row.session_id, 'session-1'); + } finally { + database.close(); + } +}); + +test('a record the migration cannot read whole keeps its identity and loses the rest', () => { + // A row that ends up half-filled would make the table's own CHECK + // unsatisfiable and take the whole migration with it, so conversion is + // all-or-nothing per row. What is left says a call happened and its cost is + // gone — which is what a read reports as unreadable. + const database = new DatabaseSync(':memory:'); + try { + blobLedger(database); + insertBlob(database, 'damaged', '{"schemaVersion":1,', 'session-1'); + insertBlob(database, 'alien', { sessionId: 'session-2' }); + const priced = attempt({ attemptId: 'priced' }); + insertBlob(database, priced.attemptId, priced, priced.sessionId); + + migrateSqliteUsageDatabase(database); + + assert.deepEqual( + database + .prepare( + 'SELECT attempt_id, session_id FROM usage_model_call_attempts WHERE cost_basis IS NULL ORDER BY attempt_id', + ) + .all() + .map((row) => ({ ...row })), + [ + { attempt_id: 'alien', session_id: 'session-2' }, + { attempt_id: 'damaged', session_id: 'session-1' }, + ], + ); + assert.equal(storedRow(database, 'priced').cost_usd, 0.004); + } finally { + database.close(); + } +}); + +test('the migration is a no-op once the ledger already holds columns', () => { + const database = new DatabaseSync(':memory:'); + try { + blobLedger(database); + const wide = wideAttempt(); + insertBlob(database, wide.attemptId, wide, wide.sessionId); + migrateSqliteUsageDatabase(database); + const once = storedRow(database, wide.attemptId); + + migrateSqliteUsageDatabase(database); + + assert.deepEqual(storedRow(database, wide.attemptId), once); + } finally { + database.close(); + } +}); + +test('the migration converts every row, however many a workspace holds', () => { + const database = new DatabaseSync(':memory:'); + try { + blobLedger(database); + for (let index = 0; index < 1_200; index += 1) { + const row = attempt({ attemptId: `attempt-${String(index).padStart(5, '0')}` }); + insertBlob(database, row.attemptId, row, row.sessionId); + } + + migrateSqliteUsageDatabase(database); + + assert.equal( + database + .prepare('SELECT COUNT(*) AS count FROM usage_model_call_attempts WHERE cost_basis IS NULL') + .get()?.count, + 0, + ); + } finally { + database.close(); + } +}); + +test('the ledger refuses a row that would make a total dishonest', () => { + const database = new DatabaseSync(':memory:'); + try { + migrateSqliteUsageDatabase(database); + const insert = (values: Record) => { + const columns = Object.keys(values); + database + .prepare( + `INSERT INTO usage_model_call_attempts(${columns.join(', ')}) VALUES (${columns + .map(() => '?') + .join(', ')})`, + ) + .run(...(Object.values(values) as (string | number | null)[])); + }; + const base = { + completed_at: NOW, + logical_call_id: 'call-1', + turn_id: 'turn-1', + call_kind: 'main', + provider_id: 'anthropic', + model_id: 'claude-opus-5', + latency_ms: 10, + status: 'completed', + usage_basis: 'reported', + }; + // A price nobody could resolve must never surface as an amount. + assert.throws(() => + insert({ ...base, attempt_id: 'a', cost_basis: 'unpriced', cost_usd: 0.004 }), + ); + // A priced call must carry one; zero is legal and means genuinely free. + assert.throws(() => insert({ ...base, attempt_id: 'b', cost_basis: 'priced' })); + // "No usage reported" and "zero tokens" are different facts. + assert.throws(() => + insert({ + ...base, + attempt_id: 'c', + usage_basis: 'missing', + input_tokens: 0, + cost_basis: 'priced', + cost_usd: 0, + }), + ); + // Half a record is not a record. + assert.throws(() => insert({ attempt_id: 'd', completed_at: NOW, cost_basis: 'unpriced' })); + } finally { + database.close(); + } +}); diff --git a/packages/storage/src/conversation-operational-state.ts b/packages/storage/src/conversation-operational-state.ts index e1f1d36df0..e390a0a657 100644 --- a/packages/storage/src/conversation-operational-state.ts +++ b/packages/storage/src/conversation-operational-state.ts @@ -85,6 +85,10 @@ class SqliteConversationOperationalStateStore implements ConversationOperational database .prepare('DELETE FROM core_root_turn_start_rejections WHERE session_id = ?') .run(sessionId); + // Cascades this run's events and the Usage projection's checkpoints. + // `usage_model_call_attempts` is deliberately absent from this list: + // deleting a conversation must not erase its spend from all-time Usage + // totals, so those rows outlive the authority they were projected from. database.prepare('DELETE FROM core_agent_runs WHERE session_id = ?').run(sessionId); database .prepare('DELETE FROM core_client_capability_session_grants WHERE session_id = ?') diff --git a/packages/storage/src/model-call-ledger.ts b/packages/storage/src/model-call-ledger.ts index 19749e1d40..eee51d310e 100644 --- a/packages/storage/src/model-call-ledger.ts +++ b/packages/storage/src/model-call-ledger.ts @@ -21,12 +21,40 @@ import { decodeModelCallAttempt, MODEL_CALL_ATTEMPT_EVENT_TYPE, type ModelCallAttempt, + type ModelCallCoverage, } from '@maka/core/model-call-attempt'; +import { + resolveUsageRange, + type ModelCallUsageBuckets, + type ModelCallUsageLogs, + type ModelCallUsageSummary, +} from '@maka/core/model-call-usage-projection'; +import { usageBucketKey } from '@maka/core/usage-stats/bucket-key'; +import type { + UsageBucket, + UsageGroupBy, + UsageLogRow, + UsageQuery, +} from '@maka/core/usage-stats/types'; import type { DatabaseSync } from 'node:sqlite'; +import { + bucketGrouping, + CACHE_READ_TOKENS, + count, + countableFilter, + COVERAGE_SUMS, + PRICED_COST, + REQUEST_SUMS, + TOKEN_SUMS, + unreadableFilter, + type SqlFilter, +} from './model-call-usage-sql.js'; import { acquireOperationalStateDatabase, type OperationalStateDatabaseLease, } from './operational-state-store.js'; +import { MODEL_CALL_COLUMNS } from './sqlite-usage-schema.js'; +import type { ModelCallLedgerResult } from './usage-stores.js'; /** * Materialization of the canonical model-call accounting ledger (#1679). @@ -41,6 +69,21 @@ import { * — a failed upsert, a crash between the two — and that is recoverable: the * authority still holds every record, so re-projecting the run restores it. * + * That recovery covers live Sessions only. Deleting a Session drops its + * `core_agent_runs` rows and cascades both their events and this projection's + * checkpoints, while these rows are deliberately left standing — spend does not + * disappear from all-time totals because a conversation was deleted. For those + * rows the projection is the last copy, so nothing may rebuild this table by + * clearing it and replaying the stream. See + * `ConversationOperationalStateStore.purge`. + * + * A row holds one column per field a cost answer reads, and nothing else. + * Request shape and provider diagnostics are answered from the AgentRun stream; + * copied here they would make a row grow with the conversation rather than with + * spend. Because they are columns, a Usage total is a `SUM` this table computes + * — the reads below return answers, not records, so asking for an all-time + * total no longer means handing every call a workspace ever made to the caller. + * * Recovery compares the AgentRun stream's durable sequence with this * projection's applied-through checkpoint. There is no second "dirty" fact to * race with the authority: any committed event beyond the checkpoint remains @@ -54,22 +97,25 @@ import { */ export interface ModelCallLedgerReader { /** - * Attempts settled within `range`, deduped by `attemptId` with the last write - * winning, alongside the number of stored rows that could not be decoded. + * Usage answers over the rows a query addresses, alongside the number of rows + * in that window whose pricing was lost before this table held columns. * - * Unreadable rows are reported rather than dropped: they are real calls whose - * cost is now unknown, and a total that silently omits them overstates what - * the ledger knows. One corrupt row must not fail the query (#1638). + * Those are real calls whose cost is now unknown: they are reported rather + * than dropped, because a total that silently omits them overstates what the + * ledger knows. One of them cannot fail the query (#1638). */ - read( - range: { readonly from: number; readonly to: number }, - sessionId?: string, - ): ModelCallLedgerPage; -} - -export interface ModelCallLedgerPage { - readonly attempts: readonly ModelCallAttempt[]; - readonly unreadableRecords: number; + summary(query: UsageQuery, now: number): ModelCallLedgerResult; + buckets( + query: UsageQuery, + groupBy: UsageGroupBy, + now: number, + ): ModelCallLedgerResult; + logs( + query: UsageQuery, + now: number, + offset: number, + limit: number, + ): ModelCallLedgerResult; } export interface CatchUpModelCallProjectionInput { @@ -163,34 +209,136 @@ class SqliteModelCallLedger implements ModelCallLedger { ); } - read( - range: { readonly from: number; readonly to: number }, - sessionId?: string, - ): ModelCallLedgerPage { - if (this.#state !== 'open') throw new ModelCallLedgerClosedError(); - const rows = this.#lease.database + summary(query: UsageQuery, now: number): ModelCallLedgerResult { + const db = this.#open(); + const range = resolveUsageRange(query.range, now); + const filter = countableFilter(query, range); + const row = db .prepare( - sessionId - ? `SELECT record_json FROM usage_model_call_attempts - WHERE session_id = ? AND completed_at >= ? AND completed_at <= ? - ORDER BY completed_at ASC, attempt_id ASC` - : `SELECT record_json FROM usage_model_call_attempts - WHERE completed_at >= ? AND completed_at <= ? - ORDER BY completed_at ASC, attempt_id ASC`, + `SELECT ${REQUEST_SUMS}, ${TOKEN_SUMS}, ${COVERAGE_SUMS} + FROM usage_model_call_attempts WHERE ${filter.sql}`, ) - .all(...(sessionId ? [sessionId, range.from, range.to] : [range.from, range.to])) as Array<{ - record_json: string; - }>; - const attempts: ModelCallAttempt[] = []; - let unreadableRecords = 0; - for (const row of rows) { - try { - attempts.push(decodeModelCallAttempt(JSON.parse(row.record_json))); - } catch { - unreadableRecords += 1; - } - } - return { attempts, unreadableRecords }; + .get(...filter.parameters) as Record | undefined; + return { + projection: { + range, + totalRequests: count(row?.totalRequests), + totalCostUsd: count(row?.totalCostUsd), + totalDurationMs: count(row?.totalDurationMs), + totalTokens: readTokens(row), + cacheHitRequests: count(row?.cacheHitRequests), + cacheCreateRequests: count(row?.cacheCreateRequests), + errorRequests: count(row?.errorRequests), + coverage: readCoverage(row), + }, + unreadableRecords: this.#unreadable(query, range), + }; + } + + buckets( + query: UsageQuery, + groupBy: UsageGroupBy, + now: number, + ): ModelCallLedgerResult { + const db = this.#open(); + const range = resolveUsageRange(query.range, now); + const filter = countableFilter(query, range); + const rows = db + .prepare( + `SELECT MIN(provider_id) AS providerId, MIN(model_id) AS modelId, + MIN(completed_at) AS ts, COUNT(*) AS requests, + SUM(${PRICED_COST}) AS costUsd, SUM(latency_ms) AS latency, + SUM(status = 'failed') AS errors, ${TOKEN_SUMS} + FROM usage_model_call_attempts WHERE ${filter.sql} + GROUP BY ${bucketGrouping(groupBy)}`, + ) + .all(...filter.parameters) as Array>; + const buckets = rows + .map((row) => { + const requests = count(row.requests); + const tokens = readTokens(row); + const key = usageBucketKey( + { + providerId: String(row.providerId ?? ''), + modelId: String(row.modelId ?? ''), + ts: count(row.ts), + }, + groupBy, + ); + return { + key, + label: key, + requests, + inputTokens: tokens.input, + outputTokens: tokens.output, + cacheMissTokens: tokens.cacheMiss, + cacheReadTokens: tokens.cacheRead, + cacheWriteTokens: tokens.cacheWrite, + reasoningTokens: tokens.reasoning, + totalTokens: tokens.total, + costUsd: count(row.costUsd), + avgLatencyMs: requests === 0 ? 0 : count(row.latency) / requests, + errorRate: requests === 0 ? 0 : count(row.errors) / requests, + } satisfies UsageBucket; + }) + .sort((left, right) => right.requests - left.requests); + return { + projection: { buckets, coverage: this.#coverage(filter) }, + unreadableRecords: this.#unreadable(query, range), + }; + } + + logs( + query: UsageQuery, + now: number, + offset: number, + limit: number, + ): ModelCallLedgerResult { + const db = this.#open(); + const range = resolveUsageRange(query.range, now); + const filter = countableFilter(query, range); + const rows = db + .prepare( + `SELECT attempt_id, completed_at, call_kind, logical_call_id, connection_slug, + provider_id, model_id, cost_basis, cost_usd, latency_ms, status, error_class, + session_id, turn_id, + COALESCE(input_tokens, 0) AS input, + COALESCE(output_tokens, 0) AS output, + COALESCE(cache_miss_input_tokens, 0) AS cacheMiss, + ${CACHE_READ_TOKENS} AS cacheRead, + COALESCE(cache_write_input_tokens, 0) AS cacheWrite, + COALESCE(reasoning_tokens, 0) AS reasoning + FROM usage_model_call_attempts WHERE ${filter.sql} + ORDER BY completed_at DESC, attempt_id DESC + LIMIT ? OFFSET ?`, + ) + .all(...filter.parameters, limit, offset) as Array>; + const coverage = this.#coverage(filter); + return { + projection: { rows: rows.map(toUsageLogRow), total: coverage.attempts, coverage }, + unreadableRecords: this.#unreadable(query, range), + }; + } + + #open(): DatabaseSync { + if (this.#state !== 'open') throw new ModelCallLedgerClosedError(); + return this.#lease.database; + } + + #coverage(filter: SqlFilter): ModelCallCoverage { + const row = this.#lease.database + .prepare(`SELECT ${COVERAGE_SUMS} FROM usage_model_call_attempts WHERE ${filter.sql}`) + .get(...filter.parameters) as Record | undefined; + return readCoverage(row); + } + + #unreadable(query: UsageQuery, range: { from: number; to: number }): number { + const filter = unreadableFilter(query, range); + return count( + this.#lease.database + .prepare(`SELECT COUNT(*) AS unreadable FROM usage_model_call_attempts WHERE ${filter.sql}`) + .get(...filter.parameters)?.unreadable, + ); } async flush(): Promise { @@ -210,24 +358,117 @@ class SqliteModelCallLedger implements ModelCallLedger { } } +function readTokens(row: Record | undefined): { + input: number; + output: number; + cacheMiss: number; + cacheRead: number; + cacheWrite: number; + reasoning: number; + total: number; +} { + return { + input: count(row?.input), + output: count(row?.output), + cacheMiss: count(row?.cacheMiss), + cacheRead: count(row?.cacheRead), + cacheWrite: count(row?.cacheWrite), + reasoning: count(row?.reasoning), + total: count(row?.total), + }; +} + +function readCoverage(row: Record | undefined): ModelCallCoverage { + return { + attempts: count(row?.attempts), + pricedAttempts: count(row?.pricedAttempts), + unpricedAttempts: count(row?.unpricedAttempts), + usageReportedAttempts: count(row?.usageReportedAttempts), + usagePartialAttempts: count(row?.usagePartialAttempts), + usageMissingAttempts: count(row?.usageMissingAttempts), + }; +} + +function toUsageLogRow(row: Record): UsageLogRow { + const costBasis = row.cost_basis as UsageLogRow['costBasis']; + return { + id: String(row.attempt_id), + ts: count(row.completed_at), + callKind: row.call_kind as UsageLogRow['callKind'], + callId: String(row.logical_call_id), + ...(row.connection_slug === null ? {} : { connectionSlug: String(row.connection_slug) }), + providerId: String(row.provider_id), + modelId: String(row.model_id), + inputTokens: count(row.input), + outputTokens: count(row.output), + cacheMissTokens: count(row.cacheMiss), + cacheReadTokens: count(row.cacheRead), + cacheWriteTokens: count(row.cacheWrite), + reasoningTokens: count(row.reasoning), + totalTokens: count(row.input) + count(row.output), + // A row keeps its basis, not just its number. Collapsing an unpriced call + // to 0 here would reproduce, per row, exactly the ambiguity the coverage + // breakdown removes from the totals. + ...(costBasis === 'priced' ? { costUsd: count(row.cost_usd) } : {}), + costBasis, + latencyMs: count(row.latency_ms), + status: row.status === 'completed' ? 'success' : row.status === 'failed' ? 'error' : 'aborted', + ...(row.error_class === null ? {} : { errorClass: String(row.error_class) }), + sessionId: String(row.session_id), + turnId: String(row.turn_id), + }; +} + function positiveInteger(value: number | undefined, fallback: number, label: string): number { const resolved = value ?? fallback; if (!Number.isSafeInteger(resolved) || resolved <= 0) throw new Error(`Invalid ${label}`); return resolved; } +const MODEL_CALL_UPSERT = ` + INSERT INTO usage_model_call_attempts(${MODEL_CALL_COLUMNS.join(', ')}) + VALUES (${MODEL_CALL_COLUMNS.map(() => '?').join(', ')}) + ON CONFLICT(attempt_id) DO UPDATE SET + ${MODEL_CALL_COLUMNS.filter((column) => column !== 'attempt_id') + .map((column) => `${column} = excluded.${column}`) + .join(', ')} +`; + +/** + * The attempt's pricing fields, in column order. + * + * Keyed by column so the binding list cannot drift from the table: a column + * added to `MODEL_CALL_COLUMNS` without a value here is a compile error. + */ +function bindModelCallAttempt(attempt: ModelCallAttempt): (string | number | null)[] { + const values: Record<(typeof MODEL_CALL_COLUMNS)[number], string | number | null> = { + attempt_id: attempt.attemptId, + completed_at: attempt.completedAt, + session_id: attempt.sessionId, + logical_call_id: attempt.logicalCallId, + turn_id: attempt.turnId, + call_kind: attempt.callKind, + connection_slug: attempt.connectionSlug ?? null, + provider_id: attempt.providerId, + model_id: attempt.modelId, + latency_ms: attempt.latencyMs, + status: attempt.status, + error_class: attempt.errorClass ?? null, + usage_basis: attempt.usageBasis, + input_tokens: attempt.inputTokens ?? null, + output_tokens: attempt.outputTokens ?? null, + cache_read_input_tokens: attempt.cacheReadInputTokens ?? null, + cache_miss_input_tokens: attempt.cacheMissInputTokens ?? null, + cache_write_input_tokens: attempt.cacheWriteInputTokens ?? null, + reasoning_tokens: attempt.reasoningTokens ?? null, + cost_basis: attempt.costBasis, + cost_usd: attempt.costUsd ?? null, + }; + return MODEL_CALL_COLUMNS.map((column) => values[column]); +} + function writeModelCallAttempt(db: DatabaseSync, attempt: ModelCallAttempt): void { - db.prepare(` - INSERT INTO usage_model_call_attempts(attempt_id, completed_at, record_json, session_id) - VALUES (?, ?, ?, ?) - ON CONFLICT(attempt_id) DO UPDATE SET - completed_at = excluded.completed_at, - record_json = excluded.record_json, - session_id = excluded.session_id - WHERE completed_at IS NOT excluded.completed_at - OR record_json IS NOT excluded.record_json - OR session_id IS NOT excluded.session_id - `).run(attempt.attemptId, attempt.completedAt, JSON.stringify(attempt), attempt.sessionId); + db.prepare(MODEL_CALL_UPSERT).run(...bindModelCallAttempt(attempt)); } interface LaggingRunRow { diff --git a/packages/storage/src/model-call-usage-sql.ts b/packages/storage/src/model-call-usage-sql.ts new file mode 100644 index 0000000000..db76af5688 --- /dev/null +++ b/packages/storage/src/model-call-usage-sql.ts @@ -0,0 +1,156 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import type { UsageGroupBy, UsageQuery } from '@maka/core/usage-stats/types'; + +/** + * The Usage aggregation, expressed over the canonical ledger's columns. + * + * These fragments are the SQL half of rules whose vocabulary lives in + * `@maka/core`. Each one names the function it mirrors; change them together. + */ + +/** Mirrors `clampCacheReadTokens`: cache reads cannot exceed the prompt they came from. */ +export const CACHE_READ_TOKENS = ` + CASE WHEN input_tokens IS NULL + THEN COALESCE(cache_read_input_tokens, 0) + ELSE MIN(COALESCE(cache_read_input_tokens, 0), input_tokens) + END`; + +/** + * Unpriced records contribute nothing rather than zero. What they cost is + * reported through coverage instead, so a total never claims a call was free + * when the price was simply never resolved. + */ +export const PRICED_COST = `CASE WHEN cost_basis = 'priced' THEN COALESCE(cost_usd, 0) ELSE 0 END`; + +/** Mirrors `usageStatusForAttempt`: only a provider failure is an error. */ +const ERROR_ROW = `status = 'failed'`; + +export const TOKEN_SUMS = ` + SUM(COALESCE(input_tokens, 0)) AS input, + SUM(COALESCE(output_tokens, 0)) AS output, + SUM(COALESCE(cache_miss_input_tokens, 0)) AS cacheMiss, + SUM(${CACHE_READ_TOKENS}) AS cacheRead, + SUM(COALESCE(cache_write_input_tokens, 0)) AS cacheWrite, + SUM(COALESCE(reasoning_tokens, 0)) AS reasoning, + SUM(COALESCE(input_tokens, 0) + COALESCE(output_tokens, 0)) AS total`; + +export const COVERAGE_SUMS = ` + COUNT(*) AS attempts, + SUM(cost_basis = 'priced') AS pricedAttempts, + SUM(cost_basis = 'unpriced') AS unpricedAttempts, + SUM(usage_basis = 'reported') AS usageReportedAttempts, + SUM(usage_basis = 'partial') AS usagePartialAttempts, + SUM(usage_basis = 'missing') AS usageMissingAttempts`; + +export const REQUEST_SUMS = ` + COUNT(*) AS totalRequests, + SUM(${PRICED_COST}) AS totalCostUsd, + SUM(latency_ms) AS totalDurationMs, + SUM((${CACHE_READ_TOKENS}) > 0) AS cacheHitRequests, + SUM(COALESCE(cache_write_input_tokens, 0) > 0) AS cacheCreateRequests, + SUM(${ERROR_ROW}) AS errorRequests`; + +export interface SqlFilter { + readonly sql: string; + readonly parameters: readonly (string | number)[]; +} + +/** + * Rows a query addresses that can be counted. + * + * A tombstone — a row whose stored form was damaged before the ledger held + * columns — matches no filter and is excluded here; {@link unreadableFilter} + * counts it instead. + */ +export function countableFilter( + query: UsageQuery, + range: { readonly from: number; readonly to: number }, +): SqlFilter { + const clauses = ['cost_basis IS NOT NULL', 'completed_at >= ?', 'completed_at <= ?']; + const parameters: (string | number)[] = [range.from, range.to]; + const equals = (column: string, value: string | undefined) => { + if (value === undefined) return; + clauses.push(`${column} = ?`); + parameters.push(value); + }; + equals('session_id', query.sessionId); + equals('provider_id', query.providerId); + equals('model_id', query.modelId); + equals('connection_slug', query.connectionSlug); + if (query.status !== undefined && query.status !== 'all') { + // `interrupted` joins `aborted`: both mean the call stopped short without + // the provider reporting a failure. + if (query.status === 'success') clauses.push(`status = 'completed'`); + else if (query.status === 'error') clauses.push(ERROR_ROW); + else clauses.push(`status NOT IN ('completed', 'failed')`); + } + return { sql: clauses.join(' AND '), parameters }; +} + +/** + * Rows a query addresses whose pricing was lost. + * + * Scoped by window and Session only — the columns a tombstone keeps. Narrowing + * it by provider or status would drop the row from the report on the strength + * of a field the row no longer has, which is how a total quietly stops + * mentioning spend it cannot account for. + */ +export function unreadableFilter( + query: UsageQuery, + range: { readonly from: number; readonly to: number }, +): SqlFilter { + const clauses = ['cost_basis IS NULL', 'completed_at >= ?', 'completed_at <= ?']; + const parameters: (string | number)[] = [range.from, range.to]; + if (query.sessionId !== undefined) { + clauses.push('session_id = ?'); + parameters.push(query.sessionId); + } + return { sql: clauses.join(' AND '), parameters }; +} + +/** + * How SQL groups rows for a bucket query. + * + * SQLite decides only which rows belong together; the key string itself is + * still built by `usageBucketKey`, so both Usage sources keep deriving it from + * one place. `MIN(completed_at)` gives the time bucket a timestamp to name + * itself from. + */ +export function bucketGrouping(groupBy: UsageGroupBy): string { + switch (groupBy) { + case 'provider': + return 'provider_id'; + case 'model': + return 'provider_id, model_id'; + case 'day': + return `strftime('%Y-%m-%d', completed_at / 1000, 'unixepoch')`; + case 'hour': + return `strftime('%Y-%m-%dT%H', completed_at / 1000, 'unixepoch')`; + case 'tool': + // Tool invocations live in their own ledger; nothing here describes them. + return `''`; + } +} + +/** SQL aggregates arrive as `null` for an empty set and as bigint-safe numbers. */ +export function count(value: unknown): number { + return Number(value ?? 0); +} diff --git a/packages/storage/src/sqlite-usage-schema.ts b/packages/storage/src/sqlite-usage-schema.ts index 95c07c7a7d..a576298f95 100644 --- a/packages/storage/src/sqlite-usage-schema.ts +++ b/packages/storage/src/sqlite-usage-schema.ts @@ -19,7 +19,140 @@ import type { DatabaseSync } from 'node:sqlite'; -export const SQLITE_USAGE_SCHEMA_VERSION = 5; +export const SQLITE_USAGE_SCHEMA_VERSION = 7; + +/** + * The canonical ledger's columns, in the order every statement binds them. + * + * Ordered so `attempt_id, completed_at, session_id` — the three a damaged row + * keeps — come first, and the pricing columns follow. + */ +export const MODEL_CALL_COLUMNS = [ + 'attempt_id', + 'completed_at', + 'session_id', + 'logical_call_id', + 'turn_id', + 'call_kind', + 'connection_slug', + 'provider_id', + 'model_id', + 'latency_ms', + 'status', + 'error_class', + 'usage_basis', + 'input_tokens', + 'output_tokens', + 'cache_read_input_tokens', + 'cache_miss_input_tokens', + 'cache_write_input_tokens', + 'reasoning_tokens', + 'cost_basis', + 'cost_usd', +] as const; + +/** + * The JSON path each pricing column was read from before the columns existed, + * used once by the migration that converted the rows. + */ +const MODEL_CALL_COLUMN_SOURCES: readonly (readonly [string, string])[] = [ + ['logical_call_id', '$.logicalCallId'], + ['turn_id', '$.turnId'], + ['call_kind', '$.callKind'], + ['connection_slug', '$.connectionSlug'], + ['provider_id', '$.providerId'], + ['model_id', '$.modelId'], + ['latency_ms', '$.latencyMs'], + ['status', '$.status'], + ['error_class', '$.errorClass'], + ['usage_basis', '$.usageBasis'], + ['input_tokens', '$.inputTokens'], + ['output_tokens', '$.outputTokens'], + ['cache_read_input_tokens', '$.cacheReadInputTokens'], + ['cache_miss_input_tokens', '$.cacheMissInputTokens'], + ['cache_write_input_tokens', '$.cacheWriteInputTokens'], + ['reasoning_tokens', '$.reasoningTokens'], + ['cost_basis', '$.costBasis'], + ['cost_usd', '$.costUsd'], +]; + +/** The columns that are present together or not at all. See the table's CHECK. */ +const MODEL_CALL_REQUIRED_COLUMNS = [ + 'logical_call_id', + 'turn_id', + 'call_kind', + 'provider_id', + 'model_id', + 'latency_ms', + 'status', + 'usage_basis', + 'cost_basis', +] as const; + +const MODEL_CALL_TOKEN_COLUMNS = [ + 'input_tokens', + 'output_tokens', + 'cache_read_input_tokens', + 'cache_miss_input_tokens', + 'cache_write_input_tokens', + 'reasoning_tokens', +] as const; + +const NO_TOKENS = MODEL_CALL_TOKEN_COLUMNS.map((column) => `${column} IS NULL`).join(' AND '); + +/** + * Canonical model-call accounting ledger (#1679). + * + * One column per field a cost answer reads, so the totals are a `SUM` the + * database can compute rather than every row of a workspace's history parsed + * into memory first. + * + * A row is either a complete pricing record or a tombstone that kept only the + * identity and timestamp of a call whose stored form was damaged. Never half of + * each — which is what makes `cost_basis IS NOT NULL` a sound test for "this row + * can be counted", and its negation the count a read reports as unreadable. + * + * The vocabularies (`status`, `call_kind`) are deliberately not constrained + * here. They are already validated where a record is decoded, and a CHECK on + * them would turn one damaged row into a failed migration for the whole + * workspace. + */ +const MODEL_CALL_TABLE = ` + CREATE TABLE IF NOT EXISTS %TABLE% ( + attempt_id TEXT PRIMARY KEY, + completed_at INTEGER NOT NULL CHECK (completed_at >= 0), + session_id TEXT, + logical_call_id TEXT, + turn_id TEXT, + call_kind TEXT, + connection_slug TEXT, + provider_id TEXT, + model_id TEXT, + latency_ms INTEGER, + status TEXT, + error_class TEXT, + usage_basis TEXT, + input_tokens INTEGER, + output_tokens INTEGER, + cache_read_input_tokens INTEGER, + cache_miss_input_tokens INTEGER, + cache_write_input_tokens INTEGER, + reasoning_tokens INTEGER, + cost_basis TEXT, + cost_usd REAL, + CHECK (${MODEL_CALL_REQUIRED_COLUMNS.map( + (column) => `(${column} IS NULL) = (cost_basis IS NULL)`, + ).join(' AND ')}), + -- A price that could not be resolved must never surface as an amount, and a + -- priced call must carry one. Zero stays legal: it is the only way to say a + -- call was genuinely free. + CHECK (cost_basis IS NOT 'priced' OR cost_usd IS NOT NULL), + CHECK (cost_basis IS NOT 'unpriced' OR cost_usd IS NULL), + -- "The provider reported no usage" and "it reported zero" are different + -- facts, so a missing-usage row carries no token counts at all. + CHECK (usage_basis IS NOT 'missing' OR (${NO_TOKENS})) + ) +`; export function migrateSqliteUsageDatabase(db: DatabaseSync): void { db.exec(` @@ -44,18 +177,7 @@ export function migrateSqliteUsageDatabase(db: DatabaseSync): void { CREATE INDEX IF NOT EXISTS usage_tool_invocations_ts ON usage_tool_invocations(ts DESC, id); - -- Canonical model-call accounting ledger (#1679). Separate from - -- usage_llm_calls, which is a frozen historical projection: these rows carry - -- usageBasis/costBasis, which that schema cannot express. - CREATE TABLE IF NOT EXISTS usage_model_call_attempts ( - attempt_id TEXT PRIMARY KEY, - completed_at INTEGER NOT NULL CHECK (completed_at >= 0), - record_json TEXT NOT NULL, - session_id TEXT - ); - - CREATE INDEX IF NOT EXISTS usage_model_call_attempts_completed_at - ON usage_model_call_attempts(completed_at DESC, attempt_id); + ${MODEL_CALL_TABLE.replace('%TABLE%', 'usage_model_call_attempts')}; -- The AgentRun sequence is the projection's sole progress authority. A run -- is behind exactly when its latest model-call event is newer than this @@ -86,26 +208,98 @@ export function migrateSqliteUsageDatabase(db: DatabaseSync): void { `); db.exec('DROP TABLE IF EXISTS usage_model_call_reprojection'); ensureColumn(db, 'usage_llm_calls', 'session_id', 'TEXT'); - ensureColumn(db, 'usage_model_call_attempts', 'session_id', 'TEXT'); db.exec(` UPDATE usage_llm_calls SET session_id = json_extract(record_json, '$.sessionId') WHERE session_id IS NULL AND json_valid(record_json); - UPDATE usage_model_call_attempts - SET session_id = json_extract(record_json, '$.sessionId') - WHERE session_id IS NULL AND json_valid(record_json); - CREATE INDEX IF NOT EXISTS usage_llm_calls_session_ts ON usage_llm_calls(session_id, ts DESC, id); + `); + // A ledger old enough to predate Session attribution has no column to carry + // through, and the conversion below reads one. + ensureColumn(db, 'usage_model_call_attempts', 'session_id', 'TEXT'); + spreadModelCallRecordJson(db); + db.exec(` + CREATE INDEX IF NOT EXISTS usage_model_call_attempts_completed_at + ON usage_model_call_attempts(completed_at DESC, attempt_id); CREATE INDEX IF NOT EXISTS usage_model_call_attempts_session_completed_at ON usage_model_call_attempts(session_id, completed_at DESC, attempt_id); `); } +/** + * Converts rows that stored the record as one JSON blob into the columns. + * + * Not rebuilt from the AgentRun authority, the usual move for a read model: + * deleting a Session drops its runs and cascades their events while these rows + * are kept on purpose, so for those calls this table is the last copy and a + * wipe-and-replay would erase their spend. See the header of + * `model-call-ledger.ts`. + * + * A blob that does not yield a whole record — damaged text, or a row some other + * schema left behind — keeps its identity and timestamp and loses the rest. That + * is the same claim the old JSON reader made by counting it as unreadable, and + * it is why the conversion is all-or-nothing per row: a half-filled row would + * make the table's own CHECK unsatisfiable and take the whole migration with it. + */ +function spreadModelCallRecordJson(db: DatabaseSync): void { + if (!hasColumn(db, 'usage_model_call_attempts', 'record_json')) return; + const extracted = MODEL_CALL_COLUMN_SOURCES.map( + ([column, path]) => + `CASE WHEN json_valid(record_json) THEN json_extract(record_json, '${path}') END AS ${column}`, + ).join(',\n '); + const readable = [ + ...MODEL_CALL_REQUIRED_COLUMNS.map((column) => `${column} IS NOT NULL`), + "(cost_basis IS NOT 'priced' OR cost_usd IS NOT NULL)", + "(cost_basis IS NOT 'unpriced' OR cost_usd IS NULL)", + `(usage_basis IS NOT 'missing' OR (${NO_TOKENS}))`, + ].join('\n AND '); + const pricing = MODEL_CALL_COLUMN_SOURCES.map( + ([column]) => `CASE WHEN readable THEN ${column} END`, + ).join(',\n '); + // The old table moves aside so the new one is created under its final name: + // a table renamed into place keeps a rewritten `CREATE` statement, and the + // schema guard compares those texts. + db.exec(` + ALTER TABLE usage_model_call_attempts RENAME TO usage_model_call_attempts_blob; + + ${MODEL_CALL_TABLE.replace('%TABLE%', 'usage_model_call_attempts')}; + + INSERT INTO usage_model_call_attempts(${MODEL_CALL_COLUMNS.join(', ')}) + WITH extracted AS ( + SELECT + attempt_id, + completed_at, + -- A tombstone keeps its Session: the row still says a call happened here. + COALESCE( + session_id, + CASE WHEN json_valid(record_json) THEN json_extract(record_json, '$.sessionId') END + ) AS session_id, + ${extracted} + FROM usage_model_call_attempts_blob + ), + classified AS ( + SELECT *, (${readable}) AS readable FROM extracted + ) + SELECT + attempt_id, + completed_at, + session_id, + ${pricing} + FROM classified; + + DROP TABLE usage_model_call_attempts_blob; + `); +} + function ensureColumn(db: DatabaseSync, table: string, column: string, definition: string): void { - const columns = db.prepare(`PRAGMA table_info(${table})`).all() as Array<{ name?: unknown }>; - if (columns.some((candidate) => candidate.name === column)) return; + if (hasColumn(db, table, column)) return; db.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`); } + +function hasColumn(db: DatabaseSync, table: string, column: string): boolean { + const columns = db.prepare(`PRAGMA table_info(${table})`).all() as Array<{ name?: unknown }>; + return columns.some((candidate) => candidate.name === column); +} diff --git a/packages/storage/src/usage-stores.ts b/packages/storage/src/usage-stores.ts index 3e4b3a8953..0148ce3645 100644 --- a/packages/storage/src/usage-stores.ts +++ b/packages/storage/src/usage-stores.ts @@ -17,6 +17,11 @@ * under the License. */ +import type { + ModelCallUsageBuckets, + ModelCallUsageLogs, + ModelCallUsageSummary, +} from '@maka/core/model-call-usage-projection'; import type { PricingConfig, UsageBucket, @@ -33,7 +38,6 @@ import { ModelCallLedgerClosedError, ModelCallLedgerPublicationError, type ModelCallLedger, - type ModelCallLedgerPage, type ModelCallLedgerReader, } from './model-call-ledger.js'; import { @@ -99,14 +103,28 @@ export interface TelemetryIndexWriter extends TelemetryIndexReader { * synchronous store beneath it — because every authority read goes through the * storage-root lease. */ +/** One Usage answer from the canonical ledger, with the rows it could not read. */ +export interface ModelCallLedgerResult { + readonly projection: T; + readonly unreadableRecords: number; +} + export interface ModelCallIndexReader { - modelCallAttempts( - range: { - readonly from: number; - readonly to: number; - }, - sessionId?: string, - ): Promise; + modelCallSummary( + query: UsageQuery, + now: number, + ): Promise>; + modelCallBuckets( + query: UsageQuery, + groupBy: UsageGroupBy, + now: number, + ): Promise>; + modelCallLogs( + query: UsageQuery, + now: number, + offset: number, + limit: number, + ): Promise>; } export interface ModelCallIndexWriter extends ModelCallIndexReader { @@ -467,7 +485,11 @@ function createWriterFacade( admitSessionUsageMutation(record.sessionId, () => telemetry.insertToolInvocation(record)), }, modelCalls: { - modelCallAttempts: (range, sessionId) => read(() => modelCalls.read(range, sessionId)), + modelCallSummary: (query, now) => read(() => modelCalls.summary(query, now)), + modelCallBuckets: (query, groupBy, now) => + read(() => modelCalls.buckets(query, groupBy, now)), + modelCallLogs: (query, now, offset, limit) => + read(() => modelCalls.logs(query, now, offset, limit)), catchUpModelCallProjection: admitModelCallProjectionCatchUp, }, pricing: { @@ -515,10 +537,11 @@ function modelCallReader( run: (operation: () => T | Promise) => Promise, ): Readonly { return Object.freeze({ - modelCallAttempts: ( - range: { readonly from: number; readonly to: number }, - sessionId?: string, - ) => run(() => ledger.read(range, sessionId)), + modelCallSummary: (query: UsageQuery, now: number) => run(() => ledger.summary(query, now)), + modelCallBuckets: (query: UsageQuery, groupBy: UsageGroupBy, now: number) => + run(() => ledger.buckets(query, groupBy, now)), + modelCallLogs: (query: UsageQuery, now: number, offset: number, limit: number) => + run(() => ledger.logs(query, now, offset, limit)), }); }