fix(hub): reindex usage events to backfill model attribution

#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".
This commit is contained in:
weishu
2026-08-05 15:44:23 +08:00
parent f157d29b20
commit 5122a3fc5f
5 changed files with 128 additions and 8 deletions
+20 -1
View File
@@ -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<string> {
const rows = this.db.prepare('PRAGMA table_info(sessions)').all() as Array<{ name: string }>
return new Set(rows.map((row) => row.name))
+3 -3
View File
@@ -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()
+3 -3
View File
@@ -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()
+1 -1
View File
@@ -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()
+101
View File
@@ -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 })
}
})
})