| author | |
| committer | |
| log | 23957c704b43c25a253baf2644829e88cf45917f |
| tree | 70a7e79303b6fff67abd5694756f811119e4b68c |
| parent | e5dca2332fed450c6f1b3c7472666fbe80531c5a |
| signature | Signed by PGP key B5690EEEBB952194 |
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>8 files changed, 378 insertions(+), 271 deletions(-)
apps/server/src/provider/builtInDrivers.ts+3-2| ... | ... | @@ -74,7 +74,8 @@ export const BUILT_IN_DRIVERS: ReadonlyArray<AnyProviderDriver<BuiltInDriversEnv |
| 74 | 74 | export type BuiltInUsageReadersEnv = |
| 75 | 75 | | ProviderUsageReaderEnv<typeof ClaudeDriver> |
| 76 | 76 | | ProviderUsageReaderEnv<typeof CodexDriver> |
| 77 | | ProviderUsageReaderEnv<typeof GrokDriver>; | |
| 77 | | ProviderUsageReaderEnv<typeof GrokDriver> | |
| 78 | | ProviderUsageReaderEnv<typeof OpenCodeDriver>; | |
| 78 | 79 | |
| 79 | 80 | /** |
| 80 | 81 | * The drivers that keep usage history, in the order the usage page reads |
| ... | ... | @@ -83,4 +84,4 @@ export type BuiltInUsageReadersEnv = |
| 83 | 84 | */ |
| 84 | 85 | export const BUILT_IN_USAGE_DRIVERS: ReadonlyArray< |
| 85 | 86 | AnyProviderDriver<BuiltInDriversEnv, BuiltInUsageReadersEnv> |
| 86 | > = [ClaudeDriver, CodexDriver, GrokDriver]; | |
| 87 | > = [ClaudeDriver, CodexDriver, GrokDriver, OpenCodeDriver]; |
apps/server/src/usage/UsageService.ts+37-29| ... | ... | @@ -63,7 +63,6 @@ import type { |
| 63 | 63 | TranscriptUsageFormat, |
| 64 | 64 | UsageRecord, |
| 65 | 65 | } from "@t3tools/provider-core/server/usage"; |
| 66 | import { readOpenCodeUsage } from "./opencodeUsageReader.ts"; | |
| 67 | 66 | import { makeAntigravityUsageCache, readAntigravityUsage } from "./antigravityUsageReader.ts"; |
| 68 | 67 | import { |
| 69 | 68 | CURSOR_ACCOUNT_CACHE_FILE_NAME, |
| ... | ... | @@ -127,6 +126,11 @@ const transcriptReaders = BUILT_IN_USAGE_DRIVERS.flatMap((driver) => |
| 127 | 126 | driver.usage?.kind === "transcripts" ? [{ driver, reader: driver.usage }] : [], |
| 128 | 127 | ); |
| 129 | 128 | |
| 129 | /** The scan readers, in driver order. */ | |
| 130 | const scanReaders = BUILT_IN_USAGE_DRIVERS.flatMap((driver) => | |
| 131 | driver.usage?.kind === "scan" ? [{ driver, reader: driver.usage }] : [], | |
| 132 | ); | |
| 133 | ||
| 130 | 134 | /** Transcript formats by provider, for decoding the persisted scan cache. */ |
| 131 | 135 | const transcriptFormats = new Map( |
| 132 | 136 | transcriptReaders.map(({ reader }) => [reader.provider, reader.format] as const), |
| ... | ... | @@ -851,32 +855,36 @@ export const make = Effect.gen(function* () { |
| 851 | 855 | } |
| 852 | 856 | return [...canonical]; |
| 853 | 857 | }); |
| 854 | const dataHome = hostEnvironment["XDG_DATA_HOME"]?.trim(); | |
| 855 | 858 | |
| 856 | const openCode = Effect.gen(function* () { | |
| 857 | const roots = yield* envRoots("OPENCODE_DATA_DIR", [ | |
| 858 | path.join( | |
| 859 | dataHome && path.isAbsolute(dataHome) ? dataHome : path.join(home, ".local", "share"), | |
| 860 | "opencode", | |
| 861 | ), | |
| 862 | ]); | |
| 863 | return yield* Effect.forEach( | |
| 864 | roots, | |
| 865 | (dir) => | |
| 866 | Effect.gen(function* () { | |
| 867 | const result = yield* Effect.promise(() => readOpenCodeUsage(dir, windowStartMs)); | |
| 868 | return { | |
| 869 | provider: "opencode", | |
| 870 | dir, | |
| 871 | volumeId: yield* Effect.promise(() => readDirectoryVolumeId(dir)), | |
| 872 | files: result.missing && !result.error ? null : result.files, | |
| 873 | status: result.error ? "partial" : "ok", | |
| 874 | ...(result.error ? { message: "Some OpenCode history could not be read." } : {}), | |
| 875 | } satisfies ScannedDir; | |
| 876 | }), | |
| 877 | { concurrency: "unbounded" }, | |
| 878 | ); | |
| 879 | }); | |
| 859 | const scans = Effect.forEach( | |
| 860 | scanReaders, | |
| 861 | ({ driver, reader }) => | |
| 862 | reader | |
| 863 | .scan({ | |
| 864 | instances: usageInstances(driver, settings), | |
| 865 | windowStartMs, | |
| 866 | retentionCutoffMs, | |
| 867 | awaitRefresh, | |
| 868 | }) | |
| 869 | .pipe( | |
| 870 | Effect.flatMap((sources) => | |
| 871 | Effect.forEach(sources, ({ volumeId, ...source }) => | |
| 872 | Effect.map( | |
| 873 | volumeId === undefined | |
| 874 | ? Effect.promise(() => readDirectoryVolumeId(source.dir)) | |
| 875 | : Effect.succeed(volumeId), | |
| 876 | (resolved): ScannedDir => ({ | |
| 877 | ...source, | |
| 878 | provider: reader.provider, | |
| 879 | volumeId: resolved, | |
| 880 | }), | |
| 881 | ), | |
| 882 | ), | |
| 883 | ), | |
| 884 | Effect.provideContext(readerContext), | |
| 885 | ), | |
| 886 | { concurrency: "unbounded" }, | |
| 887 | ).pipe(Effect.map((sources) => sources.flat())); | |
| 880 | 888 | |
| 881 | 889 | const antigravity = Effect.gen(function* () { |
| 882 | 890 | const antigravityRoots = yield* envRoots("ANTIGRAVITY_DATA_DIR", [ |
| ... | ... | @@ -999,10 +1007,10 @@ export const make = Effect.gen(function* () { |
| 999 | 1007 | // Independent sources scan together. Transcript directories go one at a |
| 1000 | 1008 | // time, so open files stay at `TRANSCRIPT_READ_CONCURRENCY`. The result |
| 1001 | 1009 | // keeps this order, since aggregation keeps the first copy of a duplicate. |
| 1002 | const [transcripts, openCodeDirs, antigravityDirs, cursorDirs] = yield* Effect.all( | |
| 1010 | const [transcripts, scanDirs, antigravityDirs, cursorDirs] = yield* Effect.all( | |
| 1003 | 1011 | [ |
| 1004 | 1012 | Effect.forEach(dirs, (dir) => scanTranscriptDir(dir, windowStartMs)), |
| 1005 | openCode, | |
| 1013 | scans, | |
| 1006 | 1014 | antigravity, |
| 1007 | 1015 | cursor, |
| 1008 | 1016 | ], |
| ... | ... | @@ -1010,7 +1018,7 @@ export const make = Effect.gen(function* () { |
| 1010 | 1018 | ); |
| 1011 | 1019 | const scanned: readonly ScannedDir[] = [ |
| 1012 | 1020 | ...transcripts, |
| 1013 | ...openCodeDirs, | |
| 1021 | ...scanDirs, | |
| 1014 | 1022 | ...antigravityDirs, |
| 1015 | 1023 | ...cursorDirs, |
| 1016 | 1024 | ]; |
apps/server/src/usage/opencodeUsageReader.ts deleted-188| ... | ... | @@ -1,188 +0,0 @@ |
| 1 | // node:sqlite reads live OpenCode databases; Node fs walks legacy JSON history. | |
| 2 | // @effect-diagnostics nodeBuiltinImport:off | |
| 3 | import * as NodeFSP from "node:fs/promises"; | |
| 4 | import * as NodePath from "node:path"; | |
| 5 | import * as NodeSqlite from "node:sqlite"; | |
| 6 | import * as NodeTimersPromises from "node:timers/promises"; | |
| 7 | ||
| 8 | import { totalTokens, type UsageRecord } from "@t3tools/provider-core/server/usage"; | |
| 9 | ||
| 10 | function object(value: unknown): Record<string, unknown> { | |
| 11 | return typeof value === "object" && value !== null && !Array.isArray(value) | |
| 12 | ? (value as Record<string, unknown>) | |
| 13 | : {}; | |
| 14 | } | |
| 15 | ||
| 16 | function tokens(value: unknown): number { | |
| 17 | return typeof value === "number" && Number.isFinite(value) && value > 0 ? Math.trunc(value) : 0; | |
| 18 | } | |
| 19 | ||
| 20 | function text(value: unknown): string { | |
| 21 | return typeof value === "string" ? value : ""; | |
| 22 | } | |
| 23 | ||
| 24 | /** OpenCode stores uncached input and reasoning separately from input/output. */ | |
| 25 | function parseOpenCodeMessage( | |
| 26 | source: string, | |
| 27 | fallback: { | |
| 28 | readonly id?: string; | |
| 29 | readonly sessionId?: string; | |
| 30 | readonly timestampMs?: number; | |
| 31 | } = {}, | |
| 32 | ): UsageRecord | null { | |
| 33 | let parsed: unknown; | |
| 34 | try { | |
| 35 | parsed = JSON.parse(source); | |
| 36 | } catch { | |
| 37 | return null; | |
| 38 | } | |
| 39 | const message = object(parsed); | |
| 40 | if (message.role !== undefined && message.role !== "assistant") return null; | |
| 41 | const usage = object(message.tokens); | |
| 42 | const cache = object(usage.cache); | |
| 43 | const modelReference = object(message.model); | |
| 44 | const model = text(modelReference.id) || text(modelReference.modelID) || text(message.modelID); | |
| 45 | const timestampMs = object(message.time).created ?? fallback.timestampMs; | |
| 46 | if (!model || typeof timestampMs !== "number" || !Number.isFinite(timestampMs)) return null; | |
| 47 | const reasoningTokens = tokens(usage.reasoning); | |
| 48 | const totals = { | |
| 49 | uncachedInputTokens: tokens(usage.input), | |
| 50 | cachedInputTokens: tokens(cache.read), | |
| 51 | cacheCreationTokens: tokens(cache.write), | |
| 52 | outputTokens: tokens(usage.output) + reasoningTokens, | |
| 53 | reasoningTokens, | |
| 54 | }; | |
| 55 | if (totalTokens(totals) === 0) return null; | |
| 56 | const id = fallback.id || text(message.id); | |
| 57 | const cost = message.cost; | |
| 58 | return { | |
| 59 | provider: "opencode", | |
| 60 | timestampMs, | |
| 61 | model, | |
| 62 | sessionId: fallback.sessionId || text(message.sessionID), | |
| 63 | totals, | |
| 64 | // OpenCode writes zero for models without a known rate, including paid | |
| 65 | // subscription models. Let the shared price table estimate those records. | |
| 66 | reportedCostUsd: typeof cost === "number" && Number.isFinite(cost) && cost > 0 ? cost : null, | |
| 67 | speed: "standard", | |
| 68 | dedupeKey: id ? `opencode:${id}` : null, | |
| 69 | }; | |
| 70 | } | |
| 71 | ||
| 72 | export interface OpenCodeUsageReadResult { | |
| 73 | readonly files: readonly { readonly path: string; readonly records: readonly UsageRecord[] }[]; | |
| 74 | readonly missing: boolean; | |
| 75 | readonly error: boolean; | |
| 76 | } | |
| 77 | ||
| 78 | /** Reads current SQLite and pre-migration JSON stores without modifying either. */ | |
| 79 | export async function readOpenCodeUsage( | |
| 80 | root: string, | |
| 81 | sinceMs: number, | |
| 82 | ): Promise<OpenCodeUsageReadResult> { | |
| 83 | const files: { path: string; records: UsageRecord[] }[] = []; | |
| 84 | const seen = new Set<string>(); | |
| 85 | let found = false; | |
| 86 | let error = false; | |
| 87 | const append = (records: UsageRecord[], record: UsageRecord | null) => { | |
| 88 | if (record === null || record.timestampMs < sinceMs) return; | |
| 89 | if (record.dedupeKey !== null) { | |
| 90 | if (seen.has(record.dedupeKey)) return; | |
| 91 | seen.add(record.dedupeKey); | |
| 92 | } | |
| 93 | records.push(record); | |
| 94 | }; | |
| 95 | ||
| 96 | let databases: string[] = []; | |
| 97 | try { | |
| 98 | databases = (await NodeFSP.readdir(root, { withFileTypes: true })) | |
| 99 | .filter((entry) => entry.isFile() && /^opencode(?:-[a-zA-Z0-9_-]+)?\.db$/.test(entry.name)) | |
| 100 | .map((entry) => entry.name) | |
| 101 | .sort((a, b) => (a === "opencode.db" ? -1 : b === "opencode.db" ? 1 : a.localeCompare(b))); | |
| 102 | } catch (cause) { | |
| 103 | if (object(cause).code !== "ENOENT") error = true; | |
| 104 | } | |
| 105 | for (const name of databases) { | |
| 106 | found = true; | |
| 107 | const file = { path: NodePath.join(root, name), records: [] as UsageRecord[] }; | |
| 108 | files.push(file); | |
| 109 | let database: NodeSqlite.DatabaseSync | undefined; | |
| 110 | try { | |
| 111 | database = new NodeSqlite.DatabaseSync(NodePath.join(root, name), { readOnly: true }); | |
| 112 | // A busy live provider should fail this source promptly rather than | |
| 113 | // stalling the server while SQLite waits for its writer. | |
| 114 | database.exec("PRAGMA busy_timeout = 100"); | |
| 115 | const tables = new Set( | |
| 116 | database | |
| 117 | .prepare("SELECT name FROM sqlite_master WHERE type = 'table'") | |
| 118 | .all() | |
| 119 | .map((row) => row.name), | |
| 120 | ); | |
| 121 | if (!tables.has("message") && !tables.has("session_message")) error = true; | |
| 122 | for (const table of ["message", "session_message"] as const) { | |
| 123 | if (!tables.has(table)) continue; | |
| 124 | const columns = new Set( | |
| 125 | database | |
| 126 | .prepare(`PRAGMA table_info(${table})`) | |
| 127 | .all() | |
| 128 | .map((row) => row.name), | |
| 129 | ); | |
| 130 | const timestamp = columns.has("time_created") ? "time_created" : "NULL"; | |
| 131 | const predicates = table === "session_message" ? ["type = 'assistant'"] : []; | |
| 132 | if (timestamp !== "NULL") predicates.push("time_created >= ?"); | |
| 133 | const where = predicates.length > 0 ? ` WHERE ${predicates.join(" AND ")}` : ""; | |
| 134 | const statement = database.prepare( | |
| 135 | `SELECT id, session_id, data, ${timestamp} AS created FROM ${table}${where}`, | |
| 136 | ); | |
| 137 | let count = 0; | |
| 138 | for (const row of statement.iterate(...(timestamp === "NULL" ? [] : [sinceMs]))) { | |
| 139 | append( | |
| 140 | file.records, | |
| 141 | parseOpenCodeMessage(text(row.data), { | |
| 142 | id: text(row.id), | |
| 143 | sessionId: text(row.session_id), | |
| 144 | ...(typeof row.created === "number" ? { timestampMs: row.created } : {}), | |
| 145 | }), | |
| 146 | ); | |
| 147 | if (++count % 256 === 0) await NodeTimersPromises.setImmediate(); | |
| 148 | } | |
| 149 | } | |
| 150 | } catch { | |
| 151 | error = true; | |
| 152 | } finally { | |
| 153 | database?.close(); | |
| 154 | } | |
| 155 | } | |
| 156 | ||
| 157 | // Do not follow symlinks, including cycles. Database records win over their | |
| 158 | // old JSON copies when OpenCode has migrated a store in place. | |
| 159 | const directories = [NodePath.join(root, "storage", "message")]; | |
| 160 | while (directories.length > 0) { | |
| 161 | const directory = directories.pop()!; | |
| 162 | try { | |
| 163 | for (const entry of await NodeFSP.readdir(directory, { withFileTypes: true })) { | |
| 164 | const path = NodePath.join(directory, entry.name); | |
| 165 | if (entry.isDirectory()) { | |
| 166 | directories.push(path); | |
| 167 | } else if (entry.isFile() && entry.name.endsWith(".json")) { | |
| 168 | found = true; | |
| 169 | const id = entry.name.slice(0, -5); | |
| 170 | if (seen.has(`opencode:${id}`)) continue; | |
| 171 | const file = { path, records: [] as UsageRecord[] }; | |
| 172 | files.push(file); | |
| 173 | try { | |
| 174 | append( | |
| 175 | file.records, | |
| 176 | parseOpenCodeMessage(await NodeFSP.readFile(path, "utf8"), { id }), | |
| 177 | ); | |
| 178 | } catch (cause) { | |
| 179 | if (object(cause).code !== "ENOENT") error = true; | |
| 180 | } | |
| 181 | } | |
| 182 | } | |
| 183 | } catch (cause) { | |
| 184 | if (object(cause).code !== "ENOENT") error = true; | |
| 185 | } | |
| 186 | } | |
| 187 | return { files, missing: !found && !error, error }; | |
| 188 | } |
apps/server/src/usage/usageTranscriptReader.test.ts-50| ... | ... | @@ -10,7 +10,6 @@ import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest"; |
| 10 | 10 | |
| 11 | 11 | import { TEST_FORMATS } from "./usageTestFormats.ts"; |
| 12 | 12 | import { readTranscriptRecords } from "./usageTranscriptReader.ts"; |
| 13 | import { readOpenCodeUsage } from "./opencodeUsageReader.ts"; | |
| 14 | 13 | import { readCursorAccountUsage } from "./cursorUsageReader.ts"; |
| 15 | 14 | import { makeAntigravityUsageCache, readAntigravityUsage } from "./antigravityUsageReader.ts"; |
| 16 | 15 | |
| ... | ... | @@ -482,55 +481,6 @@ describe("SQLite usage readers", () => { |
| 482 | 481 | assert.isFalse(requested); |
| 483 | 482 | }); |
| 484 | 483 | |
| 485 | it("counts migrated OpenCode messages once and sees subsequent WAL writes", async () => { | |
| 486 | const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "opencode.db")); | |
| 487 | try { | |
| 488 | db.exec( | |
| 489 | "PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0; CREATE TABLE message (id TEXT, session_id TEXT, data TEXT)", | |
| 490 | ); | |
| 491 | const message = { | |
| 492 | id: "msg-1", | |
| 493 | sessionID: "session-1", | |
| 494 | role: "assistant", | |
| 495 | modelID: "claude-sonnet-4-5", | |
| 496 | time: { created: 1780000000000 }, | |
| 497 | cost: 0.25, | |
| 498 | tokens: { input: 100, output: 20, reasoning: 5, cache: { read: 30, write: 10 } }, | |
| 499 | }; | |
| 500 | const insert = db.prepare("INSERT INTO message VALUES (?, ?, ?)"); | |
| 501 | insert.run(message.id, message.sessionID, JSON.stringify(message)); | |
| 502 | const legacy = NodePath.join(dir, "storage", "message", message.sessionID); | |
| 503 | await NodeFSP.mkdir(legacy, { recursive: true }); | |
| 504 | await NodeFSP.writeFile(NodePath.join(legacy, "msg-1.json"), JSON.stringify(message)); | |
| 505 | const first = await readOpenCodeUsage(dir, 0); | |
| 506 | assert.isFalse(first.error); | |
| 507 | const records = first.files.flatMap((file) => file.records); | |
| 508 | assert.strictEqual(records.length, 1); | |
| 509 | assert.deepStrictEqual(records[0]?.totals, { | |
| 510 | uncachedInputTokens: 100, | |
| 511 | cachedInputTokens: 30, | |
| 512 | cacheCreationTokens: 10, | |
| 513 | outputTokens: 25, | |
| 514 | reasoningTokens: 5, | |
| 515 | }); | |
| 516 | assert.strictEqual(records[0]?.reportedCostUsd, 0.25); | |
| 517 | insert.run( | |
| 518 | "msg-2", | |
| 519 | message.sessionID, | |
| 520 | JSON.stringify({ ...message, id: "msg-2", time: { created: 1780000001000 } }), | |
| 521 | ); | |
| 522 | const next = await readOpenCodeUsage(dir, 1780000001000); | |
| 523 | assert.isFalse(next.error); | |
| 524 | assert.deepStrictEqual( | |
| 525 | next.files.flatMap((file) => file.records).map((record) => record.dedupeKey), | |
| 526 | ["opencode:msg-2"], | |
| 527 | ); | |
| 528 | assert.isAbove((await NodeFSP.stat(NodePath.join(dir, "opencode.db-wal"))).size, 0); | |
| 529 | } finally { | |
| 530 | db.close(); | |
| 531 | } | |
| 532 | }); | |
| 533 | ||
| 534 | 484 | it("deduplicates Antigravity generation and step usage while preserving retry model and token buckets", async () => { |
| 535 | 485 | const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "session-1.db")); |
| 536 | 486 | const stamp = protoNumber(1, 1780000000); |
packages/provider-core/src/server/usage.ts+2-1| ... | ... | @@ -129,7 +129,8 @@ export interface ProviderUsageInstance<Config> { |
| 129 | 129 | /** One source a `scan` reader read. */ |
| 130 | 130 | export interface ProviderUsageScan { |
| 131 | 131 | readonly dir: string; |
| 132 | readonly volumeId: string; | |
| 132 | /** Identity of the source across hosts. Defaults to the filesystem identity of `dir`. */ | |
| 133 | readonly volumeId?: string; | |
| 133 | 134 | /** Overrides the server's host name in the source fingerprint, for account-wide sources. */ |
| 134 | 135 | readonly hostId?: string; |
| 135 | 136 | readonly status?: UsageSource["status"]; |
packages/provider-opencode/src/server/driver.ts+7-1| ... | ... | @@ -34,6 +34,7 @@ import * as OpenCode2AdapterV2 from "./v2/adapter.ts"; |
| 34 | 34 | import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; |
| 35 | 35 | import type { ProviderTextGeneration } from "@t3tools/provider-core/server/textGeneration"; |
| 36 | 36 | import { ProviderDriverError } from "@t3tools/provider-core/server/errors"; |
| 37 | import { openCodeUsageReader, type OpenCodeUsageReaderEnv } from "./usage.ts"; | |
| 37 | 38 | import { readOpenCodeGoUsageLimits } from "./usageLimits.ts"; |
| 38 | 39 | import { |
| 39 | 40 | checkOpenCodeProviderStatus, |
| ... | ... | @@ -182,7 +183,11 @@ export type OpenCodeDriverEnv = |
| 182 | 183 | | OpenCodeRuntime.OpenCodeRuntime |
| 183 | 184 | | Path.Path; |
| 184 | 185 | |
| 185 | export const OpenCodeDriver: ProviderDriver<OpenCodeSettings, OpenCodeDriverEnv> = { | |
| 186 | export const OpenCodeDriver: ProviderDriver< | |
| 187 | OpenCodeSettings, | |
| 188 | OpenCodeDriverEnv, | |
| 189 | OpenCodeUsageReaderEnv | |
| 190 | > = { | |
| 186 | 191 | driverKind: DRIVER_KIND, |
| 187 | 192 | metadata: { |
| 188 | 193 | displayName: "OpenCode", |
| ... | ... | @@ -190,6 +195,7 @@ export const OpenCodeDriver: ProviderDriver<OpenCodeSettings, OpenCodeDriverEnv> |
| 190 | 195 | }, |
| 191 | 196 | configSchema: OpenCodeSettings, |
| 192 | 197 | defaultConfig: (): OpenCodeSettings => decodeOpenCodeSettings({}), |
| 198 | usage: openCodeUsageReader, | |
| 193 | 199 | create: ({ instanceId, displayName, accentColor, environment, enabled, config }) => |
| 194 | 200 | Effect.gen(function* () { |
| 195 | 201 | const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; |
packages/provider-opencode/src/server/usage.test.ts created+71| ... | ... | @@ -0,0 +1,71 @@ |
| 1 | // @effect-diagnostics nodeBuiltinImport:off - the reader under test opens real | |
| 2 | // SQLite databases and walks a real legacy JSON store. | |
| 3 | import * as NodeFSP from "node:fs/promises"; | |
| 4 | import * as NodeOS from "node:os"; | |
| 5 | import * as NodePath from "node:path"; | |
| 6 | import * as NodeSqlite from "node:sqlite"; | |
| 7 | ||
| 8 | import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest"; | |
| 9 | ||
| 10 | import { readOpenCodeUsage } from "./usage.ts"; | |
| 11 | ||
| 12 | let dir: string; | |
| 13 | ||
| 14 | beforeEach(async () => { | |
| 15 | dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "usage-reader-test-")); | |
| 16 | }); | |
| 17 | ||
| 18 | afterEach(async () => { | |
| 19 | await NodeFSP.rm(dir, { recursive: true, force: true }); | |
| 20 | }); | |
| 21 | ||
| 22 | describe("readOpenCodeUsage", () => { | |
| 23 | it("counts migrated OpenCode messages once and sees subsequent WAL writes", async () => { | |
| 24 | const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "opencode.db")); | |
| 25 | try { | |
| 26 | db.exec( | |
| 27 | "PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0; CREATE TABLE message (id TEXT, session_id TEXT, data TEXT)", | |
| 28 | ); | |
| 29 | const message = { | |
| 30 | id: "msg-1", | |
| 31 | sessionID: "session-1", | |
| 32 | role: "assistant", | |
| 33 | modelID: "claude-sonnet-4-5", | |
| 34 | time: { created: 1780000000000 }, | |
| 35 | cost: 0.25, | |
| 36 | tokens: { input: 100, output: 20, reasoning: 5, cache: { read: 30, write: 10 } }, | |
| 37 | }; | |
| 38 | const insert = db.prepare("INSERT INTO message VALUES (?, ?, ?)"); | |
| 39 | insert.run(message.id, message.sessionID, JSON.stringify(message)); | |
| 40 | const legacy = NodePath.join(dir, "storage", "message", message.sessionID); | |
| 41 | await NodeFSP.mkdir(legacy, { recursive: true }); | |
| 42 | await NodeFSP.writeFile(NodePath.join(legacy, "msg-1.json"), JSON.stringify(message)); | |
| 43 | const first = await readOpenCodeUsage(dir, 0); | |
| 44 | assert.isFalse(first.error); | |
| 45 | const records = first.files.flatMap((file) => file.records); | |
| 46 | assert.strictEqual(records.length, 1); | |
| 47 | assert.deepStrictEqual(records[0]?.totals, { | |
| 48 | uncachedInputTokens: 100, | |
| 49 | cachedInputTokens: 30, | |
| 50 | cacheCreationTokens: 10, | |
| 51 | outputTokens: 25, | |
| 52 | reasoningTokens: 5, | |
| 53 | }); | |
| 54 | assert.strictEqual(records[0]?.reportedCostUsd, 0.25); | |
| 55 | insert.run( | |
| 56 | "msg-2", | |
| 57 | message.sessionID, | |
| 58 | JSON.stringify({ ...message, id: "msg-2", time: { created: 1780000001000 } }), | |
| 59 | ); | |
| 60 | const next = await readOpenCodeUsage(dir, 1780000001000); | |
| 61 | assert.isFalse(next.error); | |
| 62 | assert.deepStrictEqual( | |
| 63 | next.files.flatMap((file) => file.records).map((record) => record.dedupeKey), | |
| 64 | ["opencode:msg-2"], | |
| 65 | ); | |
| 66 | assert.isAbove((await NodeFSP.stat(NodePath.join(dir, "opencode.db-wal"))).size, 0); | |
| 67 | } finally { | |
| 68 | db.close(); | |
| 69 | } | |
| 70 | }); | |
| 71 | }); |
packages/provider-opencode/src/server/usage.ts created+258| ... | ... | @@ -0,0 +1,258 @@ |
| 1 | // node:sqlite reads live OpenCode databases; Node fs walks legacy JSON history. | |
| 2 | // @effect-diagnostics nodeBuiltinImport:off | |
| 3 | /** | |
| 4 | * Usage history for OpenCode, read from its SQLite databases and legacy JSON | |
| 5 | * message store under each data directory. | |
| 6 | * | |
| 7 | * @module provider-opencode/server/usage | |
| 8 | */ | |
| 9 | import * as NodeFSP from "node:fs/promises"; | |
| 10 | import * as NodeOS from "node:os"; | |
| 11 | import * as NodePath from "node:path"; | |
| 12 | import * as NodeSqlite from "node:sqlite"; | |
| 13 | import * as NodeTimersPromises from "node:timers/promises"; | |
| 14 | ||
| 15 | import { expandHomePath } from "@t3tools/provider-core/server/pathExpansion"; | |
| 16 | import { | |
| 17 | totalTokens, | |
| 18 | type ProviderUsageReader, | |
| 19 | type UsageRecord, | |
| 20 | } from "@t3tools/provider-core/server/usage"; | |
| 21 | import { HostProcessEnvironment } from "@t3tools/shared/hostProcess"; | |
| 22 | import * as Effect from "effect/Effect"; | |
| 23 | import * as FileSystem from "effect/FileSystem"; | |
| 24 | import * as Path from "effect/Path"; | |
| 25 | ||
| 26 | import type { OpenCodeSettings } from "../settings.ts"; | |
| 27 | ||
| 28 | function object(value: unknown): Record<string, unknown> { | |
| 29 | return typeof value === "object" && value !== null && !Array.isArray(value) | |
| 30 | ? (value as Record<string, unknown>) | |
| 31 | : {}; | |
| 32 | } | |
| 33 | ||
| 34 | function tokens(value: unknown): number { | |
| 35 | return typeof value === "number" && Number.isFinite(value) && value > 0 ? Math.trunc(value) : 0; | |
| 36 | } | |
| 37 | ||
| 38 | function text(value: unknown): string { | |
| 39 | return typeof value === "string" ? value : ""; | |
| 40 | } | |
| 41 | ||
| 42 | /** OpenCode stores uncached input and reasoning separately from input/output. */ | |
| 43 | function parseOpenCodeMessage( | |
| 44 | source: string, | |
| 45 | fallback: { | |
| 46 | readonly id?: string; | |
| 47 | readonly sessionId?: string; | |
| 48 | readonly timestampMs?: number; | |
| 49 | } = {}, | |
| 50 | ): UsageRecord | null { | |
| 51 | let parsed: unknown; | |
| 52 | try { | |
| 53 | parsed = JSON.parse(source); | |
| 54 | } catch { | |
| 55 | return null; | |
| 56 | } | |
| 57 | const message = object(parsed); | |
| 58 | if (message.role !== undefined && message.role !== "assistant") return null; | |
| 59 | const usage = object(message.tokens); | |
| 60 | const cache = object(usage.cache); | |
| 61 | const modelReference = object(message.model); | |
| 62 | const model = text(modelReference.id) || text(modelReference.modelID) || text(message.modelID); | |
| 63 | const timestampMs = object(message.time).created ?? fallback.timestampMs; | |
| 64 | if (!model || typeof timestampMs !== "number" || !Number.isFinite(timestampMs)) return null; | |
| 65 | const reasoningTokens = tokens(usage.reasoning); | |
| 66 | const totals = { | |
| 67 | uncachedInputTokens: tokens(usage.input), | |
| 68 | cachedInputTokens: tokens(cache.read), | |
| 69 | cacheCreationTokens: tokens(cache.write), | |
| 70 | outputTokens: tokens(usage.output) + reasoningTokens, | |
| 71 | reasoningTokens, | |
| 72 | }; | |
| 73 | if (totalTokens(totals) === 0) return null; | |
| 74 | const id = fallback.id || text(message.id); | |
| 75 | const cost = message.cost; | |
| 76 | return { | |
| 77 | provider: "opencode", | |
| 78 | timestampMs, | |
| 79 | model, | |
| 80 | sessionId: fallback.sessionId || text(message.sessionID), | |
| 81 | totals, | |
| 82 | // OpenCode writes zero for models without a known rate, including paid | |
| 83 | // subscription models. Let the shared price table estimate those records. | |
| 84 | reportedCostUsd: typeof cost === "number" && Number.isFinite(cost) && cost > 0 ? cost : null, | |
| 85 | speed: "standard", | |
| 86 | dedupeKey: id ? `opencode:${id}` : null, | |
| 87 | }; | |
| 88 | } | |
| 89 | ||
| 90 | export interface OpenCodeUsageReadResult { | |
| 91 | readonly files: readonly { readonly path: string; readonly records: readonly UsageRecord[] }[]; | |
| 92 | readonly missing: boolean; | |
| 93 | readonly error: boolean; | |
| 94 | } | |
| 95 | ||
| 96 | /** Reads current SQLite and pre-migration JSON stores without modifying either. */ | |
| 97 | export async function readOpenCodeUsage( | |
| 98 | root: string, | |
| 99 | sinceMs: number, | |
| 100 | ): Promise<OpenCodeUsageReadResult> { | |
| 101 | const files: { path: string; records: UsageRecord[] }[] = []; | |
| 102 | const seen = new Set<string>(); | |
| 103 | let found = false; | |
| 104 | let error = false; | |
| 105 | const append = (records: UsageRecord[], record: UsageRecord | null) => { | |
| 106 | if (record === null || record.timestampMs < sinceMs) return; | |
| 107 | if (record.dedupeKey !== null) { | |
| 108 | if (seen.has(record.dedupeKey)) return; | |
| 109 | seen.add(record.dedupeKey); | |
| 110 | } | |
| 111 | records.push(record); | |
| 112 | }; | |
| 113 | ||
| 114 | let databases: string[] = []; | |
| 115 | try { | |
| 116 | databases = (await NodeFSP.readdir(root, { withFileTypes: true })) | |
| 117 | .filter((entry) => entry.isFile() && /^opencode(?:-[a-zA-Z0-9_-]+)?\.db$/.test(entry.name)) | |
| 118 | .map((entry) => entry.name) | |
| 119 | .sort((a, b) => (a === "opencode.db" ? -1 : b === "opencode.db" ? 1 : a.localeCompare(b))); | |
| 120 | } catch (cause) { | |
| 121 | if (object(cause).code !== "ENOENT") error = true; | |
| 122 | } | |
| 123 | for (const name of databases) { | |
| 124 | found = true; | |
| 125 | const file = { path: NodePath.join(root, name), records: [] as UsageRecord[] }; | |
| 126 | files.push(file); | |
| 127 | let database: NodeSqlite.DatabaseSync | undefined; | |
| 128 | try { | |
| 129 | database = new NodeSqlite.DatabaseSync(NodePath.join(root, name), { readOnly: true }); | |
| 130 | // A busy live provider should fail this source promptly rather than | |
| 131 | // stalling the server while SQLite waits for its writer. | |
| 132 | database.exec("PRAGMA busy_timeout = 100"); | |
| 133 | const tables = new Set( | |
| 134 | database | |
| 135 | .prepare("SELECT name FROM sqlite_master WHERE type = 'table'") | |
| 136 | .all() | |
| 137 | .map((row) => row.name), | |
| 138 | ); | |
| 139 | if (!tables.has("message") && !tables.has("session_message")) error = true; | |
| 140 | for (const table of ["message", "session_message"] as const) { | |
| 141 | if (!tables.has(table)) continue; | |
| 142 | const columns = new Set( | |
| 143 | database | |
| 144 | .prepare(`PRAGMA table_info(${table})`) | |
| 145 | .all() | |
| 146 | .map((row) => row.name), | |
| 147 | ); | |
| 148 | const timestamp = columns.has("time_created") ? "time_created" : "NULL"; | |
| 149 | const predicates = table === "session_message" ? ["type = 'assistant'"] : []; | |
| 150 | if (timestamp !== "NULL") predicates.push("time_created >= ?"); | |
| 151 | const where = predicates.length > 0 ? ` WHERE ${predicates.join(" AND ")}` : ""; | |
| 152 | const statement = database.prepare( | |
| 153 | `SELECT id, session_id, data, ${timestamp} AS created FROM ${table}${where}`, | |
| 154 | ); | |
| 155 | let count = 0; | |
| 156 | for (const row of statement.iterate(...(timestamp === "NULL" ? [] : [sinceMs]))) { | |
| 157 | append( | |
| 158 | file.records, | |
| 159 | parseOpenCodeMessage(text(row.data), { | |
| 160 | id: text(row.id), | |
| 161 | sessionId: text(row.session_id), | |
| 162 | ...(typeof row.created === "number" ? { timestampMs: row.created } : {}), | |
| 163 | }), | |
| 164 | ); | |
| 165 | if (++count % 256 === 0) await NodeTimersPromises.setImmediate(); | |
| 166 | } | |
| 167 | } | |
| 168 | } catch { | |
| 169 | error = true; | |
| 170 | } finally { | |
| 171 | database?.close(); | |
| 172 | } | |
| 173 | } | |
| 174 | ||
| 175 | // Do not follow symlinks, including cycles. Database records win over their | |
| 176 | // old JSON copies when OpenCode has migrated a store in place. | |
| 177 | const directories = [NodePath.join(root, "storage", "message")]; | |
| 178 | while (directories.length > 0) { | |
| 179 | const directory = directories.pop()!; | |
| 180 | try { | |
| 181 | for (const entry of await NodeFSP.readdir(directory, { withFileTypes: true })) { | |
| 182 | const path = NodePath.join(directory, entry.name); | |
| 183 | if (entry.isDirectory()) { | |
| 184 | directories.push(path); | |
| 185 | } else if (entry.isFile() && entry.name.endsWith(".json")) { | |
| 186 | found = true; | |
| 187 | const id = entry.name.slice(0, -5); | |
| 188 | if (seen.has(`opencode:${id}`)) continue; | |
| 189 | const file = { path, records: [] as UsageRecord[] }; | |
| 190 | files.push(file); | |
| 191 | try { | |
| 192 | append( | |
| 193 | file.records, | |
| 194 | parseOpenCodeMessage(await NodeFSP.readFile(path, "utf8"), { id }), | |
| 195 | ); | |
| 196 | } catch (cause) { | |
| 197 | if (object(cause).code !== "ENOENT") error = true; | |
| 198 | } | |
| 199 | } | |
| 200 | } | |
| 201 | } catch (cause) { | |
| 202 | if (object(cause).code !== "ENOENT") error = true; | |
| 203 | } | |
| 204 | } | |
| 205 | return { files, missing: !found && !error, error }; | |
| 206 | } | |
| 207 | ||
| 208 | /** | |
| 209 | * The data directories to read: `OPENCODE_DATA_DIR` (comma-separated) or the | |
| 210 | * XDG default, canonicalized so aliases count once. | |
| 211 | */ | |
| 212 | const resolveOpenCodeDataDirs = Effect.fn("resolveOpenCodeDataDirs")(function* () { | |
| 213 | const fileSystem = yield* FileSystem.FileSystem; | |
| 214 | const path = yield* Path.Path; | |
| 215 | const environment = yield* HostProcessEnvironment; | |
| 216 | const roots = environment["OPENCODE_DATA_DIR"] | |
| 217 | ?.split(",") | |
| 218 | .map((value) => value.trim()) | |
| 219 | .filter(Boolean); | |
| 220 | const dataHome = environment["XDG_DATA_HOME"]?.trim(); | |
| 221 | const defaults = [ | |
| 222 | path.join( | |
| 223 | dataHome && path.isAbsolute(dataHome) | |
| 224 | ? dataHome | |
| 225 | : path.join(NodeOS.homedir(), ".local", "share"), | |
| 226 | "opencode", | |
| 227 | ), | |
| 228 | ]; | |
| 229 | const canonical = new Set<string>(); | |
| 230 | for (const root of roots?.length ? roots : defaults) { | |
| 231 | const resolved = path.resolve(expandHomePath(root)); | |
| 232 | canonical.add(yield* fileSystem.realPath(resolved).pipe(Effect.orElseSucceed(() => resolved))); | |
| 233 | } | |
| 234 | return [...canonical]; | |
| 235 | }); | |
| 236 | ||
| 237 | export type OpenCodeUsageReaderEnv = FileSystem.FileSystem | Path.Path; | |
| 238 | ||
| 239 | export const openCodeUsageReader: ProviderUsageReader<OpenCodeSettings, OpenCodeUsageReaderEnv> = { | |
| 240 | kind: "scan", | |
| 241 | provider: "opencode", | |
| 242 | scan: Effect.fn("openCodeUsageReader.scan")(function* ({ windowStartMs }) { | |
| 243 | const roots = yield* resolveOpenCodeDataDirs(); | |
| 244 | return yield* Effect.forEach( | |
| 245 | roots, | |
| 246 | (dir) => | |
| 247 | Effect.promise(() => readOpenCodeUsage(dir, windowStartMs)).pipe( | |
| 248 | Effect.map((result) => ({ | |
| 249 | dir, | |
| 250 | files: result.missing && !result.error ? null : result.files, | |
| 251 | status: result.error ? ("partial" as const) : ("ok" as const), | |
| 252 | ...(result.error ? { message: "Some OpenCode history could not be read." } : {}), | |
| 253 | })), | |
| 254 | ), | |
| 255 | { concurrency: "unbounded" }, | |
| 256 | ); | |
| 257 | }), | |
| 258 | }; |