diff --git a/apps/server/src/provider/builtInDrivers.ts b/apps/server/src/provider/builtInDrivers.ts index 4cecbc6ca50b8512749627f5d761d3a1b451af42..4576f4ea36bbcbf117b87e14d7eeee0ca5a4b5e3 100644 --- a/apps/server/src/provider/builtInDrivers.ts +++ b/apps/server/src/provider/builtInDrivers.ts @@ -74,7 +74,8 @@ export const BUILT_IN_DRIVERS: ReadonlyArray | ProviderUsageReaderEnv - | ProviderUsageReaderEnv; + | ProviderUsageReaderEnv + | ProviderUsageReaderEnv; /** * The drivers that keep usage history, in the order the usage page reads @@ -83,4 +84,4 @@ export type BuiltInUsageReadersEnv = */ export const BUILT_IN_USAGE_DRIVERS: ReadonlyArray< AnyProviderDriver -> = [ClaudeDriver, CodexDriver, GrokDriver]; +> = [ClaudeDriver, CodexDriver, GrokDriver, OpenCodeDriver]; diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 94d967286f45c12522542f9eceb2946881a0621a..c8a1891ce8f865d0bc650a829bcce7cf175dbc01 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -63,7 +63,6 @@ import type { TranscriptUsageFormat, UsageRecord, } from "@t3tools/provider-core/server/usage"; -import { readOpenCodeUsage } from "./opencodeUsageReader.ts"; import { makeAntigravityUsageCache, readAntigravityUsage } from "./antigravityUsageReader.ts"; import { CURSOR_ACCOUNT_CACHE_FILE_NAME, @@ -127,6 +126,11 @@ const transcriptReaders = BUILT_IN_USAGE_DRIVERS.flatMap((driver) => driver.usage?.kind === "transcripts" ? [{ driver, reader: driver.usage }] : [], ); +/** The scan readers, in driver order. */ +const scanReaders = BUILT_IN_USAGE_DRIVERS.flatMap((driver) => + driver.usage?.kind === "scan" ? [{ driver, reader: driver.usage }] : [], +); + /** Transcript formats by provider, for decoding the persisted scan cache. */ const transcriptFormats = new Map( transcriptReaders.map(({ reader }) => [reader.provider, reader.format] as const), @@ -851,32 +855,36 @@ export const make = Effect.gen(function* () { } return [...canonical]; }); - const dataHome = hostEnvironment["XDG_DATA_HOME"]?.trim(); - const openCode = Effect.gen(function* () { - const roots = yield* envRoots("OPENCODE_DATA_DIR", [ - path.join( - dataHome && path.isAbsolute(dataHome) ? dataHome : path.join(home, ".local", "share"), - "opencode", - ), - ]); - return yield* Effect.forEach( - roots, - (dir) => - Effect.gen(function* () { - const result = yield* Effect.promise(() => readOpenCodeUsage(dir, windowStartMs)); - return { - provider: "opencode", - dir, - volumeId: yield* Effect.promise(() => readDirectoryVolumeId(dir)), - files: result.missing && !result.error ? null : result.files, - status: result.error ? "partial" : "ok", - ...(result.error ? { message: "Some OpenCode history could not be read." } : {}), - } satisfies ScannedDir; - }), - { concurrency: "unbounded" }, - ); - }); + const scans = Effect.forEach( + scanReaders, + ({ driver, reader }) => + reader + .scan({ + instances: usageInstances(driver, settings), + windowStartMs, + retentionCutoffMs, + awaitRefresh, + }) + .pipe( + Effect.flatMap((sources) => + Effect.forEach(sources, ({ volumeId, ...source }) => + Effect.map( + volumeId === undefined + ? Effect.promise(() => readDirectoryVolumeId(source.dir)) + : Effect.succeed(volumeId), + (resolved): ScannedDir => ({ + ...source, + provider: reader.provider, + volumeId: resolved, + }), + ), + ), + ), + Effect.provideContext(readerContext), + ), + { concurrency: "unbounded" }, + ).pipe(Effect.map((sources) => sources.flat())); const antigravity = Effect.gen(function* () { const antigravityRoots = yield* envRoots("ANTIGRAVITY_DATA_DIR", [ @@ -999,10 +1007,10 @@ export const make = Effect.gen(function* () { // Independent sources scan together. Transcript directories go one at a // time, so open files stay at `TRANSCRIPT_READ_CONCURRENCY`. The result // keeps this order, since aggregation keeps the first copy of a duplicate. - const [transcripts, openCodeDirs, antigravityDirs, cursorDirs] = yield* Effect.all( + const [transcripts, scanDirs, antigravityDirs, cursorDirs] = yield* Effect.all( [ Effect.forEach(dirs, (dir) => scanTranscriptDir(dir, windowStartMs)), - openCode, + scans, antigravity, cursor, ], @@ -1010,7 +1018,7 @@ export const make = Effect.gen(function* () { ); const scanned: readonly ScannedDir[] = [ ...transcripts, - ...openCodeDirs, + ...scanDirs, ...antigravityDirs, ...cursorDirs, ]; diff --git a/apps/server/src/usage/opencodeUsageReader.ts b/apps/server/src/usage/opencodeUsageReader.ts deleted file mode 100644 index d6a835558ed6d01cc89760bfa61a5c12d886dbda..0000000000000000000000000000000000000000 --- a/apps/server/src/usage/opencodeUsageReader.ts +++ /dev/null @@ -1,188 +0,0 @@ -// node:sqlite reads live OpenCode databases; Node fs walks legacy JSON history. -// @effect-diagnostics nodeBuiltinImport:off -import * as NodeFSP from "node:fs/promises"; -import * as NodePath from "node:path"; -import * as NodeSqlite from "node:sqlite"; -import * as NodeTimersPromises from "node:timers/promises"; - -import { totalTokens, type UsageRecord } from "@t3tools/provider-core/server/usage"; - -function object(value: unknown): Record { - return typeof value === "object" && value !== null && !Array.isArray(value) - ? (value as Record) - : {}; -} - -function tokens(value: unknown): number { - return typeof value === "number" && Number.isFinite(value) && value > 0 ? Math.trunc(value) : 0; -} - -function text(value: unknown): string { - return typeof value === "string" ? value : ""; -} - -/** OpenCode stores uncached input and reasoning separately from input/output. */ -function parseOpenCodeMessage( - source: string, - fallback: { - readonly id?: string; - readonly sessionId?: string; - readonly timestampMs?: number; - } = {}, -): UsageRecord | null { - let parsed: unknown; - try { - parsed = JSON.parse(source); - } catch { - return null; - } - const message = object(parsed); - if (message.role !== undefined && message.role !== "assistant") return null; - const usage = object(message.tokens); - const cache = object(usage.cache); - const modelReference = object(message.model); - const model = text(modelReference.id) || text(modelReference.modelID) || text(message.modelID); - const timestampMs = object(message.time).created ?? fallback.timestampMs; - if (!model || typeof timestampMs !== "number" || !Number.isFinite(timestampMs)) return null; - const reasoningTokens = tokens(usage.reasoning); - const totals = { - uncachedInputTokens: tokens(usage.input), - cachedInputTokens: tokens(cache.read), - cacheCreationTokens: tokens(cache.write), - outputTokens: tokens(usage.output) + reasoningTokens, - reasoningTokens, - }; - if (totalTokens(totals) === 0) return null; - const id = fallback.id || text(message.id); - const cost = message.cost; - return { - provider: "opencode", - timestampMs, - model, - sessionId: fallback.sessionId || text(message.sessionID), - totals, - // OpenCode writes zero for models without a known rate, including paid - // subscription models. Let the shared price table estimate those records. - reportedCostUsd: typeof cost === "number" && Number.isFinite(cost) && cost > 0 ? cost : null, - speed: "standard", - dedupeKey: id ? `opencode:${id}` : null, - }; -} - -export interface OpenCodeUsageReadResult { - readonly files: readonly { readonly path: string; readonly records: readonly UsageRecord[] }[]; - readonly missing: boolean; - readonly error: boolean; -} - -/** Reads current SQLite and pre-migration JSON stores without modifying either. */ -export async function readOpenCodeUsage( - root: string, - sinceMs: number, -): Promise { - const files: { path: string; records: UsageRecord[] }[] = []; - const seen = new Set(); - let found = false; - let error = false; - const append = (records: UsageRecord[], record: UsageRecord | null) => { - if (record === null || record.timestampMs < sinceMs) return; - if (record.dedupeKey !== null) { - if (seen.has(record.dedupeKey)) return; - seen.add(record.dedupeKey); - } - records.push(record); - }; - - let databases: string[] = []; - try { - databases = (await NodeFSP.readdir(root, { withFileTypes: true })) - .filter((entry) => entry.isFile() && /^opencode(?:-[a-zA-Z0-9_-]+)?\.db$/.test(entry.name)) - .map((entry) => entry.name) - .sort((a, b) => (a === "opencode.db" ? -1 : b === "opencode.db" ? 1 : a.localeCompare(b))); - } catch (cause) { - if (object(cause).code !== "ENOENT") error = true; - } - for (const name of databases) { - found = true; - const file = { path: NodePath.join(root, name), records: [] as UsageRecord[] }; - files.push(file); - let database: NodeSqlite.DatabaseSync | undefined; - try { - database = new NodeSqlite.DatabaseSync(NodePath.join(root, name), { readOnly: true }); - // A busy live provider should fail this source promptly rather than - // stalling the server while SQLite waits for its writer. - database.exec("PRAGMA busy_timeout = 100"); - const tables = new Set( - database - .prepare("SELECT name FROM sqlite_master WHERE type = 'table'") - .all() - .map((row) => row.name), - ); - if (!tables.has("message") && !tables.has("session_message")) error = true; - for (const table of ["message", "session_message"] as const) { - if (!tables.has(table)) continue; - const columns = new Set( - database - .prepare(`PRAGMA table_info(${table})`) - .all() - .map((row) => row.name), - ); - const timestamp = columns.has("time_created") ? "time_created" : "NULL"; - const predicates = table === "session_message" ? ["type = 'assistant'"] : []; - if (timestamp !== "NULL") predicates.push("time_created >= ?"); - const where = predicates.length > 0 ? ` WHERE ${predicates.join(" AND ")}` : ""; - const statement = database.prepare( - `SELECT id, session_id, data, ${timestamp} AS created FROM ${table}${where}`, - ); - let count = 0; - for (const row of statement.iterate(...(timestamp === "NULL" ? [] : [sinceMs]))) { - append( - file.records, - parseOpenCodeMessage(text(row.data), { - id: text(row.id), - sessionId: text(row.session_id), - ...(typeof row.created === "number" ? { timestampMs: row.created } : {}), - }), - ); - if (++count % 256 === 0) await NodeTimersPromises.setImmediate(); - } - } - } catch { - error = true; - } finally { - database?.close(); - } - } - - // Do not follow symlinks, including cycles. Database records win over their - // old JSON copies when OpenCode has migrated a store in place. - const directories = [NodePath.join(root, "storage", "message")]; - while (directories.length > 0) { - const directory = directories.pop()!; - try { - for (const entry of await NodeFSP.readdir(directory, { withFileTypes: true })) { - const path = NodePath.join(directory, entry.name); - if (entry.isDirectory()) { - directories.push(path); - } else if (entry.isFile() && entry.name.endsWith(".json")) { - found = true; - const id = entry.name.slice(0, -5); - if (seen.has(`opencode:${id}`)) continue; - const file = { path, records: [] as UsageRecord[] }; - files.push(file); - try { - append( - file.records, - parseOpenCodeMessage(await NodeFSP.readFile(path, "utf8"), { id }), - ); - } catch (cause) { - if (object(cause).code !== "ENOENT") error = true; - } - } - } - } catch (cause) { - if (object(cause).code !== "ENOENT") error = true; - } - } - return { files, missing: !found && !error, error }; -} diff --git a/apps/server/src/usage/usageTranscriptReader.test.ts b/apps/server/src/usage/usageTranscriptReader.test.ts index 183424cce3a7e3d3c5c3081f6fef89755da808d8..60fdc8f27367236f45aab6a47ff8c7e47cd428e0 100644 --- a/apps/server/src/usage/usageTranscriptReader.test.ts +++ b/apps/server/src/usage/usageTranscriptReader.test.ts @@ -10,7 +10,6 @@ import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest"; import { TEST_FORMATS } from "./usageTestFormats.ts"; import { readTranscriptRecords } from "./usageTranscriptReader.ts"; -import { readOpenCodeUsage } from "./opencodeUsageReader.ts"; import { readCursorAccountUsage } from "./cursorUsageReader.ts"; import { makeAntigravityUsageCache, readAntigravityUsage } from "./antigravityUsageReader.ts"; @@ -482,55 +481,6 @@ describe("SQLite usage readers", () => { assert.isFalse(requested); }); - it("counts migrated OpenCode messages once and sees subsequent WAL writes", async () => { - const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "opencode.db")); - try { - db.exec( - "PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0; CREATE TABLE message (id TEXT, session_id TEXT, data TEXT)", - ); - const message = { - id: "msg-1", - sessionID: "session-1", - role: "assistant", - modelID: "claude-sonnet-4-5", - time: { created: 1780000000000 }, - cost: 0.25, - tokens: { input: 100, output: 20, reasoning: 5, cache: { read: 30, write: 10 } }, - }; - const insert = db.prepare("INSERT INTO message VALUES (?, ?, ?)"); - insert.run(message.id, message.sessionID, JSON.stringify(message)); - const legacy = NodePath.join(dir, "storage", "message", message.sessionID); - await NodeFSP.mkdir(legacy, { recursive: true }); - await NodeFSP.writeFile(NodePath.join(legacy, "msg-1.json"), JSON.stringify(message)); - const first = await readOpenCodeUsage(dir, 0); - assert.isFalse(first.error); - const records = first.files.flatMap((file) => file.records); - assert.strictEqual(records.length, 1); - assert.deepStrictEqual(records[0]?.totals, { - uncachedInputTokens: 100, - cachedInputTokens: 30, - cacheCreationTokens: 10, - outputTokens: 25, - reasoningTokens: 5, - }); - assert.strictEqual(records[0]?.reportedCostUsd, 0.25); - insert.run( - "msg-2", - message.sessionID, - JSON.stringify({ ...message, id: "msg-2", time: { created: 1780000001000 } }), - ); - const next = await readOpenCodeUsage(dir, 1780000001000); - assert.isFalse(next.error); - assert.deepStrictEqual( - next.files.flatMap((file) => file.records).map((record) => record.dedupeKey), - ["opencode:msg-2"], - ); - assert.isAbove((await NodeFSP.stat(NodePath.join(dir, "opencode.db-wal"))).size, 0); - } finally { - db.close(); - } - }); - it("deduplicates Antigravity generation and step usage while preserving retry model and token buckets", async () => { const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "session-1.db")); const stamp = protoNumber(1, 1780000000); diff --git a/packages/provider-core/src/server/usage.ts b/packages/provider-core/src/server/usage.ts index f2cf52e679651ca7285527fca4d3a92b7e3d9139..9371797073163441e856396e7f8c94ae619d0a51 100644 --- a/packages/provider-core/src/server/usage.ts +++ b/packages/provider-core/src/server/usage.ts @@ -129,7 +129,8 @@ export interface ProviderUsageInstance { /** One source a `scan` reader read. */ export interface ProviderUsageScan { readonly dir: string; - readonly volumeId: string; + /** Identity of the source across hosts. Defaults to the filesystem identity of `dir`. */ + readonly volumeId?: string; /** Overrides the server's host name in the source fingerprint, for account-wide sources. */ readonly hostId?: string; readonly status?: UsageSource["status"]; diff --git a/packages/provider-opencode/src/server/driver.ts b/packages/provider-opencode/src/server/driver.ts index 3834ec73534bcea43d9950bf394f7447c892ddca..0573db664969025d074f899b86b164732b550797 100644 --- a/packages/provider-opencode/src/server/driver.ts +++ b/packages/provider-opencode/src/server/driver.ts @@ -34,6 +34,7 @@ import * as OpenCode2AdapterV2 from "./v2/adapter.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import type { ProviderTextGeneration } from "@t3tools/provider-core/server/textGeneration"; import { ProviderDriverError } from "@t3tools/provider-core/server/errors"; +import { openCodeUsageReader, type OpenCodeUsageReaderEnv } from "./usage.ts"; import { readOpenCodeGoUsageLimits } from "./usageLimits.ts"; import { checkOpenCodeProviderStatus, @@ -182,7 +183,11 @@ export type OpenCodeDriverEnv = | OpenCodeRuntime.OpenCodeRuntime | Path.Path; -export const OpenCodeDriver: ProviderDriver = { +export const OpenCodeDriver: ProviderDriver< + OpenCodeSettings, + OpenCodeDriverEnv, + OpenCodeUsageReaderEnv +> = { driverKind: DRIVER_KIND, metadata: { displayName: "OpenCode", @@ -190,6 +195,7 @@ export const OpenCodeDriver: ProviderDriver }, configSchema: OpenCodeSettings, defaultConfig: (): OpenCodeSettings => decodeOpenCodeSettings({}), + usage: openCodeUsageReader, create: ({ instanceId, displayName, accentColor, environment, enabled, config }) => Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; diff --git a/packages/provider-opencode/src/server/usage.test.ts b/packages/provider-opencode/src/server/usage.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..8389f85a5f3f4a83778514836a0b676c4ffbd40d --- /dev/null +++ b/packages/provider-opencode/src/server/usage.test.ts @@ -0,0 +1,71 @@ +// @effect-diagnostics nodeBuiltinImport:off - the reader under test opens real +// SQLite databases and walks a real legacy JSON store. +import * as NodeFSP from "node:fs/promises"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; +import * as NodeSqlite from "node:sqlite"; + +import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest"; + +import { readOpenCodeUsage } from "./usage.ts"; + +let dir: string; + +beforeEach(async () => { + dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "usage-reader-test-")); +}); + +afterEach(async () => { + await NodeFSP.rm(dir, { recursive: true, force: true }); +}); + +describe("readOpenCodeUsage", () => { + it("counts migrated OpenCode messages once and sees subsequent WAL writes", async () => { + const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "opencode.db")); + try { + db.exec( + "PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0; CREATE TABLE message (id TEXT, session_id TEXT, data TEXT)", + ); + const message = { + id: "msg-1", + sessionID: "session-1", + role: "assistant", + modelID: "claude-sonnet-4-5", + time: { created: 1780000000000 }, + cost: 0.25, + tokens: { input: 100, output: 20, reasoning: 5, cache: { read: 30, write: 10 } }, + }; + const insert = db.prepare("INSERT INTO message VALUES (?, ?, ?)"); + insert.run(message.id, message.sessionID, JSON.stringify(message)); + const legacy = NodePath.join(dir, "storage", "message", message.sessionID); + await NodeFSP.mkdir(legacy, { recursive: true }); + await NodeFSP.writeFile(NodePath.join(legacy, "msg-1.json"), JSON.stringify(message)); + const first = await readOpenCodeUsage(dir, 0); + assert.isFalse(first.error); + const records = first.files.flatMap((file) => file.records); + assert.strictEqual(records.length, 1); + assert.deepStrictEqual(records[0]?.totals, { + uncachedInputTokens: 100, + cachedInputTokens: 30, + cacheCreationTokens: 10, + outputTokens: 25, + reasoningTokens: 5, + }); + assert.strictEqual(records[0]?.reportedCostUsd, 0.25); + insert.run( + "msg-2", + message.sessionID, + JSON.stringify({ ...message, id: "msg-2", time: { created: 1780000001000 } }), + ); + const next = await readOpenCodeUsage(dir, 1780000001000); + assert.isFalse(next.error); + assert.deepStrictEqual( + next.files.flatMap((file) => file.records).map((record) => record.dedupeKey), + ["opencode:msg-2"], + ); + assert.isAbove((await NodeFSP.stat(NodePath.join(dir, "opencode.db-wal"))).size, 0); + } finally { + db.close(); + } + }); +}); diff --git a/packages/provider-opencode/src/server/usage.ts b/packages/provider-opencode/src/server/usage.ts new file mode 100644 index 0000000000000000000000000000000000000000..9cc5ce2d09109a74c7ac016757835476c189fde8 --- /dev/null +++ b/packages/provider-opencode/src/server/usage.ts @@ -0,0 +1,258 @@ +// node:sqlite reads live OpenCode databases; Node fs walks legacy JSON history. +// @effect-diagnostics nodeBuiltinImport:off +/** + * Usage history for OpenCode, read from its SQLite databases and legacy JSON + * message store under each data directory. + * + * @module provider-opencode/server/usage + */ +import * as NodeFSP from "node:fs/promises"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; +import * as NodeSqlite from "node:sqlite"; +import * as NodeTimersPromises from "node:timers/promises"; + +import { expandHomePath } from "@t3tools/provider-core/server/pathExpansion"; +import { + totalTokens, + type ProviderUsageReader, + type UsageRecord, +} from "@t3tools/provider-core/server/usage"; +import { HostProcessEnvironment } from "@t3tools/shared/hostProcess"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; + +import type { OpenCodeSettings } from "../settings.ts"; + +function object(value: unknown): Record { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? (value as Record) + : {}; +} + +function tokens(value: unknown): number { + return typeof value === "number" && Number.isFinite(value) && value > 0 ? Math.trunc(value) : 0; +} + +function text(value: unknown): string { + return typeof value === "string" ? value : ""; +} + +/** OpenCode stores uncached input and reasoning separately from input/output. */ +function parseOpenCodeMessage( + source: string, + fallback: { + readonly id?: string; + readonly sessionId?: string; + readonly timestampMs?: number; + } = {}, +): UsageRecord | null { + let parsed: unknown; + try { + parsed = JSON.parse(source); + } catch { + return null; + } + const message = object(parsed); + if (message.role !== undefined && message.role !== "assistant") return null; + const usage = object(message.tokens); + const cache = object(usage.cache); + const modelReference = object(message.model); + const model = text(modelReference.id) || text(modelReference.modelID) || text(message.modelID); + const timestampMs = object(message.time).created ?? fallback.timestampMs; + if (!model || typeof timestampMs !== "number" || !Number.isFinite(timestampMs)) return null; + const reasoningTokens = tokens(usage.reasoning); + const totals = { + uncachedInputTokens: tokens(usage.input), + cachedInputTokens: tokens(cache.read), + cacheCreationTokens: tokens(cache.write), + outputTokens: tokens(usage.output) + reasoningTokens, + reasoningTokens, + }; + if (totalTokens(totals) === 0) return null; + const id = fallback.id || text(message.id); + const cost = message.cost; + return { + provider: "opencode", + timestampMs, + model, + sessionId: fallback.sessionId || text(message.sessionID), + totals, + // OpenCode writes zero for models without a known rate, including paid + // subscription models. Let the shared price table estimate those records. + reportedCostUsd: typeof cost === "number" && Number.isFinite(cost) && cost > 0 ? cost : null, + speed: "standard", + dedupeKey: id ? `opencode:${id}` : null, + }; +} + +export interface OpenCodeUsageReadResult { + readonly files: readonly { readonly path: string; readonly records: readonly UsageRecord[] }[]; + readonly missing: boolean; + readonly error: boolean; +} + +/** Reads current SQLite and pre-migration JSON stores without modifying either. */ +export async function readOpenCodeUsage( + root: string, + sinceMs: number, +): Promise { + const files: { path: string; records: UsageRecord[] }[] = []; + const seen = new Set(); + let found = false; + let error = false; + const append = (records: UsageRecord[], record: UsageRecord | null) => { + if (record === null || record.timestampMs < sinceMs) return; + if (record.dedupeKey !== null) { + if (seen.has(record.dedupeKey)) return; + seen.add(record.dedupeKey); + } + records.push(record); + }; + + let databases: string[] = []; + try { + databases = (await NodeFSP.readdir(root, { withFileTypes: true })) + .filter((entry) => entry.isFile() && /^opencode(?:-[a-zA-Z0-9_-]+)?\.db$/.test(entry.name)) + .map((entry) => entry.name) + .sort((a, b) => (a === "opencode.db" ? -1 : b === "opencode.db" ? 1 : a.localeCompare(b))); + } catch (cause) { + if (object(cause).code !== "ENOENT") error = true; + } + for (const name of databases) { + found = true; + const file = { path: NodePath.join(root, name), records: [] as UsageRecord[] }; + files.push(file); + let database: NodeSqlite.DatabaseSync | undefined; + try { + database = new NodeSqlite.DatabaseSync(NodePath.join(root, name), { readOnly: true }); + // A busy live provider should fail this source promptly rather than + // stalling the server while SQLite waits for its writer. + database.exec("PRAGMA busy_timeout = 100"); + const tables = new Set( + database + .prepare("SELECT name FROM sqlite_master WHERE type = 'table'") + .all() + .map((row) => row.name), + ); + if (!tables.has("message") && !tables.has("session_message")) error = true; + for (const table of ["message", "session_message"] as const) { + if (!tables.has(table)) continue; + const columns = new Set( + database + .prepare(`PRAGMA table_info(${table})`) + .all() + .map((row) => row.name), + ); + const timestamp = columns.has("time_created") ? "time_created" : "NULL"; + const predicates = table === "session_message" ? ["type = 'assistant'"] : []; + if (timestamp !== "NULL") predicates.push("time_created >= ?"); + const where = predicates.length > 0 ? ` WHERE ${predicates.join(" AND ")}` : ""; + const statement = database.prepare( + `SELECT id, session_id, data, ${timestamp} AS created FROM ${table}${where}`, + ); + let count = 0; + for (const row of statement.iterate(...(timestamp === "NULL" ? [] : [sinceMs]))) { + append( + file.records, + parseOpenCodeMessage(text(row.data), { + id: text(row.id), + sessionId: text(row.session_id), + ...(typeof row.created === "number" ? { timestampMs: row.created } : {}), + }), + ); + if (++count % 256 === 0) await NodeTimersPromises.setImmediate(); + } + } + } catch { + error = true; + } finally { + database?.close(); + } + } + + // Do not follow symlinks, including cycles. Database records win over their + // old JSON copies when OpenCode has migrated a store in place. + const directories = [NodePath.join(root, "storage", "message")]; + while (directories.length > 0) { + const directory = directories.pop()!; + try { + for (const entry of await NodeFSP.readdir(directory, { withFileTypes: true })) { + const path = NodePath.join(directory, entry.name); + if (entry.isDirectory()) { + directories.push(path); + } else if (entry.isFile() && entry.name.endsWith(".json")) { + found = true; + const id = entry.name.slice(0, -5); + if (seen.has(`opencode:${id}`)) continue; + const file = { path, records: [] as UsageRecord[] }; + files.push(file); + try { + append( + file.records, + parseOpenCodeMessage(await NodeFSP.readFile(path, "utf8"), { id }), + ); + } catch (cause) { + if (object(cause).code !== "ENOENT") error = true; + } + } + } + } catch (cause) { + if (object(cause).code !== "ENOENT") error = true; + } + } + return { files, missing: !found && !error, error }; +} + +/** + * The data directories to read: `OPENCODE_DATA_DIR` (comma-separated) or the + * XDG default, canonicalized so aliases count once. + */ +const resolveOpenCodeDataDirs = Effect.fn("resolveOpenCodeDataDirs")(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const environment = yield* HostProcessEnvironment; + const roots = environment["OPENCODE_DATA_DIR"] + ?.split(",") + .map((value) => value.trim()) + .filter(Boolean); + const dataHome = environment["XDG_DATA_HOME"]?.trim(); + const defaults = [ + path.join( + dataHome && path.isAbsolute(dataHome) + ? dataHome + : path.join(NodeOS.homedir(), ".local", "share"), + "opencode", + ), + ]; + const canonical = new Set(); + for (const root of roots?.length ? roots : defaults) { + const resolved = path.resolve(expandHomePath(root)); + canonical.add(yield* fileSystem.realPath(resolved).pipe(Effect.orElseSucceed(() => resolved))); + } + return [...canonical]; +}); + +export type OpenCodeUsageReaderEnv = FileSystem.FileSystem | Path.Path; + +export const openCodeUsageReader: ProviderUsageReader = { + kind: "scan", + provider: "opencode", + scan: Effect.fn("openCodeUsageReader.scan")(function* ({ windowStartMs }) { + const roots = yield* resolveOpenCodeDataDirs(); + return yield* Effect.forEach( + roots, + (dir) => + Effect.promise(() => readOpenCodeUsage(dir, windowStartMs)).pipe( + Effect.map((result) => ({ + dir, + files: result.missing && !result.error ? null : result.files, + status: result.error ? ("partial" as const) : ("ok" as const), + ...(result.error ? { message: "Some OpenCode history could not be read." } : {}), + })), + ), + { concurrency: "unbounded" }, + ); + }), +};