diff --git a/README.md b/README.md index 1c1a9dad63..3d411e8e6e 100644 --- a/README.md +++ b/README.md @@ -217,11 +217,15 @@ Desktop and Runtime Host production wiring opens these SQLite repositories; their JSON/JSONL predecessors are read only during a fingerprinted, crash-safe cutover and are never updated by later mutations. -The remaining structured operational stores are deliberately classified rather -than implied complete: +Usage telemetry and pricing authority now use that same operational database. +Legacy `telemetry.json` and `pricing.json` sources are decoded together and +fingerprinted before their rows and pricing revision are committed atomically. +After cutover, Desktop and Runtime Host write only `runtime.sqlite`; the source +files remain unchanged as migration evidence. + +The remaining storage work is deliberately classified rather than implied +complete: -- usage telemetry and pricing authority move in the next workflow-metadata - slice; - artifact metadata and lifecycle state move with a recoverable metadata/payload publication protocol, while payload bytes remain files; - StoredMessage transcript bodies remain append-only JSONL; diff --git a/apps/desktop/src/main/app-lifecycle.ts b/apps/desktop/src/main/app-lifecycle.ts index 52311b5ef9..ee050762ee 100644 --- a/apps/desktop/src/main/app-lifecycle.ts +++ b/apps/desktop/src/main/app-lifecycle.ts @@ -15,7 +15,7 @@ import type { createProjectCatalog, createSessionStore, createSettingsStore, - createTelemetryRepo, + createSqliteTelemetryRepo, openRuntimeEventPersistence, } from '@maka/storage'; import type { createAgentGraphControlStore } from '@maka/storage/agent-graph-control-store'; @@ -56,7 +56,7 @@ export interface AppLifecycleDeps { credentialStore: ReturnType; connectionStore: ReturnType; settingsStore: ReturnType; - telemetryRepo: ReturnType; + telemetryRepo: ReturnType; ensureUsageReady: () => Promise; keepSystemAwake: KeepSystemAwakeController; botRegistry: BotRegistry; diff --git a/apps/desktop/src/main/main.ts b/apps/desktop/src/main/main.ts index e6e2ebbb50..ac832fd033 100644 --- a/apps/desktop/src/main/main.ts +++ b/apps/desktop/src/main/main.ts @@ -73,7 +73,7 @@ import { createSettingsStore, createMcpConfigStore, createShellRunStore, - createTelemetryRepo, + createSqliteTelemetryRepo, } from '@maka/storage'; import { createAgentGraphControlStore } from '@maka/storage/agent-graph-control-store'; import { resolveWorkspaceIdentity } from '@maka/storage/workspace-identity'; @@ -291,7 +291,7 @@ function ensureMcpReady(): Promise { } return mcpStartup; } -const telemetryRepo = createTelemetryRepo(workspaceRoot); +const telemetryRepo = createSqliteTelemetryRepo(workspaceRoot); const dailyReviewArchiveStore = createDailyReviewArchiveStore(workspaceRoot); const artifactStore = createArtifactStore(workspaceRoot); const deepResearchStore = createSqliteDeepResearchStore(workspaceRoot); diff --git a/packages/runtime-host/src/__tests__/usage-pricing-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/usage-pricing-two-client-uds.test.ts index 080d80efdb..ee3ffd387e 100644 --- a/packages/runtime-host/src/__tests__/usage-pricing-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/usage-pricing-two-client-uds.test.ts @@ -1,8 +1,8 @@ import assert from 'node:assert/strict'; -import { lstat, mkdir, mkdtemp, open, rm } from 'node:fs/promises'; +import { lstat, mkdtemp, rename, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { describe, mock, test } from 'node:test'; +import { describe, test } from 'node:test'; import { PRICING_MODEL_KEY_MAX_CHARS } from '@maka/core/usage-stats/pricing'; import type { PricingConfig } from '@maka/core/usage-stats/types'; import { openInteractiveUsageStoresForWrite } from '@maka/storage/usage-stores'; @@ -201,59 +201,7 @@ test('pricing mutation registers backend invalidation before the next activation }); }); -test('pricing commit-unknown requests drain before the typed failure and poisoned read', { - skip: process.platform === 'win32', -}, async () => { - await withUsageAuthority('pricing-commit-unknown', async ({ root, stores }) => { - let drainRequests = 0; - let invalidations = 0; - const coordinator = new HostUsagePricingCoordinator( - stores, - () => { - drainRequests += 1; - }, - new RuntimePolicyActivationGate(), - async () => { - invalidations += 1; - }, - ); - const restoreSync = await failNextDirectorySync(root); - let mutation; - try { - mutation = await coordinator.handlers['pricing.mutate']( - { - expectedRevision: 0, - mutation: { kind: 'upsert', pricing: pricing('provider:unknown', 1) }, - }, - CONNECTION_CONTEXT, - ); - } finally { - restoreSync(); - } - assert.deepEqual(mutation, { - ok: false, - error: { - code: 'commit_outcome_unknown', - message: 'Pricing mutation commit outcome is unknown', - }, - }); - assert.equal(drainRequests, 1); - assert.equal(invalidations, 1); - assert.deepEqual( - await coordinator.handlers['pricing.query']({ kind: 'start' }, CONNECTION_CONTEXT), - { - ok: false, - error: { - code: 'persistence_failed', - message: 'Pricing authority persistence failed', - }, - }, - ); - assert.equal(drainRequests, 1); - }); -}); - -test('pricing publication failure requests drain while expected failures do not', async () => { +test('pricing root identity failure requests drain while expected failures do not', async () => { await withUsageAuthority('pricing-publication', async ({ root, stores }) => { let drainRequests = 0; const coordinator = new HostUsagePricingCoordinator( @@ -314,8 +262,8 @@ test('pricing publication failure requests drain while expected failures do not' ); assert.equal(drainRequests, 0); - await rm(join(root, 'pricing.json')); - await mkdir(join(root, 'pricing.json')); + const movedRoot = `${root}-moved`; + await rename(root, movedRoot); assert.deepEqual( await coordinator.handlers['pricing.mutate']( { @@ -333,49 +281,7 @@ test('pricing publication failure requests drain while expected failures do not' }, ); assert.equal(drainRequests, 1); - }); -}); - -test('telemetry poisoned read fails closed and requests drain once', { - skip: process.platform === 'win32', -}, async () => { - await withUsageAuthority('telemetry-commit-unknown', async ({ root, stores }) => { - let drainRequests = 0; - const coordinator = new HostUsagePricingCoordinator( - stores, - () => { - drainRequests += 1; - }, - new RuntimePolicyActivationGate(), - ); - const restoreSync = await failNextDirectorySync(root); - try { - await assert.rejects(stores.telemetry.recordToolInvocation(toolRecord('tool-poison', 30))); - } finally { - restoreSync(); - } - const query = { - kind: 'logs', - source: 'tool', - query: { range: 'all' }, - } as const; - const expected = { - ok: false, - error: { - code: 'persistence_failed', - message: 'Usage authority persistence failed', - }, - } as const; - assert.deepEqual( - await coordinator.handlers['usage.query'](query, CONNECTION_CONTEXT), - expected, - ); - assert.equal(drainRequests, 1); - assert.deepEqual( - await coordinator.handlers['usage.query'](query, CONNECTION_CONTEXT), - expected, - ); - assert.equal(drainRequests, 1); + await rename(movedRoot, root); }); }); @@ -815,25 +721,6 @@ function rejectedReasons(results: readonly PromiseSettledResult[]): unk return results.flatMap((result) => (result.status === 'rejected' ? [result.reason] : [])); } -async function failNextDirectorySync(root: string): Promise<() => void> { - const probe = await open(root, 'r'); - const fileHandlePrototype = Object.getPrototypeOf(probe) as { - sync: typeof probe.sync; - }; - const originalSync = fileHandlePrototype.sync; - await probe.close(); - let injected = false; - const syncMock = mock.method(fileHandlePrototype, 'sync', async function (this: typeof probe) { - const metadata = await this.stat(); - if (!injected && metadata.isDirectory()) { - injected = true; - throw new Error('injected usage authority directory sync failure'); - } - return originalSync.call(this); - }); - return () => syncMock.mock.restore(); -} - async function withUsageAuthority( name: string, run: (context: { diff --git a/packages/storage/src/__tests__/operational-state-store.test.ts b/packages/storage/src/__tests__/operational-state-store.test.ts index d9d9e334b8..3f6c634b2a 100644 --- a/packages/storage/src/__tests__/operational-state-store.test.ts +++ b/packages/storage/src/__tests__/operational-state-store.test.ts @@ -44,6 +44,7 @@ describe('operational state database cutover', () => { { scope: 'operational', version: 1 }, { scope: 'runtime', version: 5 }, { scope: 'session_metadata', version: SQLITE_SESSION_METADATA_SCHEMA_VERSION }, + { scope: 'usage', version: 1 }, { scope: 'workflow', version: 1 }, ], ); diff --git a/packages/storage/src/__tests__/sqlite-usage-store.test.ts b/packages/storage/src/__tests__/sqlite-usage-store.test.ts new file mode 100644 index 0000000000..867fc892aa --- /dev/null +++ b/packages/storage/src/__tests__/sqlite-usage-store.test.ts @@ -0,0 +1,160 @@ +import assert from 'node:assert/strict'; +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { describe, test } from 'node:test'; +import { acquireOperationalStateDatabase } from '../operational-state-store.js'; +import { PricingStorePublicationError } from '../pricing-store.js'; +import { createSqlitePricingStore, createSqliteTelemetryRepo } from '../sqlite-usage-store.js'; + +describe('SQLite usage and pricing stores', () => { + test('resumes a cutover crash after copied rows without exposing partial state', async () => { + await withRoot(async (root) => { + await writeFile( + join(root, 'telemetry.json'), + JSON.stringify({ usageRecords: [llmRecord()], toolInvocations: [] }), + ); + const interrupted = createSqliteTelemetryRepo(root, { + managePricing: false, + failpoint: (point) => { + if (point === 'after_cutover_rows_copied') throw new Error('simulated crash'); + }, + }); + await assert.rejects(() => interrupted.load(), /simulated crash/); + await interrupted.close(); + + const resumed = createSqliteTelemetryRepo(root, { managePricing: false }); + await resumed.load(); + assert.equal(resumed.logs({ range: 'all' }).total, 1); + await resumed.close(); + }); + }); + + test('fails closed when a legacy source changes after completed cutover', async () => { + await withRoot(async (root) => { + const path = join(root, 'telemetry.json'); + await writeFile(path, JSON.stringify({ usageRecords: [llmRecord()], toolInvocations: [] })); + const first = createSqliteTelemetryRepo(root, { managePricing: false }); + await first.load(); + await first.close(); + + await writeFile( + path, + JSON.stringify({ + usageRecords: [llmRecord(), llmRecord({ id: 'usage_2' })], + toolInvocations: [], + }), + ); + const reopened = createSqliteTelemetryRepo(root, { managePricing: false }); + await assert.rejects( + () => reopened.load(), + /Legacy usage_pricing source changed after cutover completed/, + ); + await reopened.close(); + }); + }); + + test('writes only runtime.sqlite after cutover and preserves pricing revisions', async () => { + await withRoot(async (root) => { + const telemetry = createSqliteTelemetryRepo(root, { managePricing: false }); + const pricing = createSqlitePricingStore(root); + await telemetry.load(); + await pricing.load(); + await telemetry.insertLlmCall(llmRecord()); + const changed = await pricing.upsert(0, { + modelKey: 'openai:gpt-5', + inputUsdPer1M: 1.25, + outputUsdPer1M: 10, + }); + assert.equal(changed.snapshot.revision, 1); + await telemetry.close(); + await pricing.close(); + + await assert.rejects( + () => readFile(join(root, 'telemetry.json'), 'utf8'), + (error: NodeJS.ErrnoException) => error.code === 'ENOENT', + ); + await assert.rejects( + () => readFile(join(root, 'pricing.json'), 'utf8'), + (error: NodeJS.ErrnoException) => error.code === 'ENOENT', + ); + assert.ok((await readFile(join(root, 'runtime.sqlite'))).byteLength > 0); + + const reopenedTelemetry = createSqliteTelemetryRepo(root, { managePricing: false }); + const reopenedPricing = createSqlitePricingStore(root); + await reopenedTelemetry.load(); + await reopenedPricing.load(); + assert.equal(reopenedTelemetry.logs({ range: 'all' }).total, 1); + assert.deepEqual(reopenedPricing.snapshot(), changed.snapshot); + await reopenedTelemetry.close(); + await reopenedPricing.close(); + }); + }); + + test('rolls back the whole pricing revision when an override row cannot commit', async () => { + await withRoot(async (root) => { + const pricing = createSqlitePricingStore(root); + await pricing.load(); + await pricing.upsert(0, { + modelKey: 'provider:stable', + inputUsdPer1M: 1, + outputUsdPer1M: 2, + }); + const before = pricing.snapshot(); + const inspection = acquireOperationalStateDatabase(root); + inspection.database.exec(` + CREATE TRIGGER reject_test_pricing_override + BEFORE INSERT ON usage_pricing_overrides + WHEN NEW.model_key = 'provider:reject' + BEGIN + SELECT RAISE(ABORT, 'injected pricing row failure'); + END; + `); + await assert.rejects( + () => + pricing.upsert(before.revision, { + modelKey: 'provider:reject', + inputUsdPer1M: 3, + outputUsdPer1M: 4, + }), + PricingStorePublicationError, + ); + assert.deepEqual(pricing.snapshot(), before); + inspection.database.exec('DROP TRIGGER reject_test_pricing_override'); + inspection.close(); + await pricing.close(); + }); + }); +}); + +async function withRoot(run: (root: string) => Promise): Promise { + const root = await mkdtemp(join(tmpdir(), 'maka-sqlite-usage-')); + try { + await run(root); + } finally { + await rm(root, { recursive: true, force: true }); + } +} + +function llmRecord(overrides: Record = {}) { + return { + id: 'usage_1', + providerId: 'openai', + modelId: 'gpt-5', + inputTokens: 10, + outputTokens: 20, + cacheHitInputTokens: 0, + cacheMissInputTokens: 10, + cachedInputTokens: 0, + cacheWriteInputTokens: 0, + reasoningTokens: 0, + totalTokens: 30, + costUsd: 0.001, + latencyMs: 100, + status: 'success', + date: '2026-01-01', + ts: Date.UTC(2026, 0, 1), + startedAt: Date.UTC(2026, 0, 1) - 100, + ...overrides, + } as Parameters['insertLlmCall']>[0]; +} diff --git a/packages/storage/src/__tests__/usage-stores.test.ts b/packages/storage/src/__tests__/usage-stores.test.ts index aa65e86fb2..d4e4efafb8 100644 --- a/packages/storage/src/__tests__/usage-stores.test.ts +++ b/packages/storage/src/__tests__/usage-stores.test.ts @@ -199,7 +199,8 @@ describe('InteractiveUsageStores', () => { toolInvocations: [toolRecord()], pricingOverrides: [pricing('openai:gpt-5')], }; - await writeFile(join(root, 'telemetry.json'), JSON.stringify(legacy, null, 2) + '\n'); + const legacyBytes = JSON.stringify(legacy, null, 2) + '\n'; + await writeFile(join(root, 'telemetry.json'), legacyBytes); const owner = await tryAcquireInteractiveRootOwner(capability); assert(owner); @@ -212,10 +213,11 @@ describe('InteractiveUsageStores', () => { await stores.close(); await owner.close(); - const firstTelemetry = await readFile(join(root, 'telemetry.json'), 'utf8'); - const firstPricing = await readFile(join(root, 'pricing.json'), 'utf8'); - assert.match(firstTelemetry, /"version": 1/); - assert.doesNotMatch(firstTelemetry, /pricingOverrides/); + assert.equal(await readFile(join(root, 'telemetry.json'), 'utf8'), legacyBytes); + await assert.rejects( + () => readFile(join(root, 'pricing.json'), 'utf8'), + (error: NodeJS.ErrnoException) => error.code === 'ENOENT', + ); const successor = await tryAcquireInteractiveRootOwner(capability); assert(successor); @@ -227,8 +229,11 @@ describe('InteractiveUsageStores', () => { }); await reopened.close(); await successor.close(); - assert.equal(await readFile(join(root, 'telemetry.json'), 'utf8'), firstTelemetry); - assert.equal(await readFile(join(root, 'pricing.json'), 'utf8'), firstPricing); + assert.equal(await readFile(join(root, 'telemetry.json'), 'utf8'), legacyBytes); + await assert.rejects( + () => readFile(join(root, 'pricing.json'), 'utf8'), + (error: NodeJS.ErrnoException) => error.code === 'ENOENT', + ); }); }); @@ -307,8 +312,17 @@ describe('InteractiveUsageStores', () => { ); await Promise.all([accepted, drained]); await stores.close(); - assert.match(await readFile(join(root, 'telemetry.json'), 'utf8'), /"usage_1"/); + await assert.rejects( + () => readFile(join(root, 'telemetry.json'), 'utf8'), + (error: NodeJS.ErrnoException) => error.code === 'ENOENT', + ); await owner.close(); + const successor = await tryAcquireInteractiveRootOwner(capability); + assert(successor); + const reopened = await openInteractiveUsageStoresForWrite(successor.lease); + assert.equal((await reopened.telemetry.logs({ range: 'all' })).rows[0]?.id, 'usage_1'); + await reopened.close(); + await successor.close(); }); }); diff --git a/packages/storage/src/index.ts b/packages/storage/src/index.ts index 377bd2af33..4e947ee4cf 100644 --- a/packages/storage/src/index.ts +++ b/packages/storage/src/index.ts @@ -22,6 +22,7 @@ export type { } from './credential-store.js'; export * from './settings-store.js'; export * from './telemetry-repo.js'; +export * from './sqlite-usage-store.js'; export * from './usage-stores.js'; export { ARTIFACT_BINARY_PREVIEW_LIMIT_BYTES, diff --git a/packages/storage/src/operational-state-store.ts b/packages/storage/src/operational-state-store.ts index db3496c2e1..5b424e7e14 100644 --- a/packages/storage/src/operational-state-store.ts +++ b/packages/storage/src/operational-state-store.ts @@ -20,6 +20,7 @@ import { migrateSqliteWorkflowDatabase, SQLITE_WORKFLOW_SCHEMA_VERSION, } from './sqlite-workflow-schema.js'; +import { migrateSqliteUsageDatabase, SQLITE_USAGE_SCHEMA_VERSION } from './sqlite-usage-schema.js'; export const OPERATIONAL_STATE_DATABASE_NAME = 'runtime.sqlite'; export const LEGACY_SESSION_METADATA_DATABASE_NAME = 'sessions.sqlite'; @@ -126,6 +127,7 @@ class OperationalStateDatabaseOwner { migrateSqliteSessionMetadataDatabase(this.database); migrateSqliteCoreExecutionDatabase(this.database); migrateSqliteWorkflowDatabase(this.database); + migrateSqliteUsageDatabase(this.database); migrateOperationalStateDatabase(this.database, options.now ?? Date.now); cutoverLegacySessionMetadata({ destination: this.database, @@ -204,6 +206,7 @@ function migrateOperationalStateDatabase(db: DatabaseSync, now: () => number): v registerSchema(db, 'session_metadata', SQLITE_SESSION_METADATA_SCHEMA_VERSION, appliedAt); registerSchema(db, 'core_execution', SQLITE_CORE_EXECUTION_SCHEMA_VERSION, appliedAt); registerSchema(db, 'workflow', SQLITE_WORKFLOW_SCHEMA_VERSION, appliedAt); + registerSchema(db, 'usage', SQLITE_USAGE_SCHEMA_VERSION, appliedAt); registerSchema(db, 'operational', OPERATIONAL_STATE_SCHEMA_VERSION, appliedAt); db.exec('COMMIT'); } catch (error) { diff --git a/packages/storage/src/sqlite-usage-schema.ts b/packages/storage/src/sqlite-usage-schema.ts new file mode 100644 index 0000000000..02aac6fdf7 --- /dev/null +++ b/packages/storage/src/sqlite-usage-schema.ts @@ -0,0 +1,37 @@ +import type { DatabaseSync } from 'node:sqlite'; + +export const SQLITE_USAGE_SCHEMA_VERSION = 1; + +export function migrateSqliteUsageDatabase(db: DatabaseSync): void { + db.exec(` + CREATE TABLE IF NOT EXISTS usage_llm_calls ( + storage_key TEXT PRIMARY KEY, + id TEXT NOT NULL, + ts INTEGER NOT NULL CHECK (ts >= 0), + record_json TEXT NOT NULL + ); + + CREATE INDEX IF NOT EXISTS usage_llm_calls_ts + ON usage_llm_calls(ts DESC, id); + + CREATE TABLE IF NOT EXISTS usage_tool_invocations ( + storage_key TEXT PRIMARY KEY, + id TEXT NOT NULL, + ts INTEGER NOT NULL CHECK (ts >= 0), + record_json TEXT NOT NULL + ); + + CREATE INDEX IF NOT EXISTS usage_tool_invocations_ts + ON usage_tool_invocations(ts DESC, id); + + CREATE TABLE IF NOT EXISTS usage_pricing_authority ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + revision INTEGER NOT NULL CHECK (revision >= 0) + ); + + CREATE TABLE IF NOT EXISTS usage_pricing_overrides ( + model_key TEXT PRIMARY KEY, + record_json TEXT NOT NULL + ); + `); +} diff --git a/packages/storage/src/sqlite-usage-store.ts b/packages/storage/src/sqlite-usage-store.ts new file mode 100644 index 0000000000..55db736ecc --- /dev/null +++ b/packages/storage/src/sqlite-usage-store.ts @@ -0,0 +1,814 @@ +import { createHash } from 'node:crypto'; +import { readFile } from 'node:fs/promises'; +import { join, resolve } from 'node:path'; +import type { + PricingConfig, + UsageBucket, + UsageGroupBy, + UsageLogRow, + UsageQuery, + UsageSummaryV2, +} from '@maka/core/usage-stats/types'; +import { + canonicalPricingConfigsEqual, + comparePricingModelKeys, + normalizePricingConfig, + normalizePricingModelKey, +} from '@maka/core/usage-stats/pricing'; +import { + createPricingStore, + PricingRevisionConflictError, + PricingStoreClosedError, + PricingStoreNotLoadedError, + PricingStorePublicationError, + PricingValidationError, + type CreatePricingStoreOptions, + type PricingMutationResult, + type PricingSnapshot, + type PricingStore, +} from './pricing-store.js'; +import { + acquireOperationalStateDatabase, + completeOperationalStoreCutover, + type OperationalStateDatabaseLease, + type OperationalStoreCutoverFailpoint, +} from './operational-state-store.js'; +import { + decodePersistedLlmCallRecord, + decodePersistedToolInvocationRecord, + decodeTelemetryFile, + emptyTelemetryFile, + type PersistedLlmCallRecord, + type PersistedToolInvocationRecord, + type TelemetryFile, +} from './telemetry-file-schema.js'; +import { + resolveRange, + TelemetryQueryValidationError, + TelemetryRepoClosedError, + TelemetryRepoNotLoadedError, + TelemetryRepoPublicationError, + type CreateTelemetryRepoOptions, + type TelemetryRepo, + type ToolUsageQuery, +} from './telemetry-repo.js'; + +export interface CreateSqliteUsageStoreOptions { + readonly failpoint?: (point: OperationalStoreCutoverFailpoint) => void; +} + +export function createSqliteTelemetryRepo( + workspaceRoot: string, + options: CreateTelemetryRepoOptions & CreateSqliteUsageStoreOptions = {}, +): TelemetryRepo { + return new SqliteTelemetryRepo( + workspaceRoot, + options.createIfMissing ?? true, + options.managePricing ?? true, + options.failpoint, + ); +} + +export function createSqlitePricingStore( + workspaceRoot: string, + options: CreatePricingStoreOptions & CreateSqliteUsageStoreOptions = {}, +): PricingStore { + return new SqlitePricingStore( + workspaceRoot, + options.createIfMissing ?? true, + options.initialOverrides ?? [], + options.failpoint, + ); +} + +class SqliteTelemetryRepo implements TelemetryRepo { + readonly #root: string; + readonly #lease: OperationalStateDatabaseLease; + readonly #pricingStore: PricingStore | undefined; + readonly #failpoint: ((point: OperationalStoreCutoverFailpoint) => void) | undefined; + #loaded = false; + #state: 'open' | 'draining' | 'closed' = 'open'; + #queue: Promise = Promise.resolve(); + #loadPromise: Promise | undefined; + #closePromise: Promise | undefined; + + constructor( + workspaceRoot: string, + createIfMissing: boolean, + managePricing: boolean, + failpoint?: (point: OperationalStoreCutoverFailpoint) => void, + ) { + this.#root = resolve(workspaceRoot); + this.#lease = acquireOperationalStateDatabase(this.#root); + this.#failpoint = failpoint; + this.#pricingStore = managePricing + ? createSqlitePricingStore(this.#root, { createIfMissing, failpoint }) + : undefined; + } + + load(): Promise { + if (this.#loaded) return Promise.resolve(); + this.assertOpen(); + if (this.#loadPromise) return this.#loadPromise; + const operation = importLegacyUsageState(this.#root, this.#lease, this.#failpoint).then( + async () => { + if (this.#pricingStore) await this.#pricingStore.load(); + this.#loaded = true; + }, + ); + this.#loadPromise = operation; + void operation.catch(() => { + if (this.#state === 'open' && this.#loadPromise === operation) { + this.#loadPromise = undefined; + } + }); + return operation; + } + + insertLlmCall(record: PersistedLlmCallRecord): Promise { + let admitted: PersistedLlmCallRecord; + try { + admitted = decodePersistedLlmCallRecord(record); + } catch (error) { + return Promise.reject(error); + } + return this.enqueueMutation(() => { + this.#lease.database + .prepare(` + INSERT INTO usage_llm_calls(storage_key, id, ts, record_json) + VALUES (?, ?, ?, ?) + ON CONFLICT(storage_key) DO UPDATE SET + id = excluded.id, + ts = excluded.ts, + record_json = excluded.record_json + `) + .run(usageIdentityKey(admitted.id), admitted.id, admitted.ts, JSON.stringify(admitted)); + }); + } + + insertToolInvocation(record: PersistedToolInvocationRecord): Promise { + let admitted: PersistedToolInvocationRecord; + try { + admitted = decodePersistedToolInvocationRecord(record); + } catch (error) { + return Promise.reject(error); + } + return this.enqueueMutation(() => { + this.#lease.database + .prepare(` + INSERT INTO usage_tool_invocations(storage_key, id, ts, record_json) + VALUES (?, ?, ?, ?) + ON CONFLICT(storage_key) DO UPDATE SET + id = excluded.id, + ts = excluded.ts, + record_json = excluded.record_json + `) + .run(usageIdentityKey(admitted.id), admitted.id, admitted.ts, JSON.stringify(admitted)); + }); + } + + summary(query: UsageQuery): UsageSummaryV2 { + this.assertReady(); + const { from, to } = resolveRange(query.range); + const rows = this.filteredUsageRows(query, from, to); + return detached({ + range: { from, to }, + totalRequests: rows.length, + totalCostUsd: sum(rows.map((row) => row.costUsd)), + totalTokens: { + input: sum(rows.map((row) => row.inputTokens)), + output: sum(rows.map((row) => row.outputTokens)), + cacheMiss: sum(rows.map((row) => row.cacheMissInputTokens)), + cacheRead: sum(rows.map((row) => row.cacheHitInputTokens)), + cacheWrite: sum(rows.map((row) => row.cacheWriteInputTokens)), + reasoning: sum(rows.map((row) => row.reasoningTokens)), + total: sum(rows.map((row) => row.totalTokens)), + }, + cacheHitRequests: rows.filter((row) => row.cacheHitInputTokens > 0).length, + cacheCreateRequests: rows.filter((row) => row.cacheWriteInputTokens > 0).length, + errorRequests: rows.filter((row) => row.status === 'error').length, + }); + } + + buckets(query: UsageQuery, groupBy: UsageGroupBy): UsageBucket[] { + this.assertReady(); + const { from, to } = resolveRange(query.range); + if (groupBy === 'tool') { + return detached(toolBuckets(this.filteredToolRows(query, from, to))); + } + const groups = new Map(); + for (const row of this.filteredUsageRows(query, from, to)) { + const key = bucketKey(row, groupBy); + const group = groups.get(key); + if (group) group.push(row); + else groups.set(key, [row]); + } + return detached( + [...groups.entries()] + .map(([key, rows]) => usageBucket(key, rows)) + .sort((left, right) => right.requests - left.requests), + ); + } + + logs(query: UsageQuery, offset = 0, limit = 100): { rows: UsageLogRow[]; total: number } { + this.assertReady(); + if (query.toolName !== undefined) { + throw new TelemetryQueryValidationError('toolName is not applicable to LLM logs'); + } + const { from, to } = resolveRange(query.range); + const rows = this.filteredUsageRows(query, from, to).sort((left, right) => right.ts - left.ts); + return detached({ + rows: rows.slice(offset, offset + limit).map(toUsageLogRow), + total: rows.length, + }); + } + + toolLogs( + query: ToolUsageQuery, + offset = 0, + limit = 100, + ): { rows: PersistedToolInvocationRecord[]; total: number } { + this.assertReady(); + assertToolUsageQuery(query); + const { from, to } = resolveRange(query.range); + const rows = this.filteredToolRows(query, from, to).sort((left, right) => right.ts - left.ts); + return detached({ rows: rows.slice(offset, offset + limit), total: rows.length }); + } + + latestLlmRuntimeProbe(connectionSlug: string, modelId?: string): UsageLogRow | undefined { + return this.logs({ range: 'all', connectionSlug, ...(modelId ? { modelId } : {}) }, 0, 1) + .rows[0]; + } + + listPricingOverrides(): PricingConfig[] { + return this.requireManagedPricing() + .snapshot() + .overrides.map((item) => ({ ...item })); + } + + async upsertPricing(pricing: PricingConfig): Promise { + const store = this.requireManagedPricing(); + await store.upsert(store.snapshot().revision, pricing); + } + + async deletePricing(modelKey: string): Promise { + const store = this.requireManagedPricing(); + await store.delete(store.snapshot().revision, modelKey); + } + + legacyPricingOverrides(): readonly unknown[] { + this.assertReady(); + return []; + } + + publishCanonical(): Promise { + return this.flush(); + } + + async flush(): Promise { + this.assertLoaded(); + await this.#queue; + } + + close(): Promise { + if (this.#closePromise) return this.#closePromise; + this.#state = 'draining'; + this.#closePromise = Promise.allSettled([ + this.#loadPromise ?? Promise.resolve(), + this.#queue, + this.#pricingStore?.close() ?? Promise.resolve(), + ]) + .then((results) => { + const failed = results.find((result) => result.status === 'rejected'); + if (failed?.status === 'rejected') throw failed.reason; + }) + .finally(() => { + this.#state = 'closed'; + this.#lease.close(); + }); + return this.#closePromise; + } + + private filteredUsageRows(query: UsageQuery, from: number, to: number) { + return this.readLlmRows().filter((row) => { + if (row.ts < from || row.ts > to) return false; + if (query.connectionSlug && row.connectionSlug !== query.connectionSlug) return false; + if (query.providerId && row.providerId !== query.providerId) return false; + if (query.modelId && row.modelId !== query.modelId) return false; + if (query.status && query.status !== 'all' && row.status !== query.status) return false; + return true; + }); + } + + private filteredToolRows(query: UsageQuery | ToolUsageQuery, from: number, to: number) { + return this.readToolRows().filter((row) => { + if (row.ts < from || row.ts > to) return false; + if (query.toolName && row.toolName !== query.toolName) return false; + if (query.status && query.status !== 'all' && row.status !== query.status) return false; + return true; + }); + } + + private readLlmRows(): PersistedLlmCallRecord[] { + return ( + this.#lease.database.prepare('SELECT record_json FROM usage_llm_calls').all() as Array<{ + record_json: string; + }> + ).map((row) => decodePersistedLlmCallRecord(JSON.parse(row.record_json))); + } + + private readToolRows(): PersistedToolInvocationRecord[] { + return ( + this.#lease.database + .prepare('SELECT record_json FROM usage_tool_invocations') + .all() as Array<{ record_json: string }> + ).map((row) => decodePersistedToolInvocationRecord(JSON.parse(row.record_json))); + } + + private enqueueMutation(operation: () => void): Promise { + this.assertReady(); + const accepted = this.#queue.then(() => { + try { + this.#lease.transaction('write', operation); + } catch (cause) { + throw new TelemetryRepoPublicationError(false, { cause }); + } + }); + this.#queue = accepted.catch(() => undefined); + return accepted; + } + + private requireManagedPricing(): PricingStore { + this.assertReady(); + if (!this.#pricingStore) { + throw new Error('Telemetry repository does not own the compatibility pricing facade'); + } + return this.#pricingStore; + } + + private assertLoaded(): void { + if (!this.#loaded) throw new TelemetryRepoNotLoadedError(); + } + + private assertOpen(): void { + if (this.#state !== 'open') throw new TelemetryRepoClosedError(); + } + + private assertReady(): void { + this.assertOpen(); + this.assertLoaded(); + } +} + +class SqlitePricingStore implements PricingStore { + readonly #root: string; + readonly #lease: OperationalStateDatabaseLease; + readonly #initialOverrides: readonly unknown[]; + readonly #failpoint: ((point: OperationalStoreCutoverFailpoint) => void) | undefined; + #loaded = false; + #state: 'open' | 'draining' | 'closed' = 'open'; + #queue: Promise = Promise.resolve(); + #loadPromise: Promise | undefined; + #closePromise: Promise | undefined; + + constructor( + workspaceRoot: string, + createIfMissing: boolean, + initialOverrides: readonly unknown[], + failpoint?: (point: OperationalStoreCutoverFailpoint) => void, + ) { + this.#root = resolve(workspaceRoot); + this.#lease = acquireOperationalStateDatabase(this.#root); + this.#initialOverrides = initialOverrides; + this.#failpoint = failpoint; + } + + load(): Promise { + if (this.#loaded) return Promise.resolve(); + this.assertOpen(); + if (this.#loadPromise) return this.#loadPromise; + const operation = importLegacyUsageState( + this.#root, + this.#lease, + this.#failpoint, + this.#initialOverrides, + ).then(() => { + this.#loaded = true; + }); + this.#loadPromise = operation; + void operation.catch(() => { + if (this.#state === 'open' && this.#loadPromise === operation) { + this.#loadPromise = undefined; + } + }); + return operation; + } + + snapshot(): PricingSnapshot { + this.assertReady(); + return readPricingSnapshot(this.#lease); + } + + upsert(expectedRevision: number, pricing: PricingConfig): Promise { + assertRevision(expectedRevision, 'expectedRevision'); + const normalized = normalizePricingConfig(pricing); + if (!normalized.ok) throw new PricingValidationError(normalized.error); + return this.enqueueMutation(expectedRevision, (current) => { + const existing = current.find((item) => item.modelKey === normalized.value.modelKey); + if (existing && canonicalPricingConfigsEqual(existing, normalized.value)) return current; + return [ + ...current.filter((item) => item.modelKey !== normalized.value.modelKey), + normalized.value, + ].sort((left, right) => comparePricingModelKeys(left.modelKey, right.modelKey)); + }); + } + + delete(expectedRevision: number, modelKey: string): Promise { + assertRevision(expectedRevision, 'expectedRevision'); + const normalized = normalizePricingModelKey(modelKey); + if (!normalized.ok) throw new PricingValidationError(normalized.error); + return this.enqueueMutation(expectedRevision, (current) => + current.some((item) => item.modelKey === normalized.value) + ? current.filter((item) => item.modelKey !== normalized.value) + : current, + ); + } + + async flush(): Promise { + this.assertLoaded(); + await this.#queue; + } + + beginDrain(): Promise { + if (this.#state === 'open') this.#state = 'draining'; + return this.flush(); + } + + close(): Promise { + if (this.#closePromise) return this.#closePromise; + this.#state = 'draining'; + this.#closePromise = Promise.allSettled([this.#loadPromise ?? Promise.resolve(), this.#queue]) + .then((results) => { + const failed = results.find((result) => result.status === 'rejected'); + if (failed?.status === 'rejected') throw failed.reason; + }) + .finally(() => { + this.#state = 'closed'; + this.#lease.close(); + }); + return this.#closePromise; + } + + private enqueueMutation( + expectedRevision: number, + mutate: (current: readonly Readonly[]) => readonly Readonly[], + ): Promise { + this.assertReady(); + const operation = this.#queue.then(() => { + try { + return this.#lease.transaction('write', () => { + const current = readPricingSnapshot(this.#lease); + if (current.revision !== expectedRevision) { + throw new PricingRevisionConflictError(expectedRevision, current.revision); + } + const overrides = mutate(current.overrides); + if (overrides === current.overrides) { + return { committed: false, changed: false, snapshot: current }; + } + if (current.revision === Number.MAX_SAFE_INTEGER) { + throw new PricingValidationError( + 'revision cannot advance beyond Number.MAX_SAFE_INTEGER', + ); + } + const revision = current.revision + 1; + this.#lease.database.prepare('DELETE FROM usage_pricing_overrides').run(); + const insert = this.#lease.database.prepare(` + INSERT INTO usage_pricing_overrides(model_key, record_json) + VALUES (?, ?) + `); + for (const override of overrides) { + insert.run(override.modelKey, JSON.stringify(override)); + } + this.#lease.database + .prepare('UPDATE usage_pricing_authority SET revision = ? WHERE singleton = 1') + .run(revision); + return { + committed: true, + changed: true, + snapshot: freezePricingSnapshot(revision, overrides), + }; + }); + } catch (error) { + if ( + error instanceof PricingRevisionConflictError || + error instanceof PricingValidationError + ) { + throw error; + } + throw new PricingStorePublicationError({ cause: error }); + } + }); + this.#queue = operation.then( + () => undefined, + () => undefined, + ); + return operation; + } + + private assertLoaded(): void { + if (!this.#loaded) throw new PricingStoreNotLoadedError(); + } + + private assertOpen(): void { + if (this.#state !== 'open') throw new PricingStoreClosedError(); + } + + private assertReady(): void { + this.assertOpen(); + this.assertLoaded(); + } +} + +interface LegacyUsageState { + readonly telemetry: TelemetryFile; + readonly pricing: PricingSnapshot; + readonly fingerprint: string; +} + +async function importLegacyUsageState( + root: string, + lease: OperationalStateDatabaseLease, + failpoint?: (point: OperationalStoreCutoverFailpoint) => void, + fallbackPricing: readonly unknown[] = [], +): Promise { + const legacy = await readLegacyUsageState(root, fallbackPricing); + completeOperationalStoreCutover(lease, { + storeName: 'usage_pricing', + sourcePath: root, + sourceFingerprint: legacy.fingerprint, + failpoint, + importAndValidate: (database) => { + const insertLlm = database.prepare(` + INSERT INTO usage_llm_calls(storage_key, id, ts, record_json) + VALUES (?, ?, ?, ?) + ON CONFLICT(storage_key) DO UPDATE SET + id = excluded.id, + ts = excluded.ts, + record_json = excluded.record_json + `); + for (const record of legacy.telemetry.usageRecords) { + insertLlm.run(usageIdentityKey(record.id), record.id, record.ts, JSON.stringify(record)); + } + const insertTool = database.prepare(` + INSERT INTO usage_tool_invocations(storage_key, id, ts, record_json) + VALUES (?, ?, ?, ?) + ON CONFLICT(storage_key) DO UPDATE SET + id = excluded.id, + ts = excluded.ts, + record_json = excluded.record_json + `); + for (const record of legacy.telemetry.toolInvocations) { + insertTool.run(usageIdentityKey(record.id), record.id, record.ts, JSON.stringify(record)); + } + database + .prepare(` + INSERT INTO usage_pricing_authority(singleton, revision) + VALUES (1, ?) + ON CONFLICT(singleton) DO UPDATE SET revision = excluded.revision + `) + .run(legacy.pricing.revision); + const insertPricing = database.prepare(` + INSERT INTO usage_pricing_overrides(model_key, record_json) + VALUES (?, ?) + ON CONFLICT(model_key) DO UPDATE SET record_json = excluded.record_json + `); + for (const override of legacy.pricing.overrides) { + insertPricing.run(override.modelKey, JSON.stringify(override)); + } + const counts = { + llm: countRows(database, 'usage_llm_calls'), + tools: countRows(database, 'usage_tool_invocations'), + pricing: countRows(database, 'usage_pricing_overrides'), + }; + if ( + counts.llm !== legacy.telemetry.usageRecords.length || + counts.tools !== legacy.telemetry.toolInvocations.length || + counts.pricing !== legacy.pricing.overrides.length + ) { + throw new Error('Usage/pricing cutover row-count validation failed'); + } + return counts; + }, + }); +} + +async function readLegacyUsageState( + root: string, + fallbackPricing: readonly unknown[], +): Promise { + const telemetryPath = join(root, 'telemetry.json'); + const telemetryText = await readOptionalText(telemetryPath); + const decoded = + telemetryText === undefined + ? { file: emptyTelemetryFile(), legacyPricingOverrides: fallbackPricing } + : decodeTelemetryFile(JSON.parse(telemetryText)); + const pricingPath = join(root, 'pricing.json'); + const pricingText = await readOptionalText(pricingPath); + const pricingStore = createPricingStore(root, { + createIfMissing: false, + initialOverrides: decoded.legacyPricingOverrides, + }); + await pricingStore.load(); + const pricing = pricingStore.snapshot(); + await pricingStore.close(); + const fingerprint = `sha256:${createHash('sha256') + .update(telemetryText === undefined ? 'telemetry:missing' : `telemetry:${telemetryText}`) + .update('\0') + .update(pricingText === undefined ? 'pricing:missing' : `pricing:${pricingText}`) + .digest('hex')}`; + return { telemetry: decoded.file, pricing, fingerprint }; +} + +async function readOptionalText(path: string): Promise { + try { + return await readFile(path, 'utf8'); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return undefined; + throw error; + } +} + +function readPricingSnapshot(lease: OperationalStateDatabaseLease): PricingSnapshot { + const authority = lease.database + .prepare('SELECT revision FROM usage_pricing_authority WHERE singleton = 1') + .get() as { revision?: unknown } | undefined; + if (!authority || !Number.isSafeInteger(authority.revision)) { + throw new PricingValidationError('SQLite pricing authority is missing or invalid'); + } + const overrides = ( + lease.database + .prepare('SELECT record_json FROM usage_pricing_overrides ORDER BY model_key') + .all() as Array<{ record_json: string }> + ).map((row) => { + const normalized = normalizePricingConfig(JSON.parse(row.record_json)); + if (!normalized.ok) throw new PricingValidationError(normalized.error); + return normalized.value; + }); + return freezePricingSnapshot(authority.revision as number, overrides); +} + +function freezePricingSnapshot( + revision: number, + overrides: readonly Readonly[], +): PricingSnapshot { + return Object.freeze({ + revision, + overrides: Object.freeze(overrides.map((value) => Object.freeze({ ...value }))), + }); +} + +function countRows( + database: OperationalStateDatabaseLease['database'], + table: 'usage_llm_calls' | 'usage_tool_invocations' | 'usage_pricing_overrides', +): number { + const row = database.prepare(`SELECT COUNT(*) AS count FROM ${table}`).get() as { + count?: unknown; + }; + if (!Number.isSafeInteger(row.count)) throw new Error(`Invalid row count for ${table}`); + return row.count as number; +} + +function usageIdentityKey(id: string): string { + return createHash('sha256').update(JSON.stringify(id)).digest('hex'); +} + +function toUsageLogRow(row: PersistedLlmCallRecord): UsageLogRow { + return { + id: row.id, + ts: row.ts, + ...(row.callKind ? { callKind: row.callKind } : {}), + ...(row.callId ? { callId: row.callId } : {}), + ...(row.connectionSlug ? { connectionSlug: row.connectionSlug } : {}), + providerId: row.providerId, + modelId: row.modelId, + inputTokens: row.inputTokens, + outputTokens: row.outputTokens, + cacheMissTokens: row.cacheMissInputTokens, + cacheReadTokens: row.cacheHitInputTokens, + cacheWriteTokens: row.cacheWriteInputTokens, + ...(row.cacheMissInputSource ? { cacheMissInputSource: row.cacheMissInputSource } : {}), + reasoningTokens: row.reasoningTokens, + totalTokens: row.totalTokens, + costUsd: row.costUsd, + latencyMs: row.latencyMs, + status: row.status, + ...(row.errorClass ? { errorClass: row.errorClass } : {}), + ...(row.sessionId ? { sessionId: row.sessionId } : {}), + ...(row.turnId ? { turnId: row.turnId } : {}), + ...(row.systemPromptHash ? { systemPromptHash: row.systemPromptHash } : {}), + ...(row.prefixHash ? { prefixHash: row.prefixHash } : {}), + ...(row.prefixChangeReason ? { prefixChangeReason: row.prefixChangeReason } : {}), + ...(row.requestShapeHash ? { requestShapeHash: row.requestShapeHash } : {}), + ...(row.requestShapeChangeReason + ? { requestShapeChangeReason: row.requestShapeChangeReason } + : {}), + ...(row.toolSchemaChangeReason ? { toolSchemaChangeReason: row.toolSchemaChangeReason } : {}), + ...(row.toolAvailability ? { toolAvailability: row.toolAvailability } : {}), + ...(row.promptSegments ? { promptSegments: row.promptSegments } : {}), + ...(row.contextBudget ? { contextBudget: row.contextBudget } : {}), + }; +} + +function bucketKey(row: PersistedLlmCallRecord, groupBy: UsageGroupBy): string { + switch (groupBy) { + case 'provider': + return row.providerId; + case 'model': + return `${row.providerId}:${row.modelId}`; + case 'day': + return row.date; + case 'hour': + return String(Math.floor(row.ts / (60 * 60 * 1000))); + case 'tool': + return ''; + } +} + +function usageBucket(key: string, rows: readonly PersistedLlmCallRecord[]): UsageBucket { + const errors = rows.filter((row) => row.status === 'error').length; + return { + key, + label: key, + requests: rows.length, + inputTokens: sum(rows.map((row) => row.inputTokens)), + outputTokens: sum(rows.map((row) => row.outputTokens)), + cacheMissTokens: sum(rows.map((row) => row.cacheMissInputTokens)), + cacheReadTokens: sum(rows.map((row) => row.cacheHitInputTokens)), + cacheWriteTokens: sum(rows.map((row) => row.cacheWriteInputTokens)), + reasoningTokens: sum(rows.map((row) => row.reasoningTokens)), + totalTokens: sum(rows.map((row) => row.totalTokens)), + costUsd: sum(rows.map((row) => row.costUsd)), + avgLatencyMs: rows.length ? Math.round(sum(rows.map((row) => row.latencyMs)) / rows.length) : 0, + errorRate: rows.length ? errors / rows.length : 0, + }; +} + +function toolBuckets(rows: readonly PersistedToolInvocationRecord[]): UsageBucket[] { + const groups = new Map(); + for (const row of rows) { + const group = groups.get(row.toolName); + if (group) group.push(row); + else groups.set(row.toolName, [row]); + } + return [...groups.entries()] + .map(([key, group]) => { + const errors = group.filter((row) => row.status === 'error').length; + const bytesIn = sum(group.map((row) => row.bytesIn)); + const bytesOut = sum(group.map((row) => row.bytesOut)); + return { + key, + label: key, + requests: group.length, + inputTokens: bytesIn, + outputTokens: bytesOut, + cacheMissTokens: 0, + cacheReadTokens: 0, + cacheWriteTokens: 0, + reasoningTokens: 0, + totalTokens: bytesIn + bytesOut, + costUsd: 0, + avgLatencyMs: group.length + ? Math.round(sum(group.map((row) => row.durationMs)) / group.length) + : 0, + errorRate: group.length ? errors / group.length : 0, + }; + }) + .sort((left, right) => right.requests - left.requests); +} + +function sum(values: readonly number[]): number { + return values.reduce((total, value) => total + value, 0); +} + +function assertToolUsageQuery(query: ToolUsageQuery): void { + if (Object.keys(query).some((key) => !['range', 'toolName', 'status'].includes(key))) { + throw new TelemetryQueryValidationError('tool logs accept only range, toolName, and status'); + } +} + +function assertRevision(value: unknown, label: string): asserts value is number { + if (!Number.isSafeInteger(value) || (value as number) < 0) { + throw new PricingValidationError(`${label} must be a nonnegative safe integer`); + } +} + +function detached(value: T): T { + return deepFreeze(structuredClone(value)); +} + +function deepFreeze(value: T): T { + if (!value || typeof value !== 'object' || Object.isFrozen(value)) return value; + Object.freeze(value); + for (const nested of Object.values(value)) deepFreeze(nested); + return value; +} diff --git a/packages/storage/src/usage-stores.ts b/packages/storage/src/usage-stores.ts index c345b30d12..26cae211e6 100644 --- a/packages/storage/src/usage-stores.ts +++ b/packages/storage/src/usage-stores.ts @@ -8,7 +8,6 @@ import type { } from '@maka/core/usage-stats/types'; import { throwDeduplicatedFailures } from './failure-utils.js'; import { - createPricingStore, PricingCommitUnknownError, PricingRevisionConflictError, PricingStoreClosedError, @@ -26,7 +25,6 @@ import { type StorageRootLease, } from './root-authority.js'; import { - createTelemetryRepo, TelemetryQueryValidationError, TelemetryRepoClosedError, TelemetryRepoNotLoadedError, @@ -36,6 +34,7 @@ import { type TelemetryRepo, type ToolUsageQuery, } from './telemetry-repo.js'; +import { createSqlitePricingStore, createSqliteTelemetryRepo } from './sqlite-usage-store.js'; const readerBrand: unique symbol = Symbol('InteractiveUsageStoresReader'); const writerBrand: unique symbol = Symbol('InteractiveUsageStoresWriter'); @@ -251,15 +250,14 @@ async function openRepos( root: string, createIfMissing: boolean, ): Promise<{ telemetry: TelemetryRepo; pricing: PricingStore }> { - const telemetry = createTelemetryRepo(root, { createIfMissing, managePricing: false }); + const telemetry = createSqliteTelemetryRepo(root, { createIfMissing, managePricing: false }); await telemetry.load(); - const pricing = createPricingStore(root, { + const pricing = createSqlitePricingStore(root, { createIfMissing, initialOverrides: telemetry.legacyPricingOverrides(), }); try { await pricing.load(); - if (createIfMissing) await telemetry.publishCanonical(); return { telemetry, pricing }; } catch (error) { const closed = await Promise.allSettled([telemetry.close(), pricing.close()]);