From 5122a3fc5fb0c59f2dcb4c01d655ceabca73d4e1 Mon Sep 17 00:00:00 2001 From: weishu Date: Wed, 5 Aug 2026 15:44:23 +0800 Subject: [PATCH] fix(hub): reindex usage events to backfill model attribution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #1359's session-model fallback only applies to events indexed after the upgrade. Clear usage_scan_state in a V19→V20 migration so the lazy re-index re-derives legacy rows instead of leaving them bucketed as "unknown". --- hub/src/store/index.ts | 21 +++++- hub/src/store/migration-v13.test.ts | 6 +- hub/src/store/migration-v15.test.ts | 6 +- hub/src/store/migration-v18.test.ts | 2 +- hub/src/store/migration-v19.test.ts | 101 ++++++++++++++++++++++++++++ 5 files changed, 128 insertions(+), 8 deletions(-) create mode 100644 hub/src/store/migration-v19.test.ts diff --git a/hub/src/store/index.ts b/hub/src/store/index.ts index 51064abe..b8b37f30 100644 --- a/hub/src/store/index.ts +++ b/hub/src/store/index.ts @@ -33,7 +33,7 @@ export { SessionStore } from './sessionStore' export { UserStore } from './userStore' export { UsageStore } from './usageStore' -const SCHEMA_VERSION: number = 19 +const SCHEMA_VERSION: number = 20 const REQUIRED_TABLES = [ 'sessions', 'machines', @@ -287,6 +287,7 @@ export class Store { 16: () => this.migrateFromV16ToV17(), 17: () => this.migrateFromV17ToV18(), 18: () => this.migrateFromV18ToV19(), + 19: () => this.migrateFromV19ToV20(), }) if (currentVersion === 0) { @@ -809,6 +810,24 @@ export class Store { `) } + private migrateFromV19ToV20(): void { + // tiann/hapi#1359 stamps a stable session-model fallback onto usage + // events that predate model attribution. That only helps events + // indexed after the upgrade, so clear the scan state once to force + // usageService's lazy re-index to re-derive every event. The events + // table is left alone: the re-index replaces each session's rows and + // reuses any previously indexed explicit models. + this.db.exec(` + CREATE TABLE IF NOT EXISTS usage_scan_state ( + session_id TEXT PRIMARY KEY, + message_epoch INTEGER NOT NULL DEFAULT 0, + last_seq INTEGER NOT NULL DEFAULT 0, + FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE + ); + DELETE FROM usage_scan_state; + `) + } + private getSessionColumnNames(): Set { const rows = this.db.prepare('PRAGMA table_info(sessions)').all() as Array<{ name: string }> return new Set(rows.map((row) => row.name)) diff --git a/hub/src/store/migration-v13.test.ts b/hub/src/store/migration-v13.test.ts index d53af0aa..4f128d09 100644 --- a/hub/src/store/migration-v13.test.ts +++ b/hub/src/store/migration-v13.test.ts @@ -10,7 +10,7 @@ describe('Store V12/V13→V14 schema reconciliation', () => { const store = new Store(':memory:') expect(tableExists(store, 'message_epochs')).toBe(true) expect(tableExists(store, 'session_scratchlist')).toBe(true) - expect(getUserVersion(store)).toBe(19) + expect(getUserVersion(store)).toBe(20) store.close() }) @@ -34,7 +34,7 @@ describe('Store V12/V13→V14 schema reconciliation', () => { store = new Store(dbPath) expect(tableExists(store, 'message_epochs')).toBe(true) expect(tableExists(store, 'session_scratchlist')).toBe(true) - expect(getUserVersion(store)).toBe(19) + expect(getUserVersion(store)).toBe(20) expect(store.messages.getMessageEpoch('session-1')).toBe(0) expect(store.messages.getMessages('session-1')).toHaveLength(1) } finally { @@ -72,7 +72,7 @@ describe('Store V12/V13→V14 schema reconciliation', () => { store = new Store(dbPath) expect(tableExists(store, 'message_epochs')).toBe(true) expect(tableExists(store, 'session_scratchlist')).toBe(true) - expect(getUserVersion(store)).toBe(19) + expect(getUserVersion(store)).toBe(20) expect(store.messages.getMessages('session-1')).toHaveLength(1) } finally { store?.close() diff --git a/hub/src/store/migration-v15.test.ts b/hub/src/store/migration-v15.test.ts index de05ed27..02607736 100644 --- a/hub/src/store/migration-v15.test.ts +++ b/hub/src/store/migration-v15.test.ts @@ -19,7 +19,7 @@ describe('Store V14→V15 migration: scratchlist attachments column', () => { expect(cols).toContain('attachments') expect(getColumns(store, 'usage_events')).toContain('last_input_tokens') expect(getColumns(store, 'usage_scan_state')).toContain('last_seq') - expect(getUserVersion(store)).toBe(19) + expect(getUserVersion(store)).toBe(20) store.close() }) @@ -40,7 +40,7 @@ describe('Store V14→V15 migration: scratchlist attachments column', () => { expect(cols).toContain('attachments') expect(getColumns(store, 'usage_events')).toContain('last_input_tokens') expect(getColumns(store, 'usage_scan_state')).toContain('last_seq') - expect(getUserVersion(store)).toBe(19) + expect(getUserVersion(store)).toBe(20) } finally { store?.close() rmSync(dir, { recursive: true, force: true }) @@ -60,7 +60,7 @@ describe('Store V14→V15 migration: scratchlist attachments column', () => { store2 = new Store(dbPath) const cols2 = getColumns(store2, 'session_scratchlist') expect(cols2).toEqual(cols1) - expect(getUserVersion(store2)).toBe(19) + expect(getUserVersion(store2)).toBe(20) } finally { store2?.close() store1?.close() diff --git a/hub/src/store/migration-v18.test.ts b/hub/src/store/migration-v18.test.ts index 473e8bf8..7a52f8c5 100644 --- a/hub/src/store/migration-v18.test.ts +++ b/hub/src/store/migration-v18.test.ts @@ -38,7 +38,7 @@ describe('Store V18->V19 migration: usage scan state', () => { const usageRows = internalDb.prepare('SELECT COUNT(*) AS count FROM usage_events').get() as { count: number } expect(table?.name).toBe('usage_scan_state') - expect(version.user_version).toBe(19) + expect(version.user_version).toBe(20) expect(usageRows.count).toBe(0) } finally { store?.close() diff --git a/hub/src/store/migration-v19.test.ts b/hub/src/store/migration-v19.test.ts new file mode 100644 index 00000000..3d5d5c4b --- /dev/null +++ b/hub/src/store/migration-v19.test.ts @@ -0,0 +1,101 @@ +import { describe, expect, it } from 'bun:test' +import { Database } from 'bun:sqlite' +import { mkdtempSync, rmSync } from 'node:fs' +import { join } from 'node:path' +import { tmpdir } from 'node:os' +import { Store } from './index' +import { getUsageSummary } from '../sync/usageService' + +describe('Store V19->V20 migration: usage re-index for model fallback', () => { + it('clears usage scan state while keeping derived events', () => { + const directory = mkdtempSync(join(tmpdir(), 'hapi-migration-v19-to-v20-')) + const dbPath = join(directory, 'test.db') + let store: Store | undefined + try { + store = new Store(dbPath) + store.close() + store = undefined + + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec(` + INSERT INTO sessions (id, created_at, updated_at) + VALUES ('session-1', 1, 1); + INSERT INTO usage_events ( + session_id, source_key, source_seq, created_at, agent, kind + ) VALUES ( + 'session-1', 'legacy-key', 1, 1, 'codex', 'cumulative' + ); + INSERT INTO usage_scan_state (session_id, message_epoch, last_seq) + VALUES ('session-1', 1, 42); + PRAGMA user_version = 19; + `) + db.close() + + store = new Store(dbPath) + const internalDb = (store as unknown as { db: Database }).db + const version = internalDb.prepare('PRAGMA user_version').get() as { user_version: number } + const scanRows = internalDb.prepare('SELECT COUNT(*) AS count FROM usage_scan_state').get() as { count: number } + const usageRows = internalDb.prepare('SELECT COUNT(*) AS count FROM usage_events').get() as { count: number } + + expect(version.user_version).toBe(20) + expect(scanRows.count).toBe(0) + expect(usageRows.count).toBe(1) + } finally { + store?.close() + rmSync(directory, { recursive: true, force: true }) + } + }) + + it('re-indexes legacy unknown-model events with the session model fallback', () => { + const directory = mkdtempSync(join(tmpdir(), 'hapi-migration-v19-to-v20-reindex-')) + const dbPath = join(directory, 'test.db') + let store: Store | undefined + try { + store = new Store(dbPath) + const session = store.sessions.getOrCreateSession( + 'session-1', + { path: '/tmp', host: 'test', flavor: 'codex' }, + null, + 'default', + 'gpt-test' + ) + store.messages.addMessage(session.id, { + role: 'agent', + content: { + type: 'codex', + data: { + type: 'token_count', + thread_id: 'thread-1', + turn_id: 'turn-1', + info: { + total_token_usage: { input_tokens: 100, output_tokens: 10 }, + last_token_usage: { input_tokens: 100, output_tokens: 10 } + } + } + } + }) + expect(getUsageSummary(store, 'default', 'all').totals.requests).toBe(1) + store.close() + store = undefined + + // Simulate the pre-#1359 state: derived rows without model + // attribution and a scan state that blocks any re-index. + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec(` + UPDATE usage_events SET model = NULL; + PRAGMA user_version = 19; + `) + db.close() + + store = new Store(dbPath) + const result = getUsageSummary(store, 'default', 'all') + expect(result.totals.requests).toBe(1) + expect(result.byModel).toEqual([ + expect.objectContaining({ key: 'gpt-test', totalTokens: 110 }) + ]) + } finally { + store?.close() + rmSync(directory, { recursive: true, force: true }) + } + }) +})