diff --git a/cli/src/agent/messageConverter.test.ts b/cli/src/agent/messageConverter.test.ts index 41ee4e4e..83db6be9 100644 --- a/cli/src/agent/messageConverter.test.ts +++ b/cli/src/agent/messageConverter.test.ts @@ -155,6 +155,8 @@ describe('convertAgentMessage', () => { expect(converted).toEqual({ type: 'token_count', model: 'kimi-k2.5', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache', info: { total: { inputTokens: 13879, @@ -200,7 +202,9 @@ describe('convertAgentMessage', () => { expect(converted).toMatchObject({ type: 'token_count', - model: null + model: null, + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache' }); }); it('returns null instead of echoing an unrecognized message shape', () => { diff --git a/cli/src/agent/messageConverter.ts b/cli/src/agent/messageConverter.ts index d9867be9..4df39570 100644 --- a/cli/src/agent/messageConverter.ts +++ b/cli/src/agent/messageConverter.ts @@ -1,4 +1,5 @@ import { randomUUID } from 'node:crypto'; +import { INCLUSIVE_INPUT_TOKEN_USAGE_MARKER, type InclusiveInputTokenUsageMarker } from '@hapi/protocol/usage'; import type { AgentMessage, PlanItem } from './types'; export type CodexMessage = @@ -7,6 +8,8 @@ export type CodexMessage = | { type: 'token_count'; model: string | null; + usageSchema: InclusiveInputTokenUsageMarker['usageSchema']; + inputTokenSemantics: InclusiveInputTokenUsageMarker['inputTokenSemantics']; info: { total: { inputTokens: number; @@ -57,6 +60,7 @@ export function convertAgentMessage(message: AgentMessage, model?: string | null return { type: 'token_count', model: typeof model === 'string' && model.trim() ? model.trim() : null, + ...INCLUSIVE_INPUT_TOKEN_USAGE_MARKER, info: { total: { inputTokens: message.inputTokens diff --git a/cli/src/codex/codexLocalLauncher.test.ts b/cli/src/codex/codexLocalLauncher.test.ts index 97e78fc0..785f0e64 100644 --- a/cli/src/codex/codexLocalLauncher.test.ts +++ b/cli/src/codex/codexLocalLauncher.test.ts @@ -411,7 +411,9 @@ describe('codexLocalLauncher', () => { expect(getModelReasoningEffort()).toBeNull(); expect(agentMessages).toContainEqual(expect.objectContaining({ type: 'token_count', - model: 'gpt-5.4' + model: 'gpt-5.4', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache' })); }); @@ -735,12 +737,18 @@ describe('codexLocalLauncher', () => { message as { type?: string } ).type === 'token_count') as Array>; expect(tokenMessages).toHaveLength(2); - expect(tokenMessages[0]).toMatchObject({ hapiUsageScope: 'imported-history' }); + expect(tokenMessages[0]).toMatchObject({ + hapiUsageScope: 'imported-history', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache' + }); expect(tokenMessages[0]).not.toHaveProperty('thread_id'); expect(tokenMessages[1]).toMatchObject({ threadId: 'codex-thread-import', thread_id: 'codex-thread-import', - hapiUsageScope: 'managed' + hapiUsageScope: 'managed', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache' }); }); diff --git a/cli/src/codex/codexRemoteLauncher.test.ts b/cli/src/codex/codexRemoteLauncher.test.ts index ef09648a..72472287 100644 --- a/cli/src/codex/codexRemoteLauncher.test.ts +++ b/cli/src/codex/codexRemoteLauncher.test.ts @@ -2695,6 +2695,8 @@ describe('codexRemoteLauncher', () => { expect(codexMessages).toContainEqual(expect.objectContaining({ type: 'token_count', thread_id: 'thread-1', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache', scope_role: 'parent', scope: expect.objectContaining({ role: 'parent', diff --git a/cli/src/codex/utils/appServerEventConverter.test.ts b/cli/src/codex/utils/appServerEventConverter.test.ts index c6c593cb..5567e056 100644 --- a/cli/src/codex/utils/appServerEventConverter.test.ts +++ b/cli/src/codex/utils/appServerEventConverter.test.ts @@ -643,6 +643,8 @@ describe('AppServerEventConverter', () => { }); expect(tokenEvents).toEqual([{ type: 'token_count', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache', thread_id: 'child-thread', turn_id: 'child-turn', info: { diff --git a/cli/src/codex/utils/appServerEventConverter.ts b/cli/src/codex/utils/appServerEventConverter.ts index a49a2241..50907a4a 100644 --- a/cli/src/codex/utils/appServerEventConverter.ts +++ b/cli/src/codex/utils/appServerEventConverter.ts @@ -1,3 +1,4 @@ +import { INCLUSIVE_INPUT_TOKEN_USAGE_MARKER } from '@hapi/protocol/usage'; import { logger } from '@/ui/logger'; type ConvertedEvent = { @@ -784,7 +785,7 @@ export class AppServerEventConverter { if (method === 'thread/tokenUsage/updated') { const info = asRecord(paramsRecord.tokenUsage ?? paramsRecord.token_usage ?? paramsRecord) ?? {}; - events.push(scoped({ type: 'token_count', info })); + events.push(scoped({ type: 'token_count', ...INCLUSIVE_INPUT_TOKEN_USAGE_MARKER, info })); return events; } diff --git a/cli/src/codex/utils/codexEventConverter.test.ts b/cli/src/codex/utils/codexEventConverter.test.ts index 4add86df..fa4f66e4 100644 --- a/cli/src/codex/utils/codexEventConverter.test.ts +++ b/cli/src/codex/utils/codexEventConverter.test.ts @@ -32,6 +32,27 @@ describe('convertCodexEvent', () => { expect(result?.userMessage).toBe('hello user'); }); + it('marks native token counts as inclusive of cached input', () => { + const info = { + total_token_usage: { + input_tokens: 120, + cached_input_tokens: 20, + output_tokens: 12 + } + }; + const result = convertCodexEvent({ + type: 'event_msg', + payload: { type: 'token_count', info } + }); + + expect(result?.messages?.[0]).toMatchObject({ + type: 'token_count', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache', + info + }); + }); + it('converts completed plan items into proposed plan messages', () => { const result = convertCodexEvent({ type: 'event_msg', diff --git a/cli/src/codex/utils/codexEventConverter.ts b/cli/src/codex/utils/codexEventConverter.ts index 2d10397c..c26dec50 100644 --- a/cli/src/codex/utils/codexEventConverter.ts +++ b/cli/src/codex/utils/codexEventConverter.ts @@ -1,4 +1,5 @@ import { randomUUID } from 'node:crypto'; +import { INCLUSIVE_INPUT_TOKEN_USAGE_MARKER, type InclusiveInputTokenUsageMarker } from '@hapi/protocol/usage'; import { z } from 'zod'; import { logger } from '@/ui/logger'; @@ -30,6 +31,8 @@ export type CodexMessage = { type: 'token_count'; info: Record; id: string; + usageSchema: InclusiveInputTokenUsageMarker['usageSchema']; + inputTokenSemantics: InclusiveInputTokenUsageMarker['inputTokenSemantics']; } | { type: 'tool-call'; name: string; @@ -214,6 +217,7 @@ export function convertCodexEvent(rawEvent: unknown): CodexConversionResult | nu return { messages: [{ type: 'token_count', + ...INCLUSIVE_INPUT_TOKEN_USAGE_MARKER, info, id: randomUUID() }] diff --git a/cli/src/kimi/utils/kimiWireScanner.test.ts b/cli/src/kimi/utils/kimiWireScanner.test.ts index 25aad64e..d139754d 100644 --- a/cli/src/kimi/utils/kimiWireScanner.test.ts +++ b/cli/src/kimi/utils/kimiWireScanner.test.ts @@ -81,6 +81,8 @@ describe('convertKimiWireEvent', () => { message: { type: 'token_count', model: 'kimi-k2.5', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache', info: { total: { inputTokens: 160, diff --git a/cli/src/kimi/utils/kimiWireScanner.ts b/cli/src/kimi/utils/kimiWireScanner.ts index 890e43d0..771f498e 100644 --- a/cli/src/kimi/utils/kimiWireScanner.ts +++ b/cli/src/kimi/utils/kimiWireScanner.ts @@ -1,4 +1,5 @@ import { randomUUID } from 'node:crypto'; +import { INCLUSIVE_INPUT_TOKEN_USAGE_MARKER } from '@hapi/protocol/usage'; import { open, stat } from 'node:fs/promises'; import { BaseSessionScanner, type SessionFileScanResult, type SessionFileScanStats } from '@/modules/common/session/BaseSessionScanner'; import { logger } from '@/ui/logger'; @@ -150,6 +151,7 @@ export function convertKimiWireEvent(event: KimiWireEvent, model?: string | null message: { type: 'token_count', model: typeof model === 'string' && model.trim() ? model.trim() : null, + ...INCLUSIVE_INPUT_TOKEN_USAGE_MARKER, info: { total: { inputTokens: inputOther + cacheRead + cacheCreation, diff --git a/cli/src/modules/common/piSessions.test.ts b/cli/src/modules/common/piSessions.test.ts index 90aed461..1526ecd6 100644 --- a/cli/src/modules/common/piSessions.test.ts +++ b/cli/src/modules/common/piSessions.test.ts @@ -56,6 +56,17 @@ describe('local Pi sessions', () => { 'pi:pi-session-1:custom-1:custom_message', 'pi:pi-session-1:compact-1:compaction' ]) + expect(sessions[0]?.messages.find((message) => message.localId.endsWith(':usage'))).toMatchObject({ + content: { + content: { + data: { + type: 'token_count', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache' + } + } + } + }) expect(sessions[0]?.messages[0]).toMatchObject({ entryId: 'user-1', parentEntryId: null, diff --git a/cli/src/modules/common/piSessions.ts b/cli/src/modules/common/piSessions.ts index 5666d806..c1359954 100644 --- a/cli/src/modules/common/piSessions.ts +++ b/cli/src/modules/common/piSessions.ts @@ -2,6 +2,7 @@ import { closeSync, openSync, readdirSync, readFileSync, readSync, statSync } fr import { basename, join } from 'node:path' import { homedir } from 'node:os' import { AGENT_MESSAGE_PAYLOAD_TYPE } from '@hapi/protocol' +import { INCLUSIVE_INPUT_TOKEN_USAGE_MARKER } from '@hapi/protocol/usage' import type { PiImportedMessage, PiImportedMessageContent, @@ -252,6 +253,7 @@ function convertMessageRecord( pushImportedMessage(result, sessionId, entryId, parentEntryId, createdAt, 'usage', importedAgent({ type: 'token_count', model: asString(message.model), + ...INCLUSIVE_INPUT_TOKEN_USAGE_MARKER, info: { total: { inputTokens: input + cacheRead + cacheWrite, diff --git a/hub/src/store/index.ts b/hub/src/store/index.ts index b8b37f30..e30007ac 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 = 20 +const SCHEMA_VERSION: number = 21 const REQUIRED_TABLES = [ 'sessions', 'machines', @@ -288,6 +288,7 @@ export class Store { 17: () => this.migrateFromV17ToV18(), 18: () => this.migrateFromV18ToV19(), 19: () => this.migrateFromV19ToV20(), + 20: () => this.migrateFromV20ToV21(), }) if (currentVersion === 0) { @@ -828,6 +829,16 @@ export class Store { `) } + private migrateFromV20ToV21(): void { + // Usage events and scan cursors are a derived index. v21 moves cache + // normalization to parse time, so every row must be rebuilt under the + // same inclusive-input invariant rather than mixing old and new rows. + this.db.exec(` + DELETE FROM usage_events; + 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 4f128d09..64df91f7 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(20) + expect(getUserVersion(store)).toBe(21) 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(20) + expect(getUserVersion(store)).toBe(21) 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(20) + expect(getUserVersion(store)).toBe(21) 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 02607736..aa931b96 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(20) + expect(getUserVersion(store)).toBe(21) 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(20) + expect(getUserVersion(store)).toBe(21) } 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(20) + expect(getUserVersion(store2)).toBe(21) } finally { store2?.close() store1?.close() diff --git a/hub/src/store/migration-v18.test.ts b/hub/src/store/migration-v18.test.ts index 7a52f8c5..91b49551 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(20) + expect(version.user_version).toBe(21) 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 index 3d5d5c4b..32a5e8bc 100644 --- a/hub/src/store/migration-v19.test.ts +++ b/hub/src/store/migration-v19.test.ts @@ -6,8 +6,8 @@ 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', () => { +describe('Store V19->current migration: usage re-index', () => { + it('clears stale derived usage rows and scan state', () => { const directory = mkdtempSync(join(tmpdir(), 'hapi-migration-v19-to-v20-')) const dbPath = join(directory, 'test.db') let store: Store | undefined @@ -37,9 +37,9 @@ describe('Store V19->V20 migration: usage re-index for model fallback', () => { 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(version.user_version).toBe(21) expect(scanRows.count).toBe(0) - expect(usageRows.count).toBe(1) + expect(usageRows.count).toBe(0) } finally { store?.close() rmSync(directory, { recursive: true, force: true }) diff --git a/hub/src/store/migration-v20.test.ts b/hub/src/store/migration-v20.test.ts new file mode 100644 index 00000000..4f59e8bc --- /dev/null +++ b/hub/src/store/migration-v20.test.ts @@ -0,0 +1,63 @@ +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 { getUsageSummary } from '../sync/usageService' +import { Store } from './index' + +describe('Store V20->V21 migration: usage semantics re-index', () => { + it('clears both derived tables, preserves messages, and lazily rebuilds idempotently', () => { + const directory = mkdtempSync(join(tmpdir(), 'hapi-migration-v20-to-v21-')) + 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: 'claude' }, + null, + 'default' + ) + store.messages.addMessage(session.id, { + role: 'agent', + content: { + type: 'output', + data: { + type: 'assistant', + message: { + id: 'usage-1', + usage: { input_tokens: 10, output_tokens: 2, cache_read_input_tokens: 80 } + } + } + } + }) + expect(getUsageSummary(store, 'default', 'all').totals.inputTokens).toBe(90) + store.close() + store = undefined + + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec('PRAGMA user_version = 20;') + db.close() + + store = new Store(dbPath) + const internalDb = (store as unknown as { db: Database }).db + const count = (table: string): number => (internalDb.prepare(`SELECT COUNT(*) AS count FROM ${table}`).get() as { count: number }).count + expect((internalDb.prepare('PRAGMA user_version').get() as { user_version: number }).user_version).toBe(21) + expect(count('usage_events')).toBe(0) + expect(count('usage_scan_state')).toBe(0) + expect(store.messages.getMessages(session.id)).toHaveLength(1) + + const first = getUsageSummary(store, 'default', 'all') + expect(first.totals.inputTokens).toBe(90) + expect(count('usage_events')).toBe(1) + expect(count('usage_scan_state')).toBe(1) + const second = getUsageSummary(store, 'default', 'all') + expect(second.totals).toEqual(first.totals) + expect(count('usage_events')).toBe(1) + } finally { + store?.close() + rmSync(directory, { recursive: true, force: true }) + } + }) +}) diff --git a/hub/src/sync/usageService.test.ts b/hub/src/sync/usageService.test.ts index c41dd1c6..70efa476 100644 --- a/hub/src/sync/usageService.test.ts +++ b/hub/src/sync/usageService.test.ts @@ -17,7 +17,7 @@ function addAgentMessage(store: Store, sessionId: string, content: unknown, crea } describe('usage service', () => { - it('deduplicates Claude stream fragments and normalizes cached input', () => { + it('normalizes historical Claude input that excludes cached tokens', () => { const store = new Store(':memory:') const session = store.sessions.getOrCreateSession( 'claude-usage-test', @@ -61,7 +61,7 @@ describe('usage service', () => { store.close() }) - it('uses the latest request as the baseline for a resumed Codex thread', () => { + it('normalizes historical Codex cumulative usage with inclusive input', () => { const store = new Store(':memory:') const session = store.sessions.getOrCreateSession( 'codex-usage-test', @@ -121,7 +121,7 @@ describe('usage service', () => { store.close() }) - it('treats ACP usage totals as per-request deltas', () => { + it('normalizes historical Kimi usage totals as inclusive per-request deltas', () => { const store = new Store(':memory:') const session = store.sessions.getOrCreateSession( 'kimi-usage-test', @@ -156,7 +156,7 @@ describe('usage service', () => { store.close() }) - it('counts normalized ACP cached input as processed input', () => { + it('recognizes the model-bearing historical generic ACP wire as inclusive', () => { const store = new Store(':memory:') const session = store.sessions.getOrCreateSession( 'acp-cache-usage-test', @@ -188,6 +188,243 @@ describe('usage service', () => { store.close() }) + it('falls back to legacy provenance when v1 usage semantics are malformed', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession( + 'malformed-generic-usage-test', + { path: '/tmp', host: 'test', flavor: 'opencode' }, + null, + 'default' + ) + + addAgentMessage(store, session.id, { + type: 'codex', + data: { + type: 'token_count', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'unknown', + info: { + total: { + inputTokens: 100, + outputTokens: 20, + cachedInputTokens: 40, + cacheWriteInputTokens: 10 + } + } + } + }) + + const result = getUsageSummary(store, 'default', 'all') + expect(result.totals).toEqual(expect.objectContaining({ + inputTokens: 150, + cacheReadTokens: 40, + cacheCreationTokens: 10, + totalTokens: 170 + })) + store.close() + }) + + it('repairs unmarked legacy generic ACP input that excluded cache', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession( + 'legacy-generic-usage-test', + { path: '/tmp', host: 'test', flavor: 'opencode' }, + null, + 'default' + ) + addAgentMessage(store, session.id, { + type: 'codex', + data: { + type: 'token_count', + info: { total: { inputTokens: 100, outputTokens: 5, cachedInputTokens: 80, cacheWriteInputTokens: 20 } } + } + }) + + const result = getUsageSummary(store, 'default', 'all') + expect(result.totals.inputTokens).toBe(200) + expect(result.totals.cacheReadTokens).toBe(80) + expect(result.totals.cacheCreationTokens).toBe(20) + expect(result.totals.uncachedTokens).toBe(125) + store.close() + }) + + it('normalizes mixed legacy and marked generic usage per message in one session', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession( + 'mixed-generic-usage-test', + { path: '/tmp', host: 'test', flavor: 'opencode' }, + null, + 'default' + ) + const messages = [ + { type: 'token_count', info: { total: { inputTokens: 100, outputTokens: 5, cachedInputTokens: 80 } } }, + { type: 'token_count', usageSchema: 'hapi.usage.v1', inputTokenSemantics: 'includes-cache', info: { total: { inputTokens: 180, outputTokens: 5, cachedInputTokens: 80 } } }, + { type: 'token_count', model: null, info: { total: { inputTokens: 180, outputTokens: 5, cachedInputTokens: 80 } } } + ] + for (const data of messages) addAgentMessage(store, session.id, { type: 'codex', data }) + + const result = getUsageSummary(store, 'default', 'all') + expect(result.totals.inputTokens).toBe(540) + expect(result.totals.outputTokens).toBe(15) + expect(result.totals.cacheReadTokens).toBe(240) + expect(result.totals.requests).toBe(3) + store.close() + }) + + it('uses the explicit v1 generic usage input semantics without double-counting cache', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession( + 'generic-v1-usage-test', + { path: '/tmp', host: 'test', flavor: 'opencode' }, + null, + 'default', + 'acp-model' + ) + + addAgentMessage(store, session.id, { + type: 'codex', + data: { + type: 'token_count', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache', + info: { + total: { + inputTokens: 150, + outputTokens: 20, + cachedInputTokens: 40, + cacheWriteInputTokens: 10 + } + } + } + }) + + const result = getUsageSummary(store, 'default', 'all') + expect(result.totals).toEqual(expect.objectContaining({ + inputTokens: 150, + outputTokens: 20, + cacheReadTokens: 40, + cacheCreationTokens: 10, + totalTokens: 170, + uncachedTokens: 130, + requests: 1 + })) + store.close() + }) + + it('normalizes cumulative total and last snapshots with the same declared semantics', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession( + 'codex-exclusive-cumulative-test', + { path: '/tmp', host: 'test', flavor: 'codex' }, + null, + 'default' + ) + const snapshots = [ + { + total: { inputTokens: 100, outputTokens: 10, cachedInputTokens: 80 }, + last: { inputTokens: 20, outputTokens: 2, cachedInputTokens: 10 } + }, + { + total: { inputTokens: 130, outputTokens: 15, cachedInputTokens: 100 }, + last: { inputTokens: 30, outputTokens: 5, cachedInputTokens: 20 } + } + ] + snapshots.forEach((info, index) => addAgentMessage(store, session.id, { + type: 'codex', + data: { + type: 'token_count', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'excludes-cache', + threadId: 'thread-exclusive', + turnId: `turn-${index}`, + info + } + })) + + const result = getUsageSummary(store, 'default', 'all') + expect(result.totals).toEqual(expect.objectContaining({ + inputTokens: 80, + outputTokens: 7, + cacheReadTokens: 30, + requests: 2 + })) + store.close() + }) + + it('treats a cumulative counter reset as a whole-snapshot reset', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession( + 'codex-whole-reset-test', + { path: '/tmp', host: 'test', flavor: 'codex' }, + null, + 'default' + ) + const snapshots = [ + { + total: { inputTokens: 100, outputTokens: 20, cachedInputTokens: 80 }, + last: { inputTokens: 100, outputTokens: 20, cachedInputTokens: 80 } + }, + { + // Only input/cache regressed; output alone still increased. + total: { inputTokens: 50, outputTokens: 70, cachedInputTokens: 40 }, + last: { inputTokens: 50, outputTokens: 7, cachedInputTokens: 40 } + } + ] + snapshots.forEach((info, index) => addAgentMessage(store, session.id, { + type: 'codex', + data: { + type: 'token_count', + threadId: 'thread-reset', + turnId: `turn-${index}`, + info + } + })) + + const result = getUsageSummary(store, 'default', 'all') + expect(result.totals).toEqual(expect.objectContaining({ + inputTokens: 150, + outputTokens: 27, + cacheReadTokens: 120, + requests: 2 + })) + store.close() + }) + + it('preserves primary usage and requests when the cache partition is impossible', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession( + 'invalid-cache-partition-test', + { path: '/tmp', host: 'test', flavor: 'opencode' }, + null, + 'default' + ) + addAgentMessage(store, session.id, { + type: 'codex', + data: { + type: 'usage', + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache', + inputTokens: 100, + outputTokens: 20, + cachedInputTokens: 120, + cacheWriteInputTokens: 5 + } + }) + + const result = getUsageSummary(store, 'default', 'all') + expect(result.totals).toEqual(expect.objectContaining({ + inputTokens: 100, + outputTokens: 20, + cacheReadTokens: 0, + cacheCreationTokens: 0, + totalTokens: 120, + uncachedTokens: 120, + requests: 1 + })) + expect(result.totals.sessions).toBe(1) + store.close() + }) + it('falls back to the session model for Codex events without a model', () => { const store = new Store(':memory:') const session = store.sessions.getOrCreateSession( diff --git a/hub/src/sync/usageService.ts b/hub/src/sync/usageService.ts index 6979667c..1dac14b4 100644 --- a/hub/src/sync/usageService.ts +++ b/hub/src/sync/usageService.ts @@ -25,6 +25,26 @@ function firstCount(record: RecordValue, ...keys: string[]): number { return 0 } +function normalizeInputTokens( + data: RecordValue, + inputTokens: number, + cacheReadTokens: number, + cacheCreationTokens: number, + legacySemantics: 'includes-cache' | 'excludes-cache' +): number { + // v1 generic usage messages make their input contract self-describing. + // Unknown/missing metadata intentionally falls back to the historical + // provider shape so already persisted transcripts remain readable. + const declaredSemantics = data.usageSchema === 'hapi.usage.v1' + && (data.inputTokenSemantics === 'includes-cache' || data.inputTokenSemantics === 'excludes-cache') + ? data.inputTokenSemantics + : null + const semantics = declaredSemantics ?? legacySemantics + return semantics === 'excludes-cache' + ? inputTokens + cacheReadTokens + cacheCreationTokens + : inputTokens +} + function sessionAgent(session: StoredSession): string { const metadata = asRecord(session.metadata) const flavor = metadata?.flavor @@ -67,7 +87,7 @@ function parseUsageEvent(session: StoredSession, message: StoredMessage): UsageE agent: 'claude', model, kind: 'delta', - inputTokens, + inputTokens: normalizeInputTokens(data, inputTokens, cacheReadTokens, cacheCreationTokens, 'excludes-cache'), outputTokens, cacheReadTokens, cacheCreationTokens, @@ -107,11 +127,11 @@ function parseUsageEvent(session: StoredSession, message: StoredMessage): UsageE ?? (data.type === 'usage' ? info : null) const total = cumulativeTotal ?? (agent === 'codex' ? last : asRecord(info.total) ?? info) if (!total) return null - const inputTokens = firstCount(total, 'inputTokens', 'input_tokens') + const rawInputTokens = firstCount(total, 'inputTokens', 'input_tokens') const outputTokens = firstCount(total, 'outputTokens', 'output_tokens') const cacheReadTokens = firstCount(total, 'cachedInputTokens', 'cached_input_tokens', 'cacheReadTokens', 'cache_read_input_tokens') const cacheCreationTokens = firstCount(total, 'cacheWriteInputTokens', 'cache_write_input_tokens', 'cacheCreationTokens', 'cache_creation_input_tokens') - if (inputTokens + outputTokens + cacheReadTokens + cacheCreationTokens <= 0) return null + if (rawInputTokens + outputTokens + cacheReadTokens + cacheCreationTokens <= 0) return null const threadId = explicitThreadId ?? session.id const scope = typeof data.scopeRole === 'string' ? data.scopeRole @@ -127,6 +147,40 @@ function parseUsageEvent(session: StoredSession, message: StoredMessage): UsageE const model = typeof data.model === 'string' && data.model.trim() ? data.model.trim() : null + // Codex/Kimi provider formats have always reported inclusive input. + // Imported Pi usage is known-inclusive, and for generic ACP an own + // `model` property is the only strong provenance for the unmarked + // inclusive wire introduced with the usage dashboard. Older ambiguous + // payloads are conservatively treated as cache-exclusive. + const legacyInputSemantics = agent === 'codex' + || agent === 'kimi' + || (agent === 'pi' && message.localId?.startsWith('pi:')) + || Object.prototype.hasOwnProperty.call(data, 'model') + ? 'includes-cache' + : 'excludes-cache' + const inputTokens = normalizeInputTokens( + data, + rawInputTokens, + cacheReadTokens, + cacheCreationTokens, + legacyInputSemantics + ) + const lastOutputTokens = last ? firstCount(last, 'outputTokens', 'output_tokens') : null + const lastCacheReadTokens = last + ? firstCount(last, 'cachedInputTokens', 'cached_input_tokens', 'cacheReadTokens', 'cache_read_input_tokens') + : null + const lastCacheCreationTokens = last + ? firstCount(last, 'cacheWriteInputTokens', 'cache_write_input_tokens', 'cacheCreationTokens', 'cache_creation_input_tokens') + : null + const lastInputTokens = last + ? normalizeInputTokens( + data, + firstCount(last, 'inputTokens', 'input_tokens'), + lastCacheReadTokens ?? 0, + lastCacheCreationTokens ?? 0, + legacyInputSemantics + ) + : null return { sessionId: session.id, sourceKey: isCumulative @@ -150,14 +204,10 @@ function parseUsageEvent(session: StoredSession, message: StoredMessage): UsageE outputTokens, cacheReadTokens, cacheCreationTokens, - lastInputTokens: last ? firstCount(last, 'inputTokens', 'input_tokens') : null, - lastOutputTokens: last ? firstCount(last, 'outputTokens', 'output_tokens') : null, - lastCacheReadTokens: last - ? firstCount(last, 'cachedInputTokens', 'cached_input_tokens', 'cacheReadTokens', 'cache_read_input_tokens') - : null, - lastCacheCreationTokens: last - ? firstCount(last, 'cacheWriteInputTokens', 'cache_write_input_tokens', 'cacheCreationTokens', 'cache_creation_input_tokens') - : null + lastInputTokens, + lastOutputTokens, + lastCacheReadTokens, + lastCacheCreationTokens } } @@ -240,9 +290,19 @@ function addTotals(target: Totals, inputTokens: number, outputTokens: number, ca target.requests += 1 } -function cumulativeDelta(current: number, previous: number | null, last: number | null): number { - if (previous === null) return last ?? current - return current >= previous ? current - previous : last ?? current +type UsageSnapshot = [number, number, number, number] + +function cumulativeSnapshotDelta( + current: UsageSnapshot, + previous: UsageSnapshot | null, + last: UsageSnapshot | null +): UsageSnapshot { + // A provider reset applies to the entire snapshot. Mixing a `last` value + // for one regressed counter with deltas from the old baseline for the other + // counters invents a request that never existed. + const reset = previous === null || current.some((value, index) => value < previous[index]) + if (reset) return last ?? current + return current.map((value, index) => value - previous[index]) as UsageSnapshot } function toBucket(key: string, totals: Totals): UsageSummaryBucket { @@ -293,7 +353,7 @@ export function getUsageSummary( const byAgent = new Map() const byModel = new Map() const sessionsWithUsage = new Set() - const cumulativePrevious = new Map() + const cumulativePrevious = new Map() const cumulativeFingerprints = new Set() const dayFormatter = createDayFormatter(timeZone) @@ -306,12 +366,30 @@ export function getUsageSummary( if (event.kind === 'cumulative') { const sourceParts = event.sourceKey.split('|') const streamKey = sourceParts.slice(0, 3).join('|') - const previous = cumulativePrevious.get(streamKey) - inputTokens = cumulativeDelta(inputTokens, previous?.[0] ?? null, event.lastInputTokens) - outputTokens = cumulativeDelta(outputTokens, previous?.[1] ?? null, event.lastOutputTokens) - cacheReadTokens = cumulativeDelta(cacheReadTokens, previous?.[2] ?? null, event.lastCacheReadTokens) - cacheCreationTokens = cumulativeDelta(cacheCreationTokens, previous?.[3] ?? null, event.lastCacheCreationTokens) - cumulativePrevious.set(streamKey, [event.inputTokens, event.outputTokens, event.cacheReadTokens, event.cacheCreationTokens]) + const previous = cumulativePrevious.get(streamKey) ?? null + const current: UsageSnapshot = [ + event.inputTokens, + event.outputTokens, + event.cacheReadTokens, + event.cacheCreationTokens + ] + const last: UsageSnapshot | null = event.lastInputTokens !== null + && event.lastOutputTokens !== null + && event.lastCacheReadTokens !== null + && event.lastCacheCreationTokens !== null + ? [ + event.lastInputTokens, + event.lastOutputTokens, + event.lastCacheReadTokens, + event.lastCacheCreationTokens + ] + : null + const delta = cumulativeSnapshotDelta(current, previous, last) + inputTokens = delta[0] + outputTokens = delta[1] + cacheReadTokens = delta[2] + cacheCreationTokens = delta[3] + cumulativePrevious.set(streamKey, current) const turnId = sourceParts[3] if (turnId) { const fingerprint = [ @@ -331,20 +409,25 @@ export function getUsageSummary( } } if (duplicateCumulativeEvent || !isInRange(event) || inputTokens + outputTokens + cacheReadTokens + cacheCreationTokens <= 0) continue - const normalizedInputTokens = event.agent === 'claude' - ? inputTokens + cacheReadTokens + cacheCreationTokens - : inputTokens - addTotals(totals, normalizedInputTokens, outputTokens, cacheReadTokens, cacheCreationTokens) + // Cache reads and writes partition processed input. Preserve the + // request and its primary token counts when a provider emits an + // impossible partition, but conservatively decline to credit either + // cache bucket because their split is not trustworthy. + if (cacheReadTokens + cacheCreationTokens > inputTokens) { + cacheReadTokens = 0 + cacheCreationTokens = 0 + } + addTotals(totals, inputTokens, outputTokens, cacheReadTokens, cacheCreationTokens) const eventDayKey = dayKey(event.createdAt, dayFormatter) const dailyTotals = daily.get(eventDayKey) ?? emptyTotals() - addTotals(dailyTotals, normalizedInputTokens, outputTokens, cacheReadTokens, cacheCreationTokens) + addTotals(dailyTotals, inputTokens, outputTokens, cacheReadTokens, cacheCreationTokens) daily.set(eventDayKey, dailyTotals) const agentTotals = byAgent.get(event.agent) ?? emptyTotals() - addTotals(agentTotals, normalizedInputTokens, outputTokens, cacheReadTokens, cacheCreationTokens) + addTotals(agentTotals, inputTokens, outputTokens, cacheReadTokens, cacheCreationTokens) byAgent.set(event.agent, agentTotals) const modelKey = event.model ?? 'unknown' const modelTotals = byModel.get(modelKey) ?? emptyTotals() - addTotals(modelTotals, normalizedInputTokens, outputTokens, cacheReadTokens, cacheCreationTokens) + addTotals(modelTotals, inputTokens, outputTokens, cacheReadTokens, cacheCreationTokens) byModel.set(modelKey, modelTotals) sessionsWithUsage.add(event.sessionId) } diff --git a/shared/package.json b/shared/package.json index c255e5bf..07c0c8f0 100644 --- a/shared/package.json +++ b/shared/package.json @@ -20,7 +20,8 @@ "./types": "./src/types.ts", "./voice": "./src/voice.ts", "./voicePickerCatalog": "./src/voicePickerCatalog.ts", - "./voice-personality": "./src/voicePersonality.ts" + "./voice-personality": "./src/voicePersonality.ts", + "./usage": "./src/usage.ts" }, "sideEffects": false, "scripts": { diff --git a/shared/src/index.ts b/shared/src/index.ts index f38afbed..7d182039 100644 --- a/shared/src/index.ts +++ b/shared/src/index.ts @@ -19,5 +19,6 @@ export * from './runnerCapabilities' export * from './copilotModes' export * from './slashCommands' export * from './utils' +export * from './usage' export * from './version' export type * from './types' diff --git a/shared/src/usage.ts b/shared/src/usage.ts new file mode 100644 index 00000000..06d2a055 --- /dev/null +++ b/shared/src/usage.ts @@ -0,0 +1,10 @@ +/** + * Protocol marker for token_count payloads whose input token total already + * includes cache reads/writes. Consumers must not add cache tokens again. + */ +export const INCLUSIVE_INPUT_TOKEN_USAGE_MARKER = { + usageSchema: 'hapi.usage.v1', + inputTokenSemantics: 'includes-cache' +} as const + +export type InclusiveInputTokenUsageMarker = typeof INCLUSIVE_INPUT_TOKEN_USAGE_MARKER