From 2d5776317762d382b68b3b117863dcd8a54a8a1d Mon Sep 17 00:00:00 2001 From: wusumac <736139669@qq.com> Date: Fri, 16 Oct 2026 19:38:07 +0800 Subject: [PATCH] fix(hub,cli): run OpenCode usage scans off the process event loop The 5-minute reconciliation read the multi-GB OpenCode store on the hub's event loop with bun:sqlite; a cold scan froze every request for ~88s and stalled each machine's runner the same way. - scan in a spawned `hapi usage-scan` child process (stdin/stdout JSON) - refresh on demand from the usage page (debounced) + a daily fallback - a runner whose hub is localhost skips the store the hub already owns - machine scans run every 30min and read the message table sequentially in one pass instead of per-session indexed lookups --- cli/src/commands/registry.ts | 6 +- cli/src/commands/usageScan.test.ts | 86 ++++++++++++++ cli/src/commands/usageScan.ts | 62 ++++++++++ cli/src/runner/opencodeUsageScanner.ts | 89 ++++++++++++-- cli/src/runner/run.ts | 9 +- hub/src/startHub.ts | 6 +- hub/src/sync/usageReconciliation.ts | 156 ++++++++++++++++++++++--- hub/src/web/routes/usage.ts | 14 ++- hub/src/web/server.ts | 7 +- shared/src/opencodeUsage.ts | 89 +++++++++++--- 10 files changed, 474 insertions(+), 50 deletions(-) create mode 100644 cli/src/commands/usageScan.test.ts create mode 100644 cli/src/commands/usageScan.ts diff --git a/cli/src/commands/registry.ts b/cli/src/commands/registry.ts index df1bd052..32273473 100644 --- a/cli/src/commands/registry.ts +++ b/cli/src/commands/registry.ts @@ -21,6 +21,7 @@ import { notifyCommand } from './notify' import { hubCommand } from './hub' import { pingPeerCommand } from './pingPeer' import { inspectPeerCommand } from './inspectPeer' +import { usageScanCommand } from './usageScan' import type { CommandContext, CommandDefinition } from './types' // Gemini CLI was sunset (Google stopped serving the consumer Gemini CLI on @@ -63,7 +64,10 @@ const COMMANDS: CommandDefinition[] = [ runnerCommand, notifyCommand, pingPeerCommand, - inspectPeerCommand + inspectPeerCommand, + // Internal: spawned by the hub to scan the local OpenCode store without + // blocking its event loop (see hub/src/sync/usageReconciliation.ts). + usageScanCommand ] const commandMap = new Map() diff --git a/cli/src/commands/usageScan.test.ts b/cli/src/commands/usageScan.test.ts new file mode 100644 index 00000000..9bfec956 --- /dev/null +++ b/cli/src/commands/usageScan.test.ts @@ -0,0 +1,86 @@ +import { Database } from 'bun:sqlite' +import { describe, expect, it } from 'bun:test' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { parseUsageScanRequest, runUsageScan } from './usageScan' + +function createOpencodeStore(): string { + const dir = mkdtempSync(join(tmpdir(), 'hapi-usage-scan-')) + const dbPath = join(dir, 'opencode.db') + const db = new Database(dbPath, { create: true }) + db.exec(` + CREATE TABLE message ( + id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + time_created INTEGER NOT NULL, + time_updated INTEGER NOT NULL, + data TEXT NOT NULL + ) + `) + db.prepare('INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)').run( + 'msg-1', + 'ses_scan', + Date.parse('2026-09-17T10:00:00'), + Date.parse('2026-09-17T10:00:00'), + JSON.stringify({ + role: 'assistant', + providerID: 'opencode-go', + modelID: 'deepseek-v4.1-flash', + time: { created: Date.parse('2026-09-17T10:00:00') }, + tokens: { input: 7, output: 3, reasoning: 1, cache: { read: 5, write: 2 } } + }) + ) + db.close() + return dbPath +} + +describe('usage-scan command', () => { + it('returns the hub-facing payload for the requested sessions', async () => { + const dbPath = createOpencodeStore() + try { + const payload = await runUsageScan({ dbPath, sessionIds: ['ses_scan', 'ses_missing'] }) + expect(payload.results.ses_scan.messages).toBe(1) + expect(payload.results.ses_scan.rows).toEqual([ + expect.objectContaining({ + day: '2026-09-17', + model: 'opencode-go/deepseek-v4.1-flash', + inputTokens: 14, + outputTokens: 4, + requests: 1 + }) + ]) + expect(payload.results.ses_missing).toEqual({ rows: [], messages: 0 }) + } finally { + rmSync(join(dbPath, '..'), { recursive: true, force: true }) + } + }) + + it('listAll scans every session present in the store', async () => { + const dbPath = createOpencodeStore() + try { + const payload = await runUsageScan({ dbPath, listAll: true }) + expect(Object.keys(payload.results)).toEqual(['ses_scan']) + expect(payload.results.ses_scan.messages).toBe(1) + // The sequential all-sessions pass must match the indexed + // per-session reads (this path feeds the runner's reports). + const perSession = await runUsageScan({ dbPath, sessionIds: ['ses_scan'] }) + expect(payload.results.ses_scan).toEqual(perSession.results.ses_scan) + } finally { + rmSync(join(dbPath, '..'), { recursive: true, force: true }) + } + }) + + it('rejects malformed requests', () => { + expect(() => parseUsageScanRequest('not json')).toThrow() + expect(() => parseUsageScanRequest('{"sessionIds":[]}')).toThrow() + expect(parseUsageScanRequest('{"dbPath":"/x","sessionIds":["ses_a",3,""]}')).toEqual({ + dbPath: '/x', + sessionIds: ['ses_a'] + }) + expect(parseUsageScanRequest('{"dbPath":"/x","listAll":true}')).toEqual({ + dbPath: '/x', + listAll: true + }) + }) +}) diff --git a/cli/src/commands/usageScan.ts b/cli/src/commands/usageScan.ts new file mode 100644 index 00000000..aafb8077 --- /dev/null +++ b/cli/src/commands/usageScan.ts @@ -0,0 +1,62 @@ +import { readFileSync } from 'node:fs' +import type { CommandDefinition } from './types' + +/** + * Internal command: aggregate OpenCode token usage. + * + * The hub and the runner spawn `hapi usage-scan` as a child process so the + * synchronous bun:sqlite reads over the multi-GB OpenCode store never block + * their event loops (a cold scan used to freeze the hub for a minute or + * more). The request arrives as JSON on stdin, the result leaves as JSON on + * stdout: `{ dbPath, sessionIds }` for known sessions, or + * `{ dbPath, listAll: true }` to scan every session found in the store. + */ + +export type UsageScanRequest = { + dbPath: string + sessionIds?: string[] + listAll?: boolean +} + +function readStdin(): string { + // fd 0 works for both pipe and file-backed stdin; Bun's async iteration + // over a file-backed stdin yields nothing. + return readFileSync(0, 'utf8') +} + +export function parseUsageScanRequest(raw: string): UsageScanRequest { + const parsed = JSON.parse(raw) as { dbPath?: unknown; sessionIds?: unknown; listAll?: unknown } + if (typeof parsed.dbPath !== 'string') { + throw new Error('usage-scan expects { dbPath: string, sessionIds?: string[], listAll?: boolean } on stdin') + } + if (parsed.listAll === true) { + return { dbPath: parsed.dbPath, listAll: true } + } + if (!Array.isArray(parsed.sessionIds)) { + throw new Error('usage-scan expects sessionIds or listAll: true on stdin') + } + const sessionIds = parsed.sessionIds.filter((id): id is string => typeof id === 'string' && id.length > 0) + return { dbPath: parsed.dbPath, sessionIds } +} + +/** Payload shape the hub/runner parse (see hub/src/sync/usageReconciliation.ts). */ +export async function runUsageScan(request: UsageScanRequest): Promise<{ results: Record }> { + const { readOpencodeUsageForAllSessions, readOpencodeUsageForSessions } = await import('@hapi/protocol/opencodeUsage') + const usage = request.listAll === true + ? await readOpencodeUsageForAllSessions(request.dbPath) + : await readOpencodeUsageForSessions(request.dbPath, request.sessionIds ?? []) + const results: Record = {} + for (const [sessionId, entry] of usage) { + results[sessionId] = { rows: entry.rows, messages: entry.messages } + } + return { results } +} + +export const usageScanCommand: CommandDefinition = { + name: 'usage-scan', + requiresRuntimeAssets: false, + run: async () => { + const request = parseUsageScanRequest(readStdin()) + process.stdout.write(JSON.stringify(await runUsageScan(request))) + } +} diff --git a/cli/src/runner/opencodeUsageScanner.ts b/cli/src/runner/opencodeUsageScanner.ts index 52f981a8..035d2413 100644 --- a/cli/src/runner/opencodeUsageScanner.ts +++ b/cli/src/runner/opencodeUsageScanner.ts @@ -1,5 +1,5 @@ import { listOpencodeSessionIds, readOpencodeUsageForSessions, resolveOpencodeDbPath } from '@hapi/protocol/opencodeUsage' -import type { OpencodeUsageSessionReport } from '@hapi/protocol/usage' +import type { OpencodeUsageReportRow, OpencodeUsageSessionReport } from '@hapi/protocol/usage' import { logger } from '@/ui/logger' /** @@ -12,12 +12,18 @@ import { logger } from '@/ui/logger' * their reconciliation rows, which makes repeated or concurrent reports * idempotent (see hub/src/sync/usageReconciliation.ts). * + * The scan runs in a spawned `hapi usage-scan` child process: the OpenCode + * store is multi-GB in practice and `bun:sqlite` is synchronous, so an + * in-process scan would stall this runner's event loop — machine RPCs + * (spawn, model listing) queue behind it and can time out. + * * The live chat pipeline is not involved at all — a report never becomes a * timeline message. */ -const SCAN_INTERVAL_MS = 5 * 60 * 1000 +const SCAN_INTERVAL_MS = 30 * 60 * 1000 const INITIAL_DELAY_MS = 20_000 +const SCAN_TIMEOUT_MS = 2 * 60 * 1000 export type OpencodeUsageScannerOptions = { /** Best-effort delivery; the next scan re-sends on failure. */ @@ -35,15 +41,82 @@ export type OpencodeUsageScanner = { runNow: () => void } +type ScannedUsage = Map + +async function spawnUsageScan(request: { dbPath: string; listAll: true }): Promise { + if (typeof Bun === 'undefined') return null + const executable = process.env.HAPI_CLI_EXECUTABLE?.trim() || process.execPath + if (!executable) return null + try { + const proc = Bun.spawn({ + cmd: [executable, 'usage-scan'], + stdin: 'pipe', + stdout: 'pipe', + stderr: 'ignore' + }) + const timer = setTimeout(() => { + try { + proc.kill() + } catch { + } + }, SCAN_TIMEOUT_MS) + try { + proc.stdin.write(JSON.stringify(request)) + proc.stdin.end() + const [exitCode, stdout] = await Promise.all([proc.exited, new Response(proc.stdout).text()]) + if (exitCode !== 0 || stdout.trim().length === 0) return null + const parsed = JSON.parse(stdout) as { results?: unknown } + const results = parsed.results + if (results === null || typeof results !== 'object' || Array.isArray(results)) return null + const usage: ScannedUsage = new Map() + for (const [sessionId, value] of Object.entries(results as Record)) { + const entry = value as { rows?: unknown; messages?: unknown } | null + if (!entry || typeof entry !== 'object' || !Array.isArray(entry.rows)) return null + usage.set(sessionId, { + rows: entry.rows as OpencodeUsageReportRow[], + messages: typeof entry.messages === 'number' ? entry.messages : entry.rows.length + }) + } + return usage + } finally { + clearTimeout(timer) + } + } catch { + return null + } +} + +/** Probe once whether the current executable can actually run `usage-scan`. */ +let cliScanAvailablePromise: Promise | null = null + +function cliScanAvailable(): Promise { + cliScanAvailablePromise ??= (async () => { + const probe = await spawnUsageScan({ dbPath: '', listAll: true }) + return probe !== null + })() + return cliScanAvailablePromise +} + +async function scanOpencodeStore(dbPath: string): Promise { + if (await cliScanAvailable()) { + const scanned = await spawnUsageScan({ dbPath, listAll: true }) + if (scanned) return scanned + throw new Error('opencode usage scan failed') + } + // Dev/test fallback (no hapi binary to spawn): read in-process. This is + // deliberately not used in production — the multi-GB scan must never run + // on this process's event loop. + const opencodeSessionIds = await listOpencodeSessionIds(dbPath) + if (opencodeSessionIds.length === 0) return new Map() + return await readOpencodeUsageForSessions(dbPath, opencodeSessionIds) +} + /** Collect snapshot rows for every session present in the OpenCode store. */ export async function collectOpencodeUsageReports(dbPath: string): Promise { - const opencodeSessionIds = await listOpencodeSessionIds(dbPath) - if (opencodeSessionIds.length === 0) return [] - const usageBySession = await readOpencodeUsageForSessions(dbPath, opencodeSessionIds) + const usageBySession = await scanOpencodeStore(dbPath) const reports: OpencodeUsageSessionReport[] = [] - for (const opencodeSessionId of opencodeSessionIds) { - const usage = usageBySession.get(opencodeSessionId) - if (!usage || usage.rows.length === 0) continue + for (const [opencodeSessionId, usage] of usageBySession) { + if (usage.rows.length === 0) continue reports.push({ opencodeSessionId, rows: usage.rows }) } return reports diff --git a/cli/src/runner/run.ts b/cli/src/runner/run.ts index e6294409..59e1e962 100644 --- a/cli/src/runner/run.ts +++ b/cli/src/runner/run.ts @@ -1166,8 +1166,11 @@ export async function startRunner(options: { workspaceRoots?: string[] } = {}): apiMachine.connect(); // Report this machine's OpenCode usage snapshots so the hub's token // dashboard does not under-count tool-loop turns on non-hub machines. - // The scan reads the local OpenCode store; best-effort, never blocks. - const opencodeUsageScanner = startOpencodeUsageScanner({ + // A runner whose hub is localhost would only rescan the store the hub + // already reconciles itself (see hub/src/sync/usageReconciliation.ts), + // so skip it there. The scan itself runs in a child process. + const hubIsLocal = /^https?:\/\/(localhost|127\.0\.0\.1|\[::1\])(?::|\/|$)/i.test(configuration.apiUrl); + const opencodeUsageScanner = hubIsLocal ? null : startOpencodeUsageScanner({ report: (sessions) => apiMachine.reportOpencodeUsage(sessions) }); scheduleCursorModelsPrewarm(); @@ -1435,7 +1438,7 @@ export async function startRunner(options: { workspaceRoots?: string[] } = {}): clearInterval(restartOnStaleVersionAndHeartbeat); logger.debug('[RUNNER RUN] Health check interval cleared'); } - opencodeUsageScanner.stop(); + opencodeUsageScanner?.stop(); // Update runner state before shutting down await apiMachine.updateRunnerState((state: RunnerState | null) => ({ diff --git a/hub/src/startHub.ts b/hub/src/startHub.ts index cb13a659..d21ada53 100644 --- a/hub/src/startHub.ts +++ b/hub/src/startHub.ts @@ -413,7 +413,11 @@ export async function startHub(options: StartHubOptions = {}): Promise { + await usageReconciliation?.refresh() + } }) // Start the bot if configured diff --git a/hub/src/sync/usageReconciliation.ts b/hub/src/sync/usageReconciliation.ts index 88fd25a9..efdd7afd 100644 --- a/hub/src/sync/usageReconciliation.ts +++ b/hub/src/sync/usageReconciliation.ts @@ -1,5 +1,9 @@ import { existsSync } from 'node:fs' -import { readOpencodeUsageForSessions, resolveOpencodeDbPath } from '@hapi/protocol/opencodeUsage' +import { + readOpencodeUsageForSessions, + resolveOpencodeDbPath, + type OpencodeSessionUsage +} from '@hapi/protocol/opencodeUsage' import type { Store } from '../store' import type { ReconciledUsageRow } from '../store/usage' import type { UsageSource } from '../store/usageSources' @@ -21,10 +25,18 @@ import type { UsageSource } from '../store/usageSources' * hub machine. Every other machine's runner reports the same row shape over * the `opencode-usage-report` socket event (see `@hapi/protocol/opencodeUsage` * and cli/src/runner/opencodeUsageScanner.ts). + * + * The scan itself runs in a spawned `hapi usage-scan` child process: the + * OpenCode store is multi-GB in practice, and `bun:sqlite` is synchronous, so + * an in-process scan would block the hub's event loop for the whole read + * (observed: 88s freezes under memory pressure). The hub only ever applies the + * small aggregated result. */ -const RECONCILE_INTERVAL_MS = 5 * 60 * 1000 -const INITIAL_DELAY_MS = 15_000 +const FALLBACK_INTERVAL_MS = 24 * 60 * 60 * 1000 +const INITIAL_DELAY_MS = 60_000 +const REFRESH_MAX_AGE_MS = 5 * 60 * 1000 +const SCAN_TIMEOUT_MS = 2 * 60 * 1000 export type UsageReconciliationResult = { dbPath: string | null @@ -33,6 +45,15 @@ export type UsageReconciliationResult = { messages: number } +export type OpencodeUsageScanRequest = { + dbPath: string + sessionIds: string[] +} + +export type OpencodeUsageScan = ( + request: OpencodeUsageScanRequest +) => Promise> + function asRecord(value: unknown): Record | null { return value !== null && typeof value === 'object' && !Array.isArray(value) ? value as Record @@ -43,10 +64,87 @@ function asString(value: unknown): string | null { return typeof value === 'string' && value.trim().length > 0 ? value.trim() : null } +function parseScanOutput(stdout: string): Map | null { + const parsed = JSON.parse(stdout) as { results?: unknown } + const results = asRecord(parsed.results) + if (!results) return null + const usage = new Map() + for (const [sessionId, value] of Object.entries(results)) { + const entry = asRecord(value) + if (!entry || !Array.isArray(entry.rows) || typeof entry.messages !== 'number') return null + usage.set(sessionId, { rows: entry.rows as OpencodeSessionUsage['rows'], messages: entry.messages }) + } + return usage +} + +async function spawnUsageScanCli(request: OpencodeUsageScanRequest): Promise { + const executable = process.env.HAPI_CLI_EXECUTABLE?.trim() || process.execPath + if (!executable) return null + const proc = Bun.spawn({ + cmd: [executable, 'usage-scan'], + stdin: 'pipe', + stdout: 'pipe', + stderr: 'ignore' + }) + const timer = setTimeout(() => { + try { + proc.kill() + } catch { + } + }, SCAN_TIMEOUT_MS) + try { + proc.stdin.write(JSON.stringify(request)) + proc.stdin.end() + const [exitCode, stdout] = await Promise.all([proc.exited, new Response(proc.stdout).text()]) + if (exitCode !== 0 || stdout.trim().length === 0) return null + return stdout + } catch { + return null + } finally { + clearTimeout(timer) + } +} + +/** Probe once whether the current executable can actually run `usage-scan`. */ +let scanModePromise: Promise<'cli' | 'in-process'> | null = null + +function resolveScanMode(): Promise<'cli' | 'in-process'> { + scanModePromise ??= (async () => { + try { + // Empty request: the CLI answers immediately, while a non-hapi + // executable (e.g. `bun` under tests/dev) fails fast. + const stdout = await spawnUsageScanCli({ dbPath: '', sessionIds: [] }) + return stdout !== null && parseScanOutput(stdout) !== null ? 'cli' : 'in-process' + } catch { + return 'in-process' + } + })() + return scanModePromise +} + +/** + * Default scan: a `hapi usage-scan` child process when available. A transient + * child failure is reported as an error instead of falling back to an + * in-process read — the whole point is to keep the read off this process. + * Only when the executable cannot run the command at all (dev/test) do we + * read in-process; those environments are not the multi-GB-store case. + */ +async function defaultOpencodeUsageScan( + request: OpencodeUsageScanRequest +): Promise> { + if (await resolveScanMode() === 'cli') { + const stdout = await spawnUsageScanCli(request) + const usage = stdout === null ? null : parseScanOutput(stdout) + if (usage) return usage + throw new Error('opencode usage scan failed') + } + return await readOpencodeUsageForSessions(request.dbPath, request.sessionIds) +} + /** Re-read every known OpenCode session from the local store and refresh snapshots. */ export async function reconcileOpencodeUsage( store: Store, - options?: { dbPath?: string | null; now?: number } + options?: { dbPath?: string | null; now?: number; scan?: OpencodeUsageScan } ): Promise { const dbPath = options?.dbPath === undefined ? resolveOpencodeDbPath() : options.dbPath const now = options?.now ?? Date.now() @@ -68,7 +166,11 @@ export async function reconcileOpencodeUsage( if (!targets.has(key)) targets.set(key, source) } - const usageBySession = await readOpencodeUsageForSessions(dbPath, [...targets.values()].map((target) => target.nativeSessionId)) + const scan = options?.scan ?? defaultOpencodeUsageScan + const usageBySession = await scan({ + dbPath, + sessionIds: [...targets.values()].map((target) => target.nativeSessionId) + }) let sessionsTouched = 0 let rowsTotal = 0 let messagesTotal = 0 @@ -95,19 +197,30 @@ export async function reconcileOpencodeUsage( export type UsageReconciliationJob = { /** Idempotent. */ stop: () => void - /** Run one reconciliation pass immediately (used by tests). */ - runNow: () => void + /** Run one reconciliation pass now (concurrent callers share the pass). */ + runNow: () => Promise + /** Run a pass unless the last one finished within `maxAgeMs`. */ + refresh: (maxAgeMs?: number) => Promise } -/** Schedule periodic reconciliation while the hub is running. */ +/** + * Schedule reconciliation while the hub is running. + * + * The dashboard refreshes on demand (`refresh`, triggered by the usage page), + * so the timer is only a gentle safety net — one pass shortly after startup + * and a daily fallback. The pass itself never blocks the hub (child process). + */ export function startOpencodeUsageReconciliation( store: Store, options?: { intervalMs?: number; initialDelayMs?: number } ): UsageReconciliationJob { let stopped = false - const run = (): void => { - if (stopped) return - void reconcileOpencodeUsage(store) + let lastCompletedAt = 0 + let inFlight: Promise | null = null + + const runPass = (): Promise => { + if (stopped) return Promise.resolve() + const pass = reconcileOpencodeUsage(store) .then((result) => { if (result.messages > 0) { console.log( @@ -118,10 +231,24 @@ export function startOpencodeUsageReconciliation( .catch((error) => { console.warn('[UsageReconciliation] failed:', error instanceof Error ? error.message : error) }) + .finally(() => { + lastCompletedAt = Date.now() + inFlight = null + }) + inFlight = pass + return pass } - const initial = setTimeout(run, options?.initialDelayMs ?? INITIAL_DELAY_MS) - const interval = setInterval(run, options?.intervalMs ?? RECONCILE_INTERVAL_MS) + const runNow = (): Promise => inFlight ?? runPass() + + const refresh = async (maxAgeMs?: number): Promise => { + const maxAge = typeof maxAgeMs === 'number' && Number.isFinite(maxAgeMs) ? maxAgeMs : REFRESH_MAX_AGE_MS + if (Date.now() - lastCompletedAt < maxAge) return + await runNow() + } + + const initial = setTimeout(() => void runNow(), options?.initialDelayMs ?? INITIAL_DELAY_MS) + const interval = setInterval(() => void runNow(), options?.intervalMs ?? FALLBACK_INTERVAL_MS) initial.unref?.() interval.unref?.() @@ -131,6 +258,7 @@ export function startOpencodeUsageReconciliation( clearTimeout(initial) clearInterval(interval) }, - runNow: run + runNow, + refresh } } diff --git a/hub/src/web/routes/usage.ts b/hub/src/web/routes/usage.ts index 5be1d4a8..9ac9619d 100644 --- a/hub/src/web/routes/usage.ts +++ b/hub/src/web/routes/usage.ts @@ -4,10 +4,10 @@ import type { WebAppEnv } from '../middleware/auth' import type { Store } from '../../store' import { getUsageSummary } from '../../sync/usageService' -export function createUsageRoutes(store: Store): Hono { +export function createUsageRoutes(store: Store, refreshUsage?: () => Promise): Hono { const app = new Hono() - app.get('/usage/summary', (c) => { + app.get('/usage/summary', async (c) => { if (c.get('namespace') !== 'default') { return c.json({ error: 'Usage summary is only available to the hub owner' }, 403) } @@ -21,6 +21,16 @@ export function createUsageRoutes(store: Store): Hono { } catch { return c.json({ error: 'Invalid timeZone' }, 400) } + // Refresh OpenCode snapshots on demand (debounced by the job): the + // scan runs in a child process, so waiting here never blocks the hub, + // and the first page open after a quiet period shows fresh numbers. + if (refreshUsage) { + try { + await refreshUsage() + } catch { + // A failed refresh leaves the previous snapshot in place. + } + } const response: UsageSummaryResponse = getUsageSummary(store, c.get('namespace'), range, timeZone) c.header('Cache-Control', 'no-store') return c.json(response) diff --git a/hub/src/web/server.ts b/hub/src/web/server.ts index ee86eadd..74b4e30e 100644 --- a/hub/src/web/server.ts +++ b/hub/src/web/server.ts @@ -239,6 +239,7 @@ function createWebApp(options: { relayMode?: boolean officialWebUrl?: string dataDir: string + refreshUsage?: () => Promise }): Hono { const app = new Hono() @@ -324,7 +325,7 @@ function createWebApp(options: { if (teamService) { app.route('/api', createTeamsRoutes(teamService)) } - app.route('/api', createUsageRoutes(options.store)) + app.route('/api', createUsageRoutes(options.store, options.refreshUsage)) app.route('/api', createStudioRoutes({ store: options.store, getSyncEngine: options.getSyncEngine @@ -483,6 +484,7 @@ export async function startWebServer(options: { relayMode?: boolean officialWebUrl?: string dataDir: string + refreshUsage?: () => Promise }): Promise> { const isCompiled = isBunCompiled() const embeddedAssetMap = isCompiled ? await loadEmbeddedAssetMap() : null @@ -498,7 +500,8 @@ export async function startWebServer(options: { embeddedAssetMap, dataDir: options.dataDir, relayMode: options.relayMode, - officialWebUrl: options.officialWebUrl + officialWebUrl: options.officialWebUrl, + refreshUsage: options.refreshUsage }) const configuration = getConfiguration() diff --git a/shared/src/opencodeUsage.ts b/shared/src/opencodeUsage.ts index d00badb7..db281082 100644 --- a/shared/src/opencodeUsage.ts +++ b/shared/src/opencodeUsage.ts @@ -136,6 +136,32 @@ export async function listOpencodeSessionIds(dbPath: string): Promise } } +/** Fold one message row's token counters into its (day, model) bucket. */ +function accumulateOpencodeUsage( + buckets: Map, + data: string +): boolean { + const usage = parseOpencodeAssistantUsage(data) + if (!usage) return false + const key = `${usage.day}|${usage.model}` + const bucket = buckets.get(key) ?? { + day: usage.day, + model: usage.model, + inputTokens: 0, + outputTokens: 0, + cacheReadTokens: 0, + cacheCreationTokens: 0, + requests: 0 + } + bucket.inputTokens += usage.inputTokens + bucket.outputTokens += usage.outputTokens + bucket.cacheReadTokens += usage.cacheReadTokens + bucket.cacheCreationTokens += usage.cacheCreationTokens + bucket.requests += 1 + buckets.set(key, bucket) + return true +} + /** * Aggregate absolute per-(day, model) totals for the given OpenCode session * ids. Missing database or unknown session ids yield empty entries, so @@ -163,25 +189,7 @@ export async function readOpencodeUsageForSessions( const buckets = new Map() let messages = 0 for (const row of messageRows) { - const usage = parseOpencodeAssistantUsage(row.data) - if (!usage) continue - messages += 1 - const key = `${usage.day}|${usage.model}` - const bucket = buckets.get(key) ?? { - day: usage.day, - model: usage.model, - inputTokens: 0, - outputTokens: 0, - cacheReadTokens: 0, - cacheCreationTokens: 0, - requests: 0 - } - bucket.inputTokens += usage.inputTokens - bucket.outputTokens += usage.outputTokens - bucket.cacheReadTokens += usage.cacheReadTokens - bucket.cacheCreationTokens += usage.cacheCreationTokens - bucket.requests += 1 - buckets.set(key, bucket) + if (accumulateOpencodeUsage(buckets, row.data)) messages += 1 } result.set(sessionId, { rows: Array.from(buckets.values()), messages }) } @@ -191,3 +199,46 @@ export async function readOpencodeUsageForSessions( return result } + +/** + * Like `readOpencodeUsageForSessions` but for every session present in the + * store (the machine-level runner path). Reads the message table in a single + * sequential pass: the per-session index queries above translate into one + * scattered page fetch per message across the multi-GB store, while a table + * scan reads the same pages in storage order. + */ +export async function readOpencodeUsageForAllSessions( + dbPath: string +): Promise> { + const result = new Map() + if (!existsSync(dbPath)) return result + + const { Database } = await import('bun:sqlite') + const db = new Database(dbPath, { readonly: true }) + try { + const rows = db.prepare('SELECT session_id, data FROM message').all() as Array<{ + session_id: string + data: string + }> + const bucketsBySession = new Map>() + const messagesBySession = new Map() + for (const row of rows) { + if (typeof row.session_id !== 'string' || row.session_id.length === 0) continue + const buckets = bucketsBySession.get(row.session_id) ?? new Map() + if (accumulateOpencodeUsage(buckets, row.data)) { + messagesBySession.set(row.session_id, (messagesBySession.get(row.session_id) ?? 0) + 1) + } + bucketsBySession.set(row.session_id, buckets) + } + for (const [sessionId, buckets] of bucketsBySession) { + result.set(sessionId, { + rows: Array.from(buckets.values()), + messages: messagesBySession.get(sessionId) ?? 0 + }) + } + } finally { + db.close() + } + + return result +}