storage: migrate usage and pricing to SQLite

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;

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<typeof createFileCredentialStore>;
   connectionStore: ReturnType<typeof createConnectionStore>;
   settingsStore: ReturnType<typeof createSettingsStore>;
-  telemetryRepo: ReturnType<typeof createTelemetryRepo>;
+  telemetryRepo: ReturnType<typeof createSqliteTelemetryRepo>;
   ensureUsageReady: () => Promise<void>;
   keepSystemAwake: KeepSystemAwakeController;
   botRegistry: BotRegistry;

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<void> {
   }
   return mcpStartup;
 }
-const telemetryRepo = createTelemetryRepo(workspaceRoot);
+const telemetryRepo = createSqliteTelemetryRepo(workspaceRoot);
 const dailyReviewArchiveStore = createDailyReviewArchiveStore(workspaceRoot);
 const artifactStore = createArtifactStore(workspaceRoot);
 const deepResearchStore = createSqliteDeepResearchStore(workspaceRoot);

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<unknown>[]): 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: {

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 },
         ],
       );

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<void>): Promise<void> {
+  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<string, unknown> = {}) {
+  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<ReturnType<typeof createSqliteTelemetryRepo>['insertLlmCall']>[0];
+}