diff --git a/apps/server/src/provider/Drivers/AntigravityUsage.ts b/apps/server/src/provider/Drivers/AntigravityUsage.ts index d1958788ce2dca7f72009956219908ffd6c1a16e..7b629c857c9aebb0a95993ae0499aa88a94a72ac 100644 --- a/apps/server/src/provider/Drivers/AntigravityUsage.ts +++ b/apps/server/src/provider/Drivers/AntigravityUsage.ts @@ -106,8 +106,9 @@ const make = Effect.gen(function* () { : resolvedRoot; conversationDirs.add(yield* fileSystem.realPath(dir).pipe(Effect.orElseSucceed(() => dir))); } - const result = yield* Effect.promise(() => - readAntigravityUsage([...conversationDirs], windowStartMs, cache), + const result = yield* readAntigravityUsage([...conversationDirs], windowStartMs, cache).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), ); const scanned: ProviderUsageScan[] = []; for (const dir of conversationDirs) { diff --git a/apps/server/src/provider/Drivers/antigravityUsageReader.test.ts b/apps/server/src/provider/Drivers/antigravityUsageReader.test.ts index edafd4ee89a6945371efb784a2c198464e747ca5..20c815fc9b3250be0a51b36d8aac4dc6dc515741 100644 --- a/apps/server/src/provider/Drivers/antigravityUsageReader.test.ts +++ b/apps/server/src/provider/Drivers/antigravityUsageReader.test.ts @@ -1,15 +1,43 @@ -// @effect-diagnostics nodeBuiltinImport:off - the reader under test opens real -// SQLite databases, mirroring the reader's own deliberate node:sqlite usage. +// @effect-diagnostics-next-line nodeBuiltinImport:off - Effect's File.Info has no ctime or sub-millisecond mtime, which the cache tests compare. 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 * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as SqlClient from "effect/sql/SqlClient"; import { makeAntigravityUsageCache, readAntigravityUsage } from "./antigravityUsageReader.ts"; -let dir: string; +/** A writable connection that stays open until the enclosing scope closes. */ +const openDatabase = Effect.fn("openDatabase")(function* (filename: string) { + const context = yield* Layer.build(NodeSqliteClient.layer({ filename })); + const sql = Context.get(context, SqlClient.SqlClient); + return { + exec: (statement: string) => sql.unsafe(statement), + run: (statement: string, params: ReadonlyArray) => sql.unsafe(statement, params), + }; +}); + +/** Seeds `filename` and closes it, as a provider that has finished writing. */ +const seedDatabase = ( + filename: string, + seed: (db: Effect.Success>) => Effect.Effect, +) => + Effect.scoped( + Effect.gen(function* () { + yield* seed(yield* openDatabase(filename)); + }), + ); + +const tempDir = Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + return yield* fileSystem.makeTempDirectoryScoped({ prefix: "usage-reader-test-" }); +}); function protoNumber(field: number, value: number): number[] { const varint = (number: number) => { @@ -34,224 +62,241 @@ function protoText(field: number, value: string): number[] { return protoBytes(field, [...Buffer.from(value)]); } -beforeEach(async () => { - dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "usage-reader-test-")); -}); +const TABLES = + "CREATE TABLE gen_metadata (idx INTEGER, data BLOB); CREATE TABLE steps (idx INTEGER, metadata BLOB)"; -afterEach(async () => { - await NodeFSP.rm(dir, { recursive: true, force: true }); -}); +const createTables = (db: Effect.Success>) => + Effect.forEach(TABLES.split("; "), db.exec, { discard: true }); + +const records = (result: Effect.Success>) => + result.files.flatMap((file) => file.records); describe("readAntigravityUsage", () => { - 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); - const usage = [ - ...protoNumber(2, 100), - ...protoNumber(3, 40), - ...protoNumber(4, 5), - ...protoNumber(5, 20), - ...protoNumber(9, 10), - ...protoText(11, "response-1"), - ]; - const retry = [ - ...protoNumber(1, 1026), - ...protoNumber(2, 12), - ...protoNumber(3, 3), - ...protoText(11, "retry-1"), - ]; - const generation = protoBytes(1, [ - ...protoBytes(4, usage), - ...protoText(19, "Gemini 3 Pro"), - ...protoBytes(9, protoBytes(4, stamp)), - ]); - const step = [ - ...protoBytes(9, usage), - ...protoBytes(8, stamp), - ...protoBytes(28, protoBytes(2, retry)), - ]; - try { - db.exec( - "CREATE TABLE gen_metadata (idx INTEGER, data BLOB); CREATE TABLE steps (idx INTEGER, metadata BLOB)", - ); - db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run(0, new Uint8Array(generation)); - db.prepare("INSERT INTO steps VALUES (?, ?)").run(0, new Uint8Array(step)); - } finally { - db.close(); - } - const result = await readAntigravityUsage(dir, 0); - assert.deepStrictEqual(result.errors, []); - const records = result.files.flatMap((file) => file.records); - assert.strictEqual(records.length, 2); - const main = records.find((record) => record.model === "gemini-3-pro"); - assert.isDefined(main); - assert.strictEqual(main?.timestampMs, 1780000000000); - assert.strictEqual(main?.sessionId, "session-1"); - assert.deepStrictEqual(main?.totals, { - uncachedInputTokens: 100, - cachedInputTokens: 20, - cacheCreationTokens: 5, - outputTokens: 40, - reasoningTokens: 10, - }); - assert.strictEqual( - records.find((record) => record.model === "claude-opus-4-6")?.totals.uncachedInputTokens, - 12, - ); - assert.deepStrictEqual( - (await readAntigravityUsage(dir, 1780000000001)).files.flatMap((file) => file.records), - [], - ); - }); - - it("uses the matching Antigravity generation model for each model-less step", async () => { - const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "model-switch.db")); - try { - db.exec( - "CREATE TABLE gen_metadata (idx INTEGER, data BLOB); CREATE TABLE steps (idx INTEGER, metadata BLOB)", - ); - const generation = db.prepare("INSERT INTO gen_metadata VALUES (?, ?)"); - const step = db.prepare("INSERT INTO steps VALUES (?, ?)"); - for (const [idx, name] of ["Gemini 3 Pro", "Claude Opus 4.6"].entries()) { - generation.run(idx, new Uint8Array(protoBytes(1, protoText(19, name)))); - step.run(idx, new Uint8Array(protoBytes(9, protoNumber(2, 10 + idx)))); - } - } finally { - db.close(); - } - const result = await readAntigravityUsage(dir, 0); - assert.deepStrictEqual(result.errors, []); - assert.deepStrictEqual( - result.files.flatMap((file) => file.records).map((record) => record.model), - ["gemini-3-pro", "claude-opus-4-6"], - ); - }); + it.effect( + "deduplicates Antigravity generation and step usage while preserving retry model and token buckets", + () => + Effect.gen(function* () { + const path = yield* Path.Path; + const dir = yield* tempDir; + const stamp = protoNumber(1, 1780000000); + const usage = [ + ...protoNumber(2, 100), + ...protoNumber(3, 40), + ...protoNumber(4, 5), + ...protoNumber(5, 20), + ...protoNumber(9, 10), + ...protoText(11, "response-1"), + ]; + const retry = [ + ...protoNumber(1, 1026), + ...protoNumber(2, 12), + ...protoNumber(3, 3), + ...protoText(11, "retry-1"), + ]; + const generation = protoBytes(1, [ + ...protoBytes(4, usage), + ...protoText(19, "Gemini 3 Pro"), + ...protoBytes(9, protoBytes(4, stamp)), + ]); + const step = [ + ...protoBytes(9, usage), + ...protoBytes(8, stamp), + ...protoBytes(28, protoBytes(2, retry)), + ]; + yield* seedDatabase(path.join(dir, "session-1.db"), (db) => + Effect.gen(function* () { + yield* createTables(db); + yield* db.run("INSERT INTO gen_metadata VALUES (?, ?)", [ + 0, + new Uint8Array(generation), + ]); + yield* db.run("INSERT INTO steps VALUES (?, ?)", [0, new Uint8Array(step)]); + }), + ); + const result = yield* readAntigravityUsage(dir, 0); + assert.deepStrictEqual(result.errors, []); + const read = records(result); + assert.strictEqual(read.length, 2); + const main = read.find((record) => record.model === "gemini-3-pro"); + assert.isDefined(main); + assert.strictEqual(main?.timestampMs, 1780000000000); + assert.strictEqual(main?.sessionId, "session-1"); + assert.deepStrictEqual(main?.totals, { + uncachedInputTokens: 100, + cachedInputTokens: 20, + cacheCreationTokens: 5, + outputTokens: 40, + reasoningTokens: 10, + }); + assert.strictEqual( + read.find((record) => record.model === "claude-opus-4-6")?.totals.uncachedInputTokens, + 12, + ); + assert.deepStrictEqual(records(yield* readAntigravityUsage(dir, 1780000000001)), []); + }).pipe(Effect.provide(NodeServices.layer)), + ); - it("merges Antigravity aliases that bridge previously separate step records", async () => { - const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "bridge.db")); - try { - db.exec( - "CREATE TABLE gen_metadata (idx INTEGER, data BLOB); CREATE TABLE steps (idx INTEGER, metadata BLOB)", + it.effect("uses the matching Antigravity generation model for each model-less step", () => + Effect.gen(function* () { + const path = yield* Path.Path; + const dir = yield* tempDir; + yield* seedDatabase(path.join(dir, "model-switch.db"), (db) => + Effect.gen(function* () { + yield* createTables(db); + for (const [idx, name] of ["Gemini 3 Pro", "Claude Opus 4.6"].entries()) { + yield* db.run("INSERT INTO gen_metadata VALUES (?, ?)", [ + idx, + new Uint8Array(protoBytes(1, protoText(19, name))), + ]); + yield* db.run("INSERT INTO steps VALUES (?, ?)", [ + idx, + new Uint8Array(protoBytes(9, protoNumber(2, 10 + idx))), + ]); + } + }), ); - const step = db.prepare("INSERT INTO steps VALUES (?, ?)"); - step.run( - 0, - new Uint8Array(protoBytes(9, [...protoNumber(2, 100), ...protoText(11, "response")])), + const result = yield* readAntigravityUsage(dir, 0); + assert.deepStrictEqual(result.errors, []); + assert.deepStrictEqual( + records(result).map((record) => record.model), + ["gemini-3-pro", "claude-opus-4-6"], ); - step.run( - 1, - new Uint8Array(protoBytes(9, [...protoNumber(3, 40), ...protoText(12, "provider")])), - ); - db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run( - 0, - new Uint8Array( - protoBytes(1, [ - ...protoText(19, "Gemini 3 Pro"), - ...protoBytes(4, [ - ...protoNumber(2, 50), - ...protoNumber(5, 20), - ...protoText(11, "response"), - ...protoText(12, "provider"), - ]), - ]), - ), - ); - } finally { - db.close(); - } - const result = await readAntigravityUsage(dir, 0); - assert.deepStrictEqual(result.errors, []); - const records = result.files.flatMap((file) => file.records); - assert.strictEqual(records.length, 1); - assert.deepStrictEqual(records[0]?.totals, { - uncachedInputTokens: 100, - cachedInputTokens: 20, - cacheCreationTokens: 0, - outputTokens: 40, - reasoningTokens: 0, - }); - }); + }).pipe(Effect.provide(NodeServices.layer)), + ); - it("merges Antigravity provider and message aliases across configured roots while keeping original ownership", async () => { - const roots = [NodePath.join(dir, "first"), NodePath.join(dir, "second")]; - for (const [index, root] of roots.entries()) { - await NodeFSP.mkdir(root); - const db = new NodeSqlite.DatabaseSync(NodePath.join(root, `session-${index}.db`)); - try { - db.exec("CREATE TABLE steps (idx INTEGER, metadata BLOB)"); - for (const identity of [7, 12]) { - const usage = [ - ...protoNumber(1, 246), - ...protoNumber(2, index === 0 ? 100 : 150), - ...protoText(11, `response-${index}-${identity}`), - ...protoText(identity, `shared-${identity}`), - ]; - db.prepare("INSERT INTO steps VALUES (?, ?)").run( - identity, - new Uint8Array(protoBytes(9, usage)), - ); - } - } finally { - db.close(); - } - } - const result = await readAntigravityUsage(roots, 0); - assert.deepStrictEqual(result.errors, []); - assert.strictEqual(result.files.length, 2); - assert.strictEqual(result.files[0]?.root, roots[0]); - assert.strictEqual(result.files[0]?.records.length, 2); - assert.strictEqual(result.files[1]?.records.length, 0); - assert.deepStrictEqual( - result.files[0]?.records.map((record) => record.totals.uncachedInputTokens), - [150, 150], - ); - assert.isTrue(result.files[0]?.records.every((record) => record.sessionId === "session-0")); - }); - - it("upgrades Antigravity fallback timestamps before applying the date window", async () => { - for (const fallback of ["mtime", "trajectory"]) { - const path = NodePath.join(dir, `${fallback}.db`); - const db = new NodeSqlite.DatabaseSync(path); - try { - db.exec( - "CREATE TABLE gen_metadata (idx INTEGER, data BLOB); CREATE TABLE steps (idx INTEGER, metadata BLOB)", - ); - if (fallback === "trajectory") { - db.exec("CREATE TABLE trajectory_metadata_blob (data BLOB)"); - db.prepare("INSERT INTO trajectory_metadata_blob VALUES (?)").run( - new Uint8Array(protoBytes(2, protoNumber(1, 1780000200))), - ); - } - for (const [index, seconds] of [1780000000, 1780000200].entries()) { - const usage = [...protoNumber(2, 10), ...protoText(11, `${fallback}-${index}`)]; - db.prepare("INSERT INTO steps VALUES (?, ?)").run( - index, - new Uint8Array(protoBytes(9, usage)), - ); - db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run( - index, + it.effect("merges Antigravity aliases that bridge previously separate step records", () => + Effect.gen(function* () { + const path = yield* Path.Path; + const dir = yield* tempDir; + yield* seedDatabase(path.join(dir, "bridge.db"), (db) => + Effect.gen(function* () { + yield* createTables(db); + yield* db.run("INSERT INTO steps VALUES (?, ?)", [ + 0, + new Uint8Array(protoBytes(9, [...protoNumber(2, 100), ...protoText(11, "response")])), + ]); + yield* db.run("INSERT INTO steps VALUES (?, ?)", [ + 1, + new Uint8Array(protoBytes(9, [...protoNumber(3, 40), ...protoText(12, "provider")])), + ]); + yield* db.run("INSERT INTO gen_metadata VALUES (?, ?)", [ + 0, new Uint8Array( protoBytes(1, [ - ...protoBytes(4, usage), - ...protoBytes(9, protoBytes(4, protoNumber(1, seconds))), + ...protoText(19, "Gemini 3 Pro"), + ...protoBytes(4, [ + ...protoNumber(2, 50), + ...protoNumber(5, 20), + ...protoText(11, "response"), + ...protoText(12, "provider"), + ]), ]), ), + ]); + }), + ); + const result = yield* readAntigravityUsage(dir, 0); + assert.deepStrictEqual(result.errors, []); + const read = records(result); + assert.strictEqual(read.length, 1); + assert.deepStrictEqual(read[0]?.totals, { + uncachedInputTokens: 100, + cachedInputTokens: 20, + cacheCreationTokens: 0, + outputTokens: 40, + reasoningTokens: 0, + }); + }).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect( + "merges Antigravity provider and message aliases across configured roots while keeping original ownership", + () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const dir = yield* tempDir; + const roots = [path.join(dir, "first"), path.join(dir, "second")]; + for (const [index, root] of roots.entries()) { + yield* fileSystem.makeDirectory(root); + yield* seedDatabase(path.join(root, `session-${index}.db`), (db) => + Effect.gen(function* () { + yield* db.exec("CREATE TABLE steps (idx INTEGER, metadata BLOB)"); + for (const identity of [7, 12]) { + const usage = [ + ...protoNumber(1, 246), + ...protoNumber(2, index === 0 ? 100 : 150), + ...protoText(11, `response-${index}-${identity}`), + ...protoText(identity, `shared-${identity}`), + ]; + yield* db.run("INSERT INTO steps VALUES (?, ?)", [ + identity, + new Uint8Array(protoBytes(9, usage)), + ]); + } + }), ); } - } finally { - db.close(); + const result = yield* readAntigravityUsage(roots, 0); + assert.deepStrictEqual(result.errors, []); + assert.strictEqual(result.files.length, 2); + assert.strictEqual(result.files[0]?.root, roots[0]); + assert.strictEqual(result.files[0]?.records.length, 2); + assert.strictEqual(result.files[1]?.records.length, 0); + assert.deepStrictEqual( + result.files[0]?.records.map((record) => record.totals.uncachedInputTokens), + [150, 150], + ); + assert.isTrue(result.files[0]?.records.every((record) => record.sessionId === "session-0")); + }).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("upgrades Antigravity fallback timestamps before applying the date window", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const dir = yield* tempDir; + for (const fallback of ["mtime", "trajectory"]) { + const file = path.join(dir, `${fallback}.db`); + yield* seedDatabase(file, (db) => + Effect.gen(function* () { + yield* createTables(db); + if (fallback === "trajectory") { + yield* db.exec("CREATE TABLE trajectory_metadata_blob (data BLOB)"); + yield* db.run("INSERT INTO trajectory_metadata_blob VALUES (?)", [ + new Uint8Array(protoBytes(2, protoNumber(1, 1780000200))), + ]); + } + for (const [index, seconds] of [1780000000, 1780000200].entries()) { + const usage = [...protoNumber(2, 10), ...protoText(11, `${fallback}-${index}`)]; + yield* db.run("INSERT INTO steps VALUES (?, ?)", [ + index, + new Uint8Array(protoBytes(9, usage)), + ]); + yield* db.run("INSERT INTO gen_metadata VALUES (?, ?)", [ + index, + new Uint8Array( + protoBytes(1, [ + ...protoBytes(4, usage), + ...protoBytes(9, protoBytes(4, protoNumber(1, seconds))), + ]), + ), + ]); + } + }), + ); + yield* fileSystem.utimes(file, 1780000000, 1780000000); } - await NodeFSP.utimes(path, 1780000000, 1780000000); - } - const result = await readAntigravityUsage(dir, 1780000100000); - assert.deepStrictEqual(result.errors, []); - const records = result.files.flatMap((file) => file.records); - assert.strictEqual(records.length, 2); - assert.deepStrictEqual( - records.map((record) => record.timestampMs), - [1780000200000, 1780000200000], - ); - }); + const result = yield* readAntigravityUsage(dir, 1780000100000); + assert.deepStrictEqual(result.errors, []); + const read = records(result); + assert.strictEqual(read.length, 2); + assert.deepStrictEqual( + read.map((record) => record.timestampMs), + [1780000200000, 1780000200000], + ); + }).pipe(Effect.provide(NodeServices.layer)), + ); const antigravityGeneration = (responseId: string) => new Uint8Array( @@ -266,126 +311,138 @@ describe("readAntigravityUsage", () => { ]), ); - it("reuses an unchanged Antigravity database instead of decoding it again", async () => { - const path = NodePath.join(dir, "session-1.db"); - const db = new NodeSqlite.DatabaseSync(path); - try { - db.exec("CREATE TABLE gen_metadata (idx INTEGER, data BLOB)"); - db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run(0, antigravityGeneration("r-1")); + it.effect("reuses an unchanged Antigravity database instead of decoding it again", () => + Effect.gen(function* () { + const path = yield* Path.Path; + const dir = yield* tempDir; + const db = yield* openDatabase(path.join(dir, "session-1.db")); + yield* db.exec("CREATE TABLE gen_metadata (idx INTEGER, data BLOB)"); + yield* db.run("INSERT INTO gen_metadata VALUES (?, ?)", [0, antigravityGeneration("r-1")]); const cache = makeAntigravityUsageCache(); - assert.deepStrictEqual((await readAntigravityUsage(dir, 0, cache)).errors, []); + assert.deepStrictEqual((yield* readAntigravityUsage(dir, 0, cache)).errors, []); // An exclusive lock makes a fresh read fail without touching the file, so // only a cache hit can still return the earlier records. - db.exec("BEGIN EXCLUSIVE"); - assert.strictEqual((await readAntigravityUsage(dir, 0)).errors.length, 1); - const cached = await readAntigravityUsage(dir, 0, cache); + yield* db.exec("BEGIN EXCLUSIVE"); + assert.strictEqual((yield* readAntigravityUsage(dir, 0)).errors.length, 1); + const cached = yield* readAntigravityUsage(dir, 0, cache); assert.deepStrictEqual(cached.errors, []); assert.deepStrictEqual( - cached.files.flatMap((file) => file.records).map((record) => record.dedupeKey), + records(cached).map((record) => record.dedupeKey), ["antigravity:11:r-1"], ); - db.exec("ROLLBACK"); - } finally { - db.close(); - } - }); + yield* db.exec("ROLLBACK"); + }).pipe(Effect.provide(NodeServices.layer)), + ); - it("rereads an Antigravity database rewritten with its size and mtime restored", async () => { - const path = NodePath.join(dir, "session-1.db"); - const db = new NodeSqlite.DatabaseSync(path); - try { - db.exec("CREATE TABLE gen_metadata (idx INTEGER, data BLOB)"); - db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run(0, antigravityGeneration("r-1")); - } finally { - db.close(); - } - await NodeFSP.utimes(path, 1780000000, 1780000000); - const cache = makeAntigravityUsageCache(); - assert.deepStrictEqual((await readAntigravityUsage(dir, 0, cache)).errors, []); - - // ctime has the kernel's timestamp granularity, which can be a few - // milliseconds, so repeat the forged rewrite until it lands on a later tick - // than the cached read, as any real rewrite does. - const cached = await NodeFSP.stat(path); - do { - await NodeFSP.writeFile(path, Buffer.alloc(cached.size)); - await NodeFSP.utimes(path, 1780000000, 1780000000); - } while ((await NodeFSP.stat(path)).ctimeMs === cached.ctimeMs); - const restored = await NodeFSP.stat(path); - assert.strictEqual(restored.size, cached.size); - assert.strictEqual(restored.mtimeMs, cached.mtimeMs); - const next = await readAntigravityUsage(dir, 0, cache); - assert.deepStrictEqual(next.errors, [path]); - assert.deepStrictEqual( - next.files.flatMap((file) => file.records), - [], - ); - }); - - it("rereads an Antigravity database when only its WAL changed", async () => { - const path = NodePath.join(dir, "session-1.db"); - const db = new NodeSqlite.DatabaseSync(path); - try { - db.exec( - "PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0; CREATE TABLE gen_metadata (idx INTEGER, data BLOB)", + it.effect("rereads an Antigravity database rewritten with its size and mtime restored", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const pathService = yield* Path.Path; + const dir = yield* tempDir; + const path = pathService.join(dir, "session-1.db"); + yield* seedDatabase(path, (db) => + Effect.gen(function* () { + yield* db.exec("CREATE TABLE gen_metadata (idx INTEGER, data BLOB)"); + yield* db.run("INSERT INTO gen_metadata VALUES (?, ?)", [ + 0, + antigravityGeneration("r-1"), + ]); + }), ); - const insert = db.prepare("INSERT INTO gen_metadata VALUES (?, ?)"); - insert.run(0, antigravityGeneration("r-1")); + yield* fileSystem.utimes(path, 1780000000, 1780000000); const cache = makeAntigravityUsageCache(); - const first = await readAntigravityUsage(dir, 0, cache); - assert.strictEqual(first.files.flatMap((file) => file.records).length, 1); + assert.deepStrictEqual((yield* readAntigravityUsage(dir, 0, cache)).errors, []); - const before = await NodeFSP.stat(path); - insert.run(1, antigravityGeneration("r-2")); - const after = await NodeFSP.stat(path); + // ctime has the kernel's timestamp granularity, which can be a few + // milliseconds, so repeat the forged rewrite until it lands on a later tick + // than the cached read, as any real rewrite does. + const stat = () => Effect.promise(() => NodeFSP.stat(path)); + const cached = yield* stat(); + do { + yield* fileSystem.writeFile(path, new Uint8Array(cached.size)); + yield* fileSystem.utimes(path, 1780000000, 1780000000); + } while ((yield* stat()).ctimeMs === cached.ctimeMs); + const restored = yield* stat(); + assert.strictEqual(restored.size, cached.size); + assert.strictEqual(restored.mtimeMs, cached.mtimeMs); + const next = yield* readAntigravityUsage(dir, 0, cache); + assert.deepStrictEqual(next.errors, [path]); + assert.deepStrictEqual(records(next), []); + }).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("rereads an Antigravity database when only its WAL changed", () => + Effect.gen(function* () { + const pathService = yield* Path.Path; + const dir = yield* tempDir; + const path = pathService.join(dir, "session-1.db"); + const db = yield* openDatabase(path); + yield* db.exec("PRAGMA journal_mode = WAL"); + yield* db.exec("PRAGMA wal_autocheckpoint = 0"); + yield* db.exec("CREATE TABLE gen_metadata (idx INTEGER, data BLOB)"); + const insert = (idx: number, responseId: string) => + db.run("INSERT INTO gen_metadata VALUES (?, ?)", [idx, antigravityGeneration(responseId)]); + yield* insert(0, "r-1"); + const cache = makeAntigravityUsageCache(); + const first = yield* readAntigravityUsage(dir, 0, cache); + assert.strictEqual(records(first).length, 1); + + const before = yield* Effect.promise(() => NodeFSP.stat(path)); + yield* insert(1, "r-2"); + const after = yield* Effect.promise(() => NodeFSP.stat(path)); assert.strictEqual(after.size, before.size); assert.strictEqual(after.mtimeMs, before.mtimeMs); - const next = await readAntigravityUsage(dir, 0, cache); + const next = yield* readAntigravityUsage(dir, 0, cache); assert.deepStrictEqual( - next.files.flatMap((file) => file.records).map((record) => record.dedupeKey), + records(next).map((record) => record.dedupeKey), ["antigravity:11:r-1", "antigravity:11:r-2"], ); - } finally { - db.close(); - } - }); + }).pipe(Effect.provide(NodeServices.layer)), + ); - it("reads Antigravity step-only stores and reports malformed databases", async () => { - const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "steps.db")); - try { - db.exec("CREATE TABLE steps (idx INTEGER, metadata BLOB)"); - const usage = [...protoNumber(1, 246), ...protoNumber(2, 10), ...protoNumber(3, 5)]; - db.prepare("INSERT INTO steps VALUES (?, ?)").run( - 0, - new Uint8Array([...protoBytes(9, usage), ...protoBytes(8, protoNumber(1, 1780000000))]), + it.effect("reads Antigravity step-only stores and reports malformed databases", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const dir = yield* tempDir; + yield* seedDatabase(path.join(dir, "steps.db"), (db) => + Effect.gen(function* () { + yield* db.exec("CREATE TABLE steps (idx INTEGER, metadata BLOB)"); + const usage = [...protoNumber(1, 246), ...protoNumber(2, 10), ...protoNumber(3, 5)]; + yield* db.run("INSERT INTO steps VALUES (?, ?)", [ + 0, + new Uint8Array([...protoBytes(9, usage), ...protoBytes(8, protoNumber(1, 1780000000))]), + ]); + }), ); - } finally { - db.close(); - } - await NodeFSP.writeFile(NodePath.join(dir, "broken.db"), "not a sqlite database"); - const result = await readAntigravityUsage(dir, 0); - assert.strictEqual(result.errors.length, 1); - assert.strictEqual(result.files.flatMap((file) => file.records)[0]?.model, "gemini-2.5-pro"); - assert.strictEqual(result.files.flatMap((file) => file.records)[0]?.totals.outputTokens, 5); - }); + yield* fileSystem.writeFileString(path.join(dir, "broken.db"), "not a sqlite database"); + const result = yield* readAntigravityUsage(dir, 0); + assert.strictEqual(result.errors.length, 1); + assert.strictEqual(records(result)[0]?.model, "gemini-2.5-pro"); + assert.strictEqual(records(result)[0]?.totals.outputTokens, 5); + }).pipe(Effect.provide(NodeServices.layer)), + ); - it("ignores large values in unused Antigravity protobuf fields", async () => { - const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "large-varint.db")); - try { - db.exec("CREATE TABLE steps (idx INTEGER, metadata BLOB)"); - const unusedField = [...protoNumber(99, 0).slice(0, -1), ...Array(9).fill(0xff), 0x01]; - const usage = [...protoNumber(1, 246), ...protoNumber(2, 10), ...unusedField]; - db.prepare("INSERT INTO steps VALUES (?, ?)").run(0, new Uint8Array(protoBytes(9, usage))); - } finally { - db.close(); - } - const result = await readAntigravityUsage(dir, 0); - assert.deepStrictEqual(result.errors, []); - assert.strictEqual( - result.files.flatMap((file) => file.records)[0]?.totals.uncachedInputTokens, - 10, - ); - }); + it.effect("ignores large values in unused Antigravity protobuf fields", () => + Effect.gen(function* () { + const path = yield* Path.Path; + const dir = yield* tempDir; + yield* seedDatabase(path.join(dir, "large-varint.db"), (db) => + Effect.gen(function* () { + yield* db.exec("CREATE TABLE steps (idx INTEGER, metadata BLOB)"); + const unusedField = [...protoNumber(99, 0).slice(0, -1), ...Array(9).fill(0xff), 0x01]; + const usage = [...protoNumber(1, 246), ...protoNumber(2, 10), ...unusedField]; + yield* db.run("INSERT INTO steps VALUES (?, ?)", [ + 0, + new Uint8Array(protoBytes(9, usage)), + ]); + }), + ); + const result = yield* readAntigravityUsage(dir, 0); + assert.deepStrictEqual(result.errors, []); + assert.strictEqual(records(result)[0]?.totals.uncachedInputTokens, 10); + }).pipe(Effect.provide(NodeServices.layer)), + ); }); diff --git a/apps/server/src/provider/Drivers/antigravityUsageReader.ts b/apps/server/src/provider/Drivers/antigravityUsageReader.ts index 9e91b6eacbb6fb94d7942818cbfed071a8e2d7cb..e48609cf874d23416ef8f9c03dfe4285f05ba967 100644 --- a/apps/server/src/provider/Drivers/antigravityUsageReader.ts +++ b/apps/server/src/provider/Drivers/antigravityUsageReader.ts @@ -1,11 +1,14 @@ -// node:sqlite reads live conversation databases while Node fs discovers them. -// @effect-diagnostics nodeBuiltinImport:off +// @effect-diagnostics-next-line nodeBuiltinImport:off - Effect's File.Info has no ctime, which the cache fingerprint needs. 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 type { UsageRecord } from "@t3tools/provider-core/server/usage"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/sql/SqlClient"; type FieldValue = number | bigint | Uint8Array; type Fields = Map; @@ -158,103 +161,136 @@ interface UsageCandidate { timestampQuality: number; } -async function readDatabase(path: string, fallbackTimestamp: number): Promise { - const db = new NodeSqlite.DatabaseSync(path, { readOnly: true }); - try { - db.exec("PRAGMA busy_timeout = 100; BEGIN"); +/** A conversation database whose usage tables are missing or undecodable. */ +class AntigravityDatabaseDecodeError extends Schema.TaggedError()( + "AntigravityDatabaseDecodeError", + { path: Schema.String, cause: Schema.optional(Schema.Defect()) }, +) { + override get message(): string { + return `Antigravity usage could not be decoded from ${this.path}.`; + } +} + +const readDatabase = Effect.fn("readDatabase")(function* ( + path: string, + sessionId: string, + fallbackTimestamp: number, +) { + const sql = yield* SqlClient.SqlClient; + const decode = (evaluate: () => A) => + Effect.try({ + try: evaluate, + catch: (cause) => new AntigravityDatabaseDecodeError({ path, cause }), + }); + yield* sql.unsafe("PRAGMA busy_timeout = 100"); + return yield* Effect.gen(function* () { const tables = new Set( - db - .prepare("SELECT name FROM sqlite_master WHERE type = 'table'") - .all() - .map((row) => row.name), + (yield* sql.unsafe<{ readonly name: unknown }>( + "SELECT name FROM sqlite_master WHERE type = 'table'", + )).map((row) => row.name), ); if (!tables.has("gen_metadata") && !tables.has("steps")) { - throw new Error("Missing Antigravity usage tables"); + return yield* new AntigravityDatabaseDecodeError({ path }); } - const readMetadata = async (query: string, column: string, step: boolean) => { + const readMetadata = Effect.fnUntraced(function* ( + query: string, + column: string, + step: boolean, + ) { const entries: Array<{ idx: number; entry: Metadata }> = []; - for (const row of db.prepare(query).iterate()) { - if (typeof row.idx !== "number") throw new Error("Invalid Antigravity metadata index"); - entries.push({ idx: row.idx, entry: metadata(blob(row[column]), step) }); - if (entries.length % 256 === 0) await NodeTimersPromises.setImmediate(); + for (const row of yield* sql.unsafe>(query)) { + const idx = row.idx; + if (typeof idx !== "number") { + return yield* new AntigravityDatabaseDecodeError({ path }); + } + entries.push({ idx, entry: yield* decode(() => metadata(blob(row[column]), step)) }); + if (entries.length % 256 === 0) yield* Effect.yieldNow; } return entries; - }; + }); const generations = tables.has("gen_metadata") - ? await readMetadata("SELECT idx, data FROM gen_metadata ORDER BY idx", "data", false) + ? yield* readMetadata("SELECT idx, data FROM gen_metadata ORDER BY idx", "data", false) : []; let trajectoryTimestamp: number | null = null; if (tables.has("trajectory_metadata_blob")) { - for (const row of db.prepare("SELECT data FROM trajectory_metadata_blob").iterate()) { - trajectoryTimestamp ??= timestamp(nested(fields(blob(row.data)), 2)); + for (const row of yield* sql.unsafe<{ readonly data: unknown }>( + "SELECT data FROM trajectory_metadata_blob", + )) { + trajectoryTimestamp ??= yield* decode(() => timestamp(nested(fields(blob(row.data)), 2))); } } const steps = tables.has("steps") - ? await readMetadata( + ? yield* readMetadata( "SELECT idx, metadata FROM steps WHERE metadata IS NOT NULL ORDER BY idx", "metadata", true, ) : []; - const sessionId = NodePath.basename(path, ".db"); - const records: UsageCandidate[] = []; - const generationModels = new Map(generations.map(({ idx, entry }) => [idx, entry.model])); - for (const [source, entries] of [ - ["step", steps], - ["generation", generations], - ] as const) { - for (const [index, { idx, entry }] of entries.entries()) { - for (const [usageIndex, usage] of entry.usages.entries()) { - const outputTokens = Math.max( - numberAt(usage, 3), - numberAt(usage, 9) + numberAt(usage, 10), - ); - const totals = { - uncachedInputTokens: numberAt(usage, 2), - cachedInputTokens: numberAt(usage, 5), - cacheCreationTokens: numberAt(usage, 4), - outputTokens, - reasoningTokens: Math.min(outputTokens, numberAt(usage, 9)), - }; - if ( - totals.uncachedInputTokens + - totals.cachedInputTokens + - totals.cacheCreationTokens + - outputTokens === - 0 - ) - continue; - const keys = ([11, 12, 7] as const).flatMap((key) => { - const id = textAt(usage, key); - return id ? [`antigravity:${key}:${id}`] : []; - }); - const record: UsageRecord = { - provider: "antigravity", - sessionId, - timestampMs: entry.timestampMs ?? trajectoryTimestamp ?? fallbackTimestamp, - model: - MODEL_IDS[numberAt(usage, 1)] || - entry.model || - (source === "step" ? generationModels.get(idx) : "") || - modelName("", numberAt(usage, 1)) || - "antigravity-unknown", - totals, - reportedCostUsd: null, - speed: "standard", - dedupeKey: keys[0] ?? `antigravity:${sessionId}:${source}:${index}:${usageIndex}`, - }; - records.push({ - record, - keys, - timestampQuality: entry.timestampMs !== null ? 2 : trajectoryTimestamp !== null ? 1 : 0, - }); - } + return yield* decode(() => + usageCandidates(sessionId, generations, steps, trajectoryTimestamp, fallbackTimestamp), + ); + }).pipe(sql.withTransaction); +}); + +function usageCandidates( + sessionId: string, + generations: ReadonlyArray<{ idx: number; entry: Metadata }>, + steps: ReadonlyArray<{ idx: number; entry: Metadata }>, + trajectoryTimestamp: number | null, + fallbackTimestamp: number, +): UsageCandidate[] { + const records: UsageCandidate[] = []; + const generationModels = new Map(generations.map(({ idx, entry }) => [idx, entry.model])); + for (const [source, entries] of [ + ["step", steps], + ["generation", generations], + ] as const) { + for (const [index, { idx, entry }] of entries.entries()) { + for (const [usageIndex, usage] of entry.usages.entries()) { + const outputTokens = Math.max(numberAt(usage, 3), numberAt(usage, 9) + numberAt(usage, 10)); + const totals = { + uncachedInputTokens: numberAt(usage, 2), + cachedInputTokens: numberAt(usage, 5), + cacheCreationTokens: numberAt(usage, 4), + outputTokens, + reasoningTokens: Math.min(outputTokens, numberAt(usage, 9)), + }; + if ( + totals.uncachedInputTokens + + totals.cachedInputTokens + + totals.cacheCreationTokens + + outputTokens === + 0 + ) + continue; + const keys = ([11, 12, 7] as const).flatMap((key) => { + const id = textAt(usage, key); + return id ? [`antigravity:${key}:${id}`] : []; + }); + const record: UsageRecord = { + provider: "antigravity", + sessionId, + timestampMs: entry.timestampMs ?? trajectoryTimestamp ?? fallbackTimestamp, + model: + MODEL_IDS[numberAt(usage, 1)] || + entry.model || + (source === "step" ? generationModels.get(idx) : "") || + modelName("", numberAt(usage, 1)) || + "antigravity-unknown", + totals, + reportedCostUsd: null, + speed: "standard", + dedupeKey: keys[0] ?? `antigravity:${sessionId}:${source}:${index}:${usageIndex}`, + }; + records.push({ + record, + keys, + timestampQuality: entry.timestampMs !== null ? 2 : trajectoryTimestamp !== null ? 1 : 0, + }); } } - return records; - } finally { - db.close(); } + return records; } interface CachedDatabase { @@ -275,11 +311,13 @@ export const makeAntigravityUsageCache = () => new Map() * filtering. With a cache, databases unchanged since the previous read are not * decoded again. */ -export async function readAntigravityUsage( +export const readAntigravityUsage = Effect.fn("readAntigravityUsage")(function* ( conversationsDirectories: string | readonly string[], sinceMs: number, cache?: Map, ) { + const fileSystem = yield* FileSystem.FileSystem; + const pathService = yield* Path.Path; const roots = typeof conversationsDirectories === "string" ? [conversationsDirectories] @@ -350,50 +388,69 @@ export async function readAntigravityUsage( } }; const visited = new Set(); - const walk = async (directory: string, root: string): Promise => { - let entries; - try { - entries = await NodeFSP.readdir(directory, { withFileTypes: true }); - } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "ENOENT") errors.push(directory); - return; - } - entries.sort((a, b) => a.name.localeCompare(b.name)); - for (const entry of entries) { - const path = NodePath.join(directory, entry.name); - if (entry.isDirectory()) { - await walk(path, root); - } else if (entry.isFile() && entry.name.endsWith(".db")) { - try { - const canonical = await NodeFSP.realpath(path); - if (visited.has(canonical)) continue; - visited.add(canonical); - const stat = await NodeFSP.stat(path); - const wal = await NodeFSP.stat(`${path}-wal`).catch(() => null); - const fingerprint = [stat, wal] - .map((file) => (file ? `${file.size}:${file.mtimeMs}:${file.ctimeMs}` : "-")) - .join("/"); - const cached = cache?.get(canonical); - let candidates: readonly UsageCandidate[]; - if (cached?.fingerprint === fingerprint) { - candidates = cached.candidates; - } else { - candidates = await readDatabase(path, stat.mtimeMs); - cache?.set(canonical, { fingerprint, candidates }); - } + const readFile = Effect.fn("readAntigravityUsage.readFile")(function* (path: string) { + const canonical = yield* fileSystem.realPath(path); + if (visited.has(canonical)) return null; + visited.add(canonical); + const fingerprintOf = (file: string) => + Effect.tryPromise(() => NodeFSP.stat(file)).pipe( + Effect.map((stat) => ({ stat, part: `${stat.size}:${stat.mtimeMs}:${stat.ctimeMs}` })), + ); + const { stat, part } = yield* fingerprintOf(path); + const wal = yield* fingerprintOf(`${path}-wal`).pipe(Effect.option); + const fingerprint = `${part}/${wal._tag === "Some" ? wal.value.part : "-"}`; + const cached = cache?.get(canonical); + if (cached?.fingerprint === fingerprint) return cached.candidates; + const candidates = yield* readDatabase( + path, + pathService.basename(path, ".db"), + stat.mtimeMs, + ).pipe(Effect.provide(NodeSqliteClient.layer({ filename: path, readonly: true }))); + cache?.set(canonical, { fingerprint, candidates }); + return candidates; + }); + const walk = (directory: string, root: string): Effect.Effect => + Effect.gen(function* () { + const names = yield* fileSystem.readDirectory(directory).pipe( + Effect.catchTags({ + PlatformError: (cause) => { + if (cause.reason._tag !== "NotFound") errors.push(directory); + return Effect.succeed(null); + }, + }), + ); + if (names === null) return; + names.sort((a, b) => a.localeCompare(b)); + for (const name of names) { + const path = pathService.join(directory, name); + // Like a directory listing's entry types: symlinks are neither files nor directories. + const isLink = Exit.isSuccess(yield* Effect.exit(fileSystem.readLink(path))); + const type = isLink + ? "SymbolicLink" + : yield* fileSystem.stat(path).pipe( + Effect.map((info) => info.type), + Effect.orElseSucceed(() => null), + ); + if (type === "Directory") { + yield* walk(path, root); + } else if ((type === "File" || type === null) && name.endsWith(".db")) { + const candidates = yield* readFile(path).pipe( + Effect.catch(() => { + errors.push(path); + return Effect.succeed(null); + }), + ); + if (candidates === null) continue; const fileIndex = files.length; files.push({ root, path, records: [] }); for (const [index, candidate] of candidates.entries()) { append(candidate, fileIndex); - if (index % 256 === 255) await NodeTimersPromises.setImmediate(); + if (index % 256 === 255) yield* Effect.yieldNow; } - } catch { - errors.push(path); } } - } - }; - for (const root of roots) await walk(root, root); + }); + for (const root of roots) yield* walk(root, root); if (cache !== undefined) { for (const key of cache.keys()) if (!visited.has(key)) cache.delete(key); } @@ -403,4 +460,4 @@ export async function readAntigravityUsage( } } return { files, errors }; -} +}); diff --git a/packages/provider-cursor/src/server/CursorAccountReader.ts b/packages/provider-cursor/src/server/CursorAccountReader.ts index a6922ef87a53713a3c7b3148cf29dec7e23b46ab..886bfe2f87371627026ab620ec2ee6b89ea29323 100644 --- a/packages/provider-cursor/src/server/CursorAccountReader.ts +++ b/packages/provider-cursor/src/server/CursorAccountReader.ts @@ -5,9 +5,11 @@ * @module provider-cursor/server/CursorAccountReader */ import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as HttpClient from "effect/http/HttpClient"; import * as Layer from "effect/Layer"; -import * as Result from "effect/Result"; import type { CursorCredentialSource } from "./accountCache.ts"; import { readCursorAccountUsage, type CursorAccountUsageReadResult } from "./accountUsage.ts"; @@ -29,23 +31,16 @@ export const layer = Layer.effect( CursorAccountReader, Effect.gen(function* () { const keychain = yield* CursorKeychain.CursorKeychain; + const context = yield* Effect.context< + FileSystem.FileSystem | Crypto.Crypto | HttpClient.HttpClient + >(); return CursorAccountReader.of({ + // Only a Keychain login asks the Keychain. The dashboard reader turns + // its outcome, a failure included, into the source's message. read: (credentialSource, sinceMs, untilMs) => - Effect.gen(function* () { - // Only a Keychain login asks the Keychain. The dashboard reader - // turns its outcome, a failure included, into the source's message. - const token = - typeof credentialSource === "string" - ? Result.succeed(null) - : yield* Effect.result(keychain.accessToken); - return yield* Effect.promise(() => - readCursorAccountUsage(credentialSource, sinceMs, untilMs, globalThis.fetch, () => - Result.isSuccess(token) - ? Promise.resolve(token.success) - : Promise.reject(token.failure), - ), - ); - }), + readCursorAccountUsage(credentialSource, sinceMs, untilMs, keychain.accessToken).pipe( + Effect.provideContext(context), + ), }); }), ); diff --git a/packages/provider-cursor/src/server/accountUsage.test.ts b/packages/provider-cursor/src/server/accountUsage.test.ts index 32a0f41c0612489c99f677910f0cbc1fbe374ba9..3924576b43dcae05d33bad8320edaee7a0920b49 100644 --- a/packages/provider-cursor/src/server/accountUsage.test.ts +++ b/packages/provider-cursor/src/server/accountUsage.test.ts @@ -1,340 +1,411 @@ -// @effect-diagnostics nodeBuiltinImport:off - the reader under test reads a -// real CLI credential file from disk. -import * as NodeFSP from "node:fs/promises"; -import * as NodeOS from "node:os"; -import * as NodePath from "node:path"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as FetchHttpClient from "effect/http/FetchHttpClient"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientError from "effect/http/HttpClientError"; +import * as HttpClientRequest from "effect/http/HttpClientRequest"; +import * as HttpClientResponse from "effect/http/HttpClientResponse"; +import * as Path from "effect/Path"; -import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest"; - -import { CursorKeychainTimeoutError } from "./CursorKeychain.ts"; +import * as CursorKeychain from "./CursorKeychain.ts"; import { readCursorAccountUsage } from "./accountUsage.ts"; -let dir: string; +/** A scoped temporary directory holding `auth.json` with `accessToken`. */ +const writeAuth = (accessToken: string) => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const dir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "usage-reader-test-" }); + const authPath = path.join(dir, "auth.json"); + yield* fileSystem.writeFileString(authPath, JSON.stringify({ accessToken })); + return { dir, authPath }; + }); -beforeEach(async () => { - dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "usage-reader-test-")); -}); +/** One dashboard request as Cursor sees it: the URL, the fetch options and the decoded page. */ +interface DashboardRequest { + readonly url: string; + readonly init: RequestInit; + readonly page: number; + readonly headers: Headers; +} -afterEach(async () => { - await NodeFSP.rm(dir, { recursive: true, force: true }); -}); +/** Stands in for Cursor's dashboard: `answer` replies to each page request. */ +const dashboard = (answer: (request: DashboardRequest) => Response | Promise) => + HttpClient.make((request, url, signal, fiber) => + Effect.tryPromise({ + try: async () => { + const init = Context.getOrUndefined(fiber.context, FetchHttpClient.RequestInit); + const body = + request.body._tag === "Uint8Array" ? new TextDecoder().decode(request.body.body) : ""; + const response = await answer({ + url: url.toString(), + init: { ...init, signal, body }, + page: JSON.parse(body).page, + headers: new Headers(request.headers), + }); + return HttpClientResponse.fromWeb(request, response); + }, + catch: (cause) => + new HttpClientError.HttpClientError({ + reason: new HttpClientError.TransportError({ request, cause }), + }), + }), + ); + +/** Runs `readCursorAccountUsage` against `answer` standing in for Cursor. */ +const readAccountUsage = ( + credentialSource: Parameters[0], + sinceMs: number, + endDate: number, + answer: (request: DashboardRequest) => Response | Promise, + keychainToken?: Parameters[3], +) => + readCursorAccountUsage(credentialSource, sinceMs, endDate, keychainToken).pipe( + Effect.provideService(HttpClient.HttpClient, dashboard(answer)), + ); describe("Cursor account history", () => { - it("reads Cursor account history with the default macOS Keychain login", async () => { - const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; - let keychainReads = 0; - const result = await readCursorAccountUsage( - { kind: "keychain" }, - 0, - 1781000000000, - async (_url, init) => { - assert.include(new Headers(init.headers).get("cookie") ?? "", "demo%3A%3A"); - return Response.json({ totalUsageEventsCount: 0, usageEventsDisplay: [] }); - }, - async () => { - keychainReads++; - return accessToken; - }, - ); - assert.strictEqual(keychainReads, 1); - assert.isNull(result.error); - assert.isFalse(result.missing); - assert.isNotNull(result.accountKey); - }); - - it("reads paginated Cursor account history including headless calls with separate cache tokens", async () => { - const authPath = NodePath.join(dir, "auth.json"); - const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo", exp: 4102444800 })).toString("base64url")}.signature`; - await NodeFSP.writeFile(authPath, JSON.stringify({ accessToken })); - const pages: number[] = []; - const signals: AbortSignal[] = []; - const request = async (url: string, init: RequestInit) => { - assert.strictEqual(String(url), "https://cursor.com/api/dashboard/get-filtered-usage-events"); - assert.strictEqual(init?.redirect, "error"); - const headers = new Headers(init?.headers); - assert.strictEqual(headers.get("origin"), "https://cursor.com"); - assert.include(headers.get("cookie") ?? "", "WorkosCursorSessionToken=demo%3A%3A"); - const body = JSON.parse(String(init?.body)); - pages.push(body.page); - if (init.signal) signals.push(init.signal); - return Response.json({ - totalUsageEventsCount: 1001, - usageEventsDisplay: Array.from({ length: body.page === 1 ? 1000 : 1 }, (_, index) => ({ - timestamp: String(1780000000000 + ((body.page - 1) * 1000 + index) * 1000), - model: "claude-sonnet-4-5", - conversationId: `conversation-${body.page}`, - isHeadless: body.page === 2, - chargedCents: 0, - tokenUsage: { - inputTokens: 10, - outputTokens: 5, - cacheReadTokens: 30, - cacheWriteTokens: 2, - totalCents: 25, - }, - })), - }); - }; - const result = await readCursorAccountUsage(authPath, 0, 1781000000000, request); - assert.isNull(result.error); - assert.deepStrictEqual(pages, [1, 2]); - assert.lengthOf(signals, 2); - assert.notStrictEqual(signals[0], signals[1]); - assert.strictEqual(result.records.length, 1001); - assert.strictEqual(result.records.at(-1)?.sessionId, "conversation-2"); - assert.deepStrictEqual(result.records[0]?.totals, { - uncachedInputTokens: 10, - cachedInputTokens: 30, - cacheCreationTokens: 2, - outputTokens: 5, - reasoningTokens: 0, - }); - assert.strictEqual(result.records[0]?.reportedCostUsd, 0.25); - assert.isFalse(result.accountKey?.includes("demo") ?? true); - }); - - it("reads Cursor account history beyond 100 pages", async () => { - const authPath = NodePath.join(dir, "auth.json"); - const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; - await NodeFSP.writeFile(authPath, JSON.stringify({ accessToken })); - const fullPage = Array.from({ length: 1000 }, () => ({ tokenUsage: null })); - let requests = 0; - const result = await readCursorAccountUsage(authPath, 0, 1781000000000, async () => { - requests += 1; - return Response.json({ - totalUsageEventsCount: 100_001, - usageEventsDisplay: requests <= 100 ? fullPage : [{ tokenUsage: null }], - }); - }); - assert.isNull(result.error); - assert.strictEqual(requests, 101); - assert.deepStrictEqual(result.records, []); - }); - - it("accepts confirmed empty Cursor usage but rejects error envelopes", async () => { - const authPath = NodePath.join(dir, "auth.json"); - const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; - await NodeFSP.writeFile(authPath, JSON.stringify({ accessToken })); - for (const body of [ - {}, - { totalUsageEventsCount: 0 }, - { totalUsageEventsCount: 0, usageEventsDisplay: [] }, - ]) { - const result = await readCursorAccountUsage(authPath, 0, 1781000000000, async () => - Response.json(body), + it.effect("reads Cursor account history with the default macOS Keychain login", () => + Effect.gen(function* () { + const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; + let keychainReads = 0; + const result = yield* readAccountUsage( + { kind: "keychain" }, + 0, + 1781000000000, + ({ headers }) => { + assert.include(headers.get("cookie") ?? "", "demo%3A%3A"); + return Response.json({ totalUsageEventsCount: 0, usageEventsDisplay: [] }); + }, + Effect.sync(() => { + keychainReads++; + return accessToken; + }), ); + assert.strictEqual(keychainReads, 1); assert.isNull(result.error); - assert.deepStrictEqual(result.records, []); assert.isFalse(result.missing); - } - for (const body of [ - { error: "upstream error" }, - { detail: "unknown error envelope" }, - { totalUsageEventsCount: 0, error: "upstream error" }, - null, - [], - "invalid", - 0, - ]) { - const result = await readCursorAccountUsage(authPath, 0, 1781000000000, async () => - Response.json(body), - ); - assert.isNotNull(result.error); + assert.isNotNull(result.accountKey); + }).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect( + "reads paginated Cursor account history including headless calls with separate cache tokens", + () => + Effect.gen(function* () { + const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo", exp: 4102444800 })).toString("base64url")}.signature`; + const { authPath } = yield* writeAuth(accessToken); + const pages: number[] = []; + const signals: AbortSignal[] = []; + const request = ({ url, init, headers, page }: DashboardRequest) => { + assert.strictEqual(url, "https://cursor.com/api/dashboard/get-filtered-usage-events"); + assert.strictEqual(init.redirect, "error"); + assert.strictEqual(headers.get("origin"), "https://cursor.com"); + assert.strictEqual(headers.get("content-type"), "application/json"); + assert.include(headers.get("cookie") ?? "", "WorkosCursorSessionToken=demo%3A%3A"); + const body = { page }; + pages.push(body.page); + if (init.signal) signals.push(init.signal); + return Response.json({ + totalUsageEventsCount: 1001, + usageEventsDisplay: Array.from({ length: body.page === 1 ? 1000 : 1 }, (_, index) => ({ + timestamp: String(1780000000000 + ((body.page - 1) * 1000 + index) * 1000), + model: "claude-sonnet-4-5", + conversationId: `conversation-${body.page}`, + isHeadless: body.page === 2, + chargedCents: 0, + tokenUsage: { + inputTokens: 10, + outputTokens: 5, + cacheReadTokens: 30, + cacheWriteTokens: 2, + totalCents: 25, + }, + })), + }); + }; + const result = yield* readAccountUsage(authPath, 0, 1781000000000, request); + assert.isNull(result.error); + assert.deepStrictEqual(pages, [1, 2]); + assert.lengthOf(signals, 2); + assert.notStrictEqual(signals[0], signals[1]); + assert.strictEqual(result.records.length, 1001); + assert.strictEqual(result.records.at(-1)?.sessionId, "conversation-2"); + assert.deepStrictEqual(result.records[0]?.totals, { + uncachedInputTokens: 10, + cachedInputTokens: 30, + cacheCreationTokens: 2, + outputTokens: 5, + reasoningTokens: 0, + }); + assert.strictEqual(result.records[0]?.reportedCostUsd, 0.25); + assert.isFalse(result.accountKey?.includes("demo") ?? true); + }).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("reads Cursor account history beyond 100 pages", () => + Effect.gen(function* () { + const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; + const { authPath } = yield* writeAuth(accessToken); + const fullPage = Array.from({ length: 1000 }, () => ({ tokenUsage: null })); + let requests = 0; + const result = yield* readAccountUsage(authPath, 0, 1781000000000, async () => { + requests += 1; + return Response.json({ + totalUsageEventsCount: 100_001, + usageEventsDisplay: requests <= 100 ? fullPage : [{ tokenUsage: null }], + }); + }); + assert.isNull(result.error); + assert.strictEqual(requests, 101); assert.deepStrictEqual(result.records, []); - } - }); + }).pipe(Effect.provide(NodeServices.layer)), + ); - it("requires a terminal Cursor page after a full page reaches the reported count", async () => { - const authPath = NodePath.join(dir, "auth.json"); - const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; - await NodeFSP.writeFile(authPath, JSON.stringify({ accessToken })); - let requests = 0; - const result = await readCursorAccountUsage(authPath, 0, 1781000000000, async () => { - requests++; - return Response.json( - requests === 1 - ? { - totalUsageEventsCount: 1000, - usageEventsDisplay: Array.from({ length: 1000 }, (_, index) => ({ - timestamp: String(1780000000000 + index), - model: "gpt-5", - tokenUsage: { inputTokens: 10, outputTokens: 5 }, - })), - } - : { totalUsageEventsCount: 1000 }, - ); - }); - assert.isNull(result.error); - assert.strictEqual(result.records.length, 1000); - assert.strictEqual(requests, 2); - }); + it.effect("accepts confirmed empty Cursor usage but rejects error envelopes", () => + Effect.gen(function* () { + const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; + const { authPath } = yield* writeAuth(accessToken); + for (const body of [ + {}, + { totalUsageEventsCount: 0 }, + { totalUsageEventsCount: 0, usageEventsDisplay: [] }, + ]) { + const result = yield* readAccountUsage(authPath, 0, 1781000000000, async () => + Response.json(body), + ); + assert.isNull(result.error); + assert.deepStrictEqual(result.records, []); + assert.isFalse(result.missing); + } + for (const body of [ + { error: "upstream error" }, + { detail: "unknown error envelope" }, + { totalUsageEventsCount: 0, error: "upstream error" }, + null, + [], + "invalid", + 0, + ]) { + const result = yield* readAccountUsage(authPath, 0, 1781000000000, async () => + Response.json(body), + ); + assert.isNotNull(result.error); + assert.deepStrictEqual(result.records, []); + } + }).pipe(Effect.provide(NodeServices.layer)), + ); - it("removes only count-proven Cursor boundary copies and preserves identical billed events", async () => { - const authPath = NodePath.join(dir, "auth.json"); - const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; - await NodeFSP.writeFile(authPath, JSON.stringify({ accessToken })); - const event = (index: number) => ({ - timestamp: String(1780000000000 + index), - model: "gpt-5", - tokenUsage: { inputTokens: 10, outputTokens: 5, totalCents: 1 }, - }); - for (const total of [2000, 2001]) { + it.effect("requires a terminal Cursor page after a full page reaches the reported count", () => + Effect.gen(function* () { + const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; + const { authPath } = yield* writeAuth(accessToken); let requests = 0; - const result = await readCursorAccountUsage(authPath, 0, 1781000000000, async () => { + const result = yield* readAccountUsage(authPath, 0, 1781000000000, async () => { requests++; - return Response.json({ - totalUsageEventsCount: total, - usageEventsDisplay: - requests === 1 - ? Array.from({ length: 1000 }, (_, index) => event(index)) - : requests === 2 - ? Array.from({ length: 1000 }, (_, index) => event(999 + index)) - : [event(1999)], - }); - }); - assert.isNull(result.error); - assert.strictEqual(result.records.length, total); - assert.strictEqual(requests, 3); - assert.strictEqual(result.records.at(-1)?.timestampMs, 1780000001999); - assert.strictEqual( - result.records.filter((record) => record.timestampMs === 1780000000999).length, - total === 2000 ? 1 : 2, - ); - assert.strictEqual(new Set(result.records.map((record) => record.dedupeKey)).size, total); - } - let requests = 0; - const inconsistent = await readCursorAccountUsage(authPath, 0, 1781000000000, async () => { - requests++; - return Response.json({ - totalUsageEventsCount: 1001, - usageEventsDisplay: + return Response.json( requests === 1 - ? Array.from({ length: 1000 }, (_, index) => event(index)) - : [event(500), event(1000)], + ? { + totalUsageEventsCount: 1000, + usageEventsDisplay: Array.from({ length: 1000 }, (_, index) => ({ + timestamp: String(1780000000000 + index), + model: "gpt-5", + tokenUsage: { inputTokens: 10, outputTokens: 5 }, + })), + } + : { totalUsageEventsCount: 1000 }, + ); }); - }); - assert.isNotNull(inconsistent.error); - assert.deepStrictEqual(inconsistent.records, []); - }); + assert.isNull(result.error); + assert.strictEqual(result.records.length, 1000); + assert.strictEqual(requests, 2); + }).pipe(Effect.provide(NodeServices.layer)), + ); - it("does not present truncated Cursor account pages or authentication failures as complete history", async () => { - const authPath = NodePath.join(dir, "auth.json"); - const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo", exp: 4102444800 })).toString("base64url")}.signature`; - await NodeFSP.writeFile(authPath, JSON.stringify({ accessToken })); - const truncated = await readCursorAccountUsage(authPath, 0, 1781000000000, async () => - Response.json({ totalUsageEventsCount: 101, usageEventsDisplay: [] }), - ); - assert.isNotNull(truncated.error); - assert.deepStrictEqual(truncated.records, []); - const denied = await readCursorAccountUsage( - authPath, - 0, - 1781000000000, - async () => new Response(accessToken, { status: 401 }), - ); - assert.isNotNull(denied.error); - assert.isFalse(denied.error?.includes(accessToken) ?? true); - assert.deepStrictEqual(denied.records, []); - let requested = false; - const missing = await readCursorAccountUsage( - NodePath.join(dir, "missing.json"), - 0, - 1781000000000, - async () => { - requested = true; - return Response.json({}); - }, - ); - assert.isTrue(missing.missing); - assert.isFalse(requested); - }); + it.effect( + "removes only count-proven Cursor boundary copies and preserves identical billed events", + () => + Effect.gen(function* () { + const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; + const { authPath } = yield* writeAuth(accessToken); + const event = (index: number) => ({ + timestamp: String(1780000000000 + index), + model: "gpt-5", + tokenUsage: { inputTokens: 10, outputTokens: 5, totalCents: 1 }, + }); + for (const total of [2000, 2001]) { + let requests = 0; + const result = yield* readAccountUsage(authPath, 0, 1781000000000, async () => { + requests++; + return Response.json({ + totalUsageEventsCount: total, + usageEventsDisplay: + requests === 1 + ? Array.from({ length: 1000 }, (_, index) => event(index)) + : requests === 2 + ? Array.from({ length: 1000 }, (_, index) => event(999 + index)) + : [event(1999)], + }); + }); + assert.isNull(result.error); + assert.strictEqual(result.records.length, total); + assert.strictEqual(requests, 3); + assert.strictEqual(result.records.at(-1)?.timestampMs, 1780000001999); + assert.strictEqual( + result.records.filter((record) => record.timestampMs === 1780000000999).length, + total === 2000 ? 1 : 2, + ); + assert.strictEqual(new Set(result.records.map((record) => record.dedupeKey)).size, total); + } + let requests = 0; + const inconsistent = yield* readAccountUsage(authPath, 0, 1781000000000, async () => { + requests++; + return Response.json({ + totalUsageEventsCount: 1001, + usageEventsDisplay: + requests === 1 + ? Array.from({ length: 1000 }, (_, index) => event(index)) + : [event(500), event(1000)], + }); + }); + assert.isNotNull(inconsistent.error); + assert.deepStrictEqual(inconsistent.records, []); + }).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect( + "does not present truncated Cursor account pages or authentication failures as complete history", + () => + Effect.gen(function* () { + const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo", exp: 4102444800 })).toString("base64url")}.signature`; + const { dir, authPath } = yield* writeAuth(accessToken); + const truncated = yield* readAccountUsage(authPath, 0, 1781000000000, async () => + Response.json({ totalUsageEventsCount: 101, usageEventsDisplay: [] }), + ); + assert.isNotNull(truncated.error); + assert.deepStrictEqual(truncated.records, []); + const denied = yield* readAccountUsage( + authPath, + 0, + 1781000000000, + async () => new Response(accessToken, { status: 401 }), + ); + assert.isNotNull(denied.error); + assert.isFalse(denied.error?.includes(accessToken) ?? true); + assert.deepStrictEqual(denied.records, []); + let requested = false; + const missing = yield* readAccountUsage( + (yield* Path.Path).join(dir, "missing.json"), + 0, + 1781000000000, + async () => { + requested = true; + return Response.json({}); + }, + ); + assert.isTrue(missing.missing); + assert.isFalse(requested); + }).pipe(Effect.provide(NodeServices.layer)), + ); }); describe("readCursorAccountUsage", () => { - it("asks for Keychain approval when the prompt goes unanswered", async () => { - const result = await readCursorAccountUsage( - { kind: "keychain" }, - 0, - 1, - () => Promise.reject(new Error("no network expected")), - () => Promise.reject(new CursorKeychainTimeoutError()), - ); - assert.deepStrictEqual(result, { - accountKey: null, - records: [], - missing: false, - error: "Allow Keychain access on the Mac running T3 Code, then refresh.", - }); - }); + it.effect("asks for Keychain approval when the prompt goes unanswered", () => + Effect.gen(function* () { + const result = yield* readAccountUsage( + { kind: "keychain" }, + 0, + 1, + () => { + throw new Error("no network expected"); + }, + Effect.fail(new CursorKeychain.CursorKeychainTimeoutError()), + ); + assert.deepStrictEqual(result, { + accountKey: null, + records: [], + missing: false, + error: "Allow Keychain access on the Mac running T3 Code, then refresh.", + }); + }).pipe(Effect.provide(NodeServices.layer)), + ); - it("reads the pages behind the first together and keeps them in page order", async () => { - const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; - let inFlight = 0; - let mostInFlight = 0; - const result = await readCursorAccountUsage( - { kind: "keychain" }, - 0, - 1781000000000, - async (_url, init) => { - const page: number = JSON.parse(String(init.body)).page; - inFlight++; - mostInFlight = Math.max(mostInFlight, inFlight); - // Later pages answer first, so record order cannot come from arrival order. - for (let turn = page; turn < 9; turn++) await Promise.resolve(); - inFlight--; - return Response.json({ - totalUsageEventsCount: 8001, - usageEventsDisplay: Array.from({ length: page === 9 ? 1 : 1000 }, (_, index) => ({ - timestamp: String(1780000000000 + (page - 1) * 1000 + index), - model: "gpt-5", - tokenUsage: { inputTokens: 1 }, - })), - }); - }, - async () => accessToken, - ); - assert.isNull(result.error); - assert.strictEqual(mostInFlight, 6); - assert.deepStrictEqual( - result.records.map((record) => record.timestampMs), - Array.from({ length: 8001 }, (_, index) => 1780000000000 + index), - ); - }); - - it("ends like a page-by-page read when a page fails, leaving no request open", async () => { - const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; - const fullPage = { - totalUsageEventsCount: 8001, - usageEventsDisplay: Array.from({ length: 1000 }, () => ({ tokenUsage: null })), - }; - // Page 3 fails while the pages after it are still waiting on Cursor. - for (const [secondPage, error] of [ - [() => Response.json(fullPage), "Cursor account usage could not be read."], - [() => new Response(null, { status: 401 }), "Sign in to Cursor again to read account usage."], - ] as const) { - let open = 0; - const result = await readCursorAccountUsage( + it.effect("reads the pages behind the first together and keeps them in page order", () => + Effect.gen(function* () { + const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; + let inFlight = 0; + let mostInFlight = 0; + const result = yield* readAccountUsage( { kind: "keychain" }, 0, 1781000000000, - (_url, init) => { - const page: number = JSON.parse(String(init.body)).page; - if (page === 1) return Promise.resolve(Response.json(fullPage)); - if (page === 2) return Promise.resolve(secondPage()); - if (page === 3) return Promise.reject(new Error("connection reset")); - open++; - return new Promise((_resolve, reject) => { - init.signal?.addEventListener("abort", () => { - open--; - reject(init.signal?.reason); - }); + async ({ page }) => { + inFlight++; + mostInFlight = Math.max(mostInFlight, inFlight); + // Later pages answer first, so record order cannot come from arrival order. + for (let turn = page; turn < 9; turn++) await Promise.resolve(); + inFlight--; + return Response.json({ + totalUsageEventsCount: 8001, + usageEventsDisplay: Array.from({ length: page === 9 ? 1 : 1000 }, (_, index) => ({ + timestamp: String(1780000000000 + (page - 1) * 1000 + index), + model: "gpt-5", + tokenUsage: { inputTokens: 1 }, + })), }); }, - async () => accessToken, + Effect.succeed(accessToken), ); - assert.strictEqual(result.error, error); - assert.deepStrictEqual(result.records, []); - assert.strictEqual(open, 0); - } - }); + assert.isNull(result.error); + assert.strictEqual(mostInFlight, 6); + assert.deepStrictEqual( + result.records.map((record) => record.timestampMs), + Array.from({ length: 8001 }, (_, index) => 1780000000000 + index), + ); + }).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("ends like a page-by-page read when a page fails, leaving no request open", () => + Effect.gen(function* () { + const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`; + const fullPage = { + totalUsageEventsCount: 8001, + usageEventsDisplay: Array.from({ length: 1000 }, () => ({ tokenUsage: null })), + }; + // Page 3 fails while the pages after it are still waiting on Cursor. + for (const [secondPage, error] of [ + [() => Response.json(fullPage), "Cursor account usage could not be read."], + [ + () => new Response(null, { status: 401 }), + "Sign in to Cursor again to read account usage.", + ], + ] as const) { + let open = 0; + const result = yield* readAccountUsage( + { kind: "keychain" }, + 0, + 1781000000000, + ({ init, page }) => { + if (page === 1) return Promise.resolve(Response.json(fullPage)); + if (page === 2) return Promise.resolve(secondPage()); + if (page === 3) return Promise.reject(new Error("connection reset")); + open++; + return new Promise((_resolve, reject) => { + init.signal?.addEventListener("abort", () => { + open--; + reject(init.signal?.reason); + }); + }); + }, + Effect.succeed(accessToken), + ); + assert.strictEqual(result.error, error); + assert.deepStrictEqual(result.records, []); + assert.strictEqual(open, 0); + } + }).pipe(Effect.provide(NodeServices.layer)), + ); }); diff --git a/packages/provider-cursor/src/server/accountUsage.ts b/packages/provider-cursor/src/server/accountUsage.ts index 4ecab1ace8769ee97b79e6dac94358da57cf8f4a..11976a2e0f1da5c378b5ca632768b4a1d28a81d1 100644 --- a/packages/provider-cursor/src/server/accountUsage.ts +++ b/packages/provider-cursor/src/server/accountUsage.ts @@ -1,5 +1,3 @@ -// Node fs reads CLI credentials, and crypto hashes account IDs for deduplication. -// @effect-diagnostics nodeBuiltinImport:off /** * Cursor account usage from its dashboard API, read with the saved CLI or * Keychain login. The dashboard covers CLI, desktop and headless usage from @@ -7,13 +5,22 @@ * * @module provider-cursor/server/accountUsage */ -import * as NodeFSP from "node:fs/promises"; -import * as NodeCrypto from "node:crypto"; -import * as NodeTimersPromises from "node:timers/promises"; - import type { UsageRecord } from "@t3tools/provider-core/server/usage"; +import * as Crypto from "effect/Crypto"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Base64Url from "effect/encoding/Base64Url"; +import * as Hex from "effect/encoding/Hex"; +import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +import * as FetchHttpClient from "effect/http/FetchHttpClient"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientRequest from "effect/http/HttpClientRequest"; +import * as PlatformError from "effect/PlatformError"; +import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; -import { isCursorKeychainTimeoutError } from "./CursorKeychain.ts"; +import * as CursorKeychain from "./CursorKeychain.ts"; function object(value: unknown): Record { return typeof value === "object" && value !== null && !Array.isArray(value) @@ -44,7 +51,27 @@ export interface CursorAccountUsageReadResult { readonly error: string | null; } -const accountHash = (value: string) => NodeCrypto.createHash("sha256").update(value).digest("hex"); +/** Why a dashboard read was abandoned; the caller only sees the generic message. */ +class CursorAccountUsageInvalidError extends Schema.TaggedError()( + "CursorAccountUsageInvalidError", + { reason: Schema.String }, +) {} + +const invalid = (reason: string) => Effect.fail(new CursorAccountUsageInvalidError({ reason })); + +const decodeJson = Schema.decodeEffect(Schema.fromJsonString(Schema.Unknown)); + +const pageRequestBody = (page: number, pageSize: number, sinceMs: number, endDate: number) => + JSON.stringify({ page, pageSize, startDate: String(sinceMs), endDate: String(endDate) }); + +const billedEventIdentity = (...fields: readonly unknown[]) => JSON.stringify(fields); + +const textEncoder = new TextEncoder(); + +const accountHash = Effect.fn("accountHash")(function* (value: string) { + const crypto = yield* Crypto.Crypto; + return Hex.encode(yield* crypto.digest("SHA-256", textEncoder.encode(value))); +}); function canonicalJson(value: unknown): string { if (Array.isArray(value)) return `[${value.map(canonicalJson).join(",")}]`; @@ -70,26 +97,54 @@ function boundaryOverlap(previous: readonly string[], current: readonly string[] return lengths.at(-1) ?? 0; } +const DASHBOARD_USAGE_URL = "https://cursor.com/api/dashboard/get-filtered-usage-events"; + +/** The session cookie must never follow a redirect off cursor.com. */ +const NO_REDIRECT: RequestInit = { redirect: "error" }; + /** * Dashboard usage includes headless agents and reports fresh input separately - * from cache reads. `keychainToken` is only called for a Keychain login. + * from cache reads. `keychainToken` is only run for a Keychain login. */ -export async function readCursorAccountUsage( +export const readCursorAccountUsage = Effect.fn("readCursorAccountUsage")(function* ( credentialSource: string | { readonly kind: "keychain" }, sinceMs: number, endDate: number, - request: (url: string, init: RequestInit) => Promise = globalThis.fetch, - keychainToken: () => Promise = () => - Promise.reject(new Error("No Keychain reader")), -): Promise { - let accessToken: unknown; - try { - accessToken = - typeof credentialSource === "string" - ? object(JSON.parse(await NodeFSP.readFile(credentialSource, "utf8"))).accessToken - : await keychainToken(); - } catch (cause) { - const missing = typeof credentialSource === "string" && object(cause).code === "ENOENT"; + keychainToken: Effect.Effect< + string | null, + | CursorKeychain.CursorKeychainTimeoutError + | CursorKeychain.CursorKeychainReadError + | CursorAccountUsageInvalidError + > = invalid("No Keychain reader"), +): Effect.fn.Return< + CursorAccountUsageReadResult, + never, + FileSystem.FileSystem | Crypto.Crypto | HttpClient.HttpClient +> { + const fileSystem = yield* FileSystem.FileSystem; + const httpClient = yield* HttpClient.HttpClient; + const credential = yield* Effect.result< + unknown, + | PlatformError.PlatformError + | Schema.SchemaError + | CursorKeychain.CursorKeychainTimeoutError + | CursorKeychain.CursorKeychainReadError + | CursorAccountUsageInvalidError, + never + >( + typeof credentialSource === "string" + ? fileSystem.readFileString(credentialSource).pipe( + Effect.flatMap(decodeJson), + Effect.map((credentials) => object(credentials).accessToken), + ) + : keychainToken, + ); + if (Result.isFailure(credential)) { + const cause = credential.failure; + const missing = + typeof credentialSource === "string" && + PlatformError.isPlatformError(cause) && + cause.reason._tag === "NotFound"; return { accountKey: null, records: [], @@ -98,11 +153,12 @@ export async function readCursorAccountUsage( ? null : typeof credentialSource === "string" ? "Cursor credentials could not be read." - : isCursorKeychainTimeoutError(cause) + : CursorKeychain.isCursorKeychainTimeoutError(cause) ? "Allow Keychain access on the Mac running T3 Code, then refresh." : "Cursor Keychain credentials could not be read.", }; } + const accessToken = credential.success; if (typeof accessToken !== "string" || !accessToken) { return { accountKey: null, @@ -115,95 +171,112 @@ export async function readCursorAccountUsage( }; } let accountKey: string | null = null; - const cancel = new AbortController(); - try { - const payload = accessToken.split(".")[1]; + const read = Effect.gen(function* () { + const payload = Base64Url.decodeString(accessToken.split(".")[1] ?? ""); + if (Result.isFailure(payload)) return yield* invalid("Invalid authentication"); const subject = object( - JSON.parse(Buffer.from(payload ?? "", "base64url").toString("utf8")), + yield* decodeJson(payload.success).pipe( + Effect.catch(() => invalid("Invalid authentication")), + ), ).sub; - if (typeof subject !== "string" || !subject) throw new Error("Invalid authentication"); + if (typeof subject !== "string" || !subject) return yield* invalid("Invalid authentication"); const userId = subject.split("|").at(-1); - if (!userId) throw new Error("Invalid authentication"); - accountKey = accountHash(subject); + if (!userId) return yield* invalid("Invalid authentication"); + accountKey = yield* accountHash(subject); if (!Number.isFinite(sinceMs) || !Number.isFinite(endDate) || sinceMs < 0 || sinceMs > endDate) - throw new Error("Invalid date window"); - const deadline = AbortSignal.any([cancel.signal, AbortSignal.timeout(60_000)]); + return yield* invalid("Invalid date window"); const records: UsageRecord[] = []; const occurrences = new Map(); const pages: unknown[][] = []; let completed = false; const pageSize = 1000; let total: number | undefined; - const readPage = async (page: number) => { - // A count can include overlapping page boundaries. Allow room to - // reconcile them without imposing a fixed account-size limit. - if (page > (total === undefined ? 1000 : Math.ceil(total / pageSize) * 2 + 1)) { - throw new Error("Account usage page limit exceeded"); - } - const response = await request("https://cursor.com/api/dashboard/get-filtered-usage-events", { - method: "POST", - redirect: "error", - signal: AbortSignal.any([deadline, AbortSignal.timeout(10_000)]), - headers: { - "Content-Type": "application/json", - Origin: "https://cursor.com", - Cookie: `WorkosCursorSessionToken=${encodeURIComponent(`${userId}::${accessToken}`)}`, - }, - body: JSON.stringify({ - page, - pageSize, - startDate: String(sinceMs), - endDate: String(endDate), + const readPage = (page: number) => + Effect.gen(function* () { + // A count can include overlapping page boundaries. Allow room to + // reconcile them without imposing a fixed account-size limit. + if (page > (total === undefined ? 1000 : Math.ceil(total / pageSize) * 2 + 1)) { + return yield* invalid("Account usage page limit exceeded"); + } + const response = yield* httpClient + .execute( + HttpClientRequest.post(DASHBOARD_USAGE_URL).pipe( + HttpClientRequest.setHeaders({ + Origin: "https://cursor.com", + Cookie: `WorkosCursorSessionToken=${encodeURIComponent(`${userId}::${accessToken}`)}`, + }), + HttpClientRequest.bodyText( + pageRequestBody(page, pageSize, sinceMs, endDate), + "application/json", + ), + ), + ) + .pipe( + Effect.provideService(FetchHttpClient.RequestInit, NO_REDIRECT), + Effect.mapError( + () => new CursorAccountUsageInvalidError({ reason: "Account usage request failed" }), + ), + ); + if (response.status === 401 || response.status === 403) { + return { + accountKey, + records: [], + missing: false, + error: "Sign in to Cursor again to read account usage.", + } satisfies CursorAccountUsageReadResult; + } + if (response.status < 200 || response.status >= 300) + return yield* invalid("Account usage request failed"); + const parsed: unknown = yield* response.json.pipe( + Effect.mapError( + () => new CursorAccountUsageInvalidError({ reason: "Invalid account usage page" }), + ), + ); + if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) { + return yield* invalid("Invalid account usage page"); + } + const body = object(parsed); + const keys = Object.keys(body); + if ("error" in body || "message" in body || "code" in body) + return yield* invalid("Account usage error response"); + const count = keys.length === 0 ? 0 : body.totalUsageEventsCount; + const events = + keys.length === 0 || (keys.length === 1 && keys[0] === "totalUsageEventsCount") + ? [] + : body.usageEventsDisplay; + if ( + (count !== undefined && + (typeof count !== "number" || + !Number.isSafeInteger(count) || + count < 0 || + (total !== undefined && count !== total))) || + !Array.isArray(events) || + events.length > pageSize || + (count === undefined && !Array.isArray(body.usageEventsDisplay)) + ) { + return yield* invalid("Inconsistent account usage page"); + } + if (typeof count === "number") total = count; + return events as unknown[]; + }).pipe( + Effect.timeoutOrElse({ + duration: Duration.seconds(10), + orElse: () => invalid("Account usage request timed out"), }), - }); - if (response.status === 401 || response.status === 403) { - return { - accountKey, - records: [], - missing: false, - error: "Sign in to Cursor again to read account usage.", - }; - } - if (!response.ok) throw new Error("Account usage request failed"); - const parsed: unknown = await response.json(); - if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) { - throw new Error("Invalid account usage page"); - } - const body = object(parsed); - const keys = Object.keys(body); - if ("error" in body || "message" in body || "code" in body) - throw new Error("Account usage error response"); - const count = keys.length === 0 ? 0 : body.totalUsageEventsCount; - const events = - keys.length === 0 || (keys.length === 1 && keys[0] === "totalUsageEventsCount") - ? [] - : body.usageEventsDisplay; - if ( - (count !== undefined && - (typeof count !== "number" || - !Number.isSafeInteger(count) || - count < 0 || - (total !== undefined && count !== total))) || - !Array.isArray(events) || - events.length > pageSize || - (count === undefined && !Array.isArray(body.usageEventsDisplay)) - ) { - throw new Error("Inconsistent account usage page"); - } - if (typeof count === "number") total = count; - return events; - }; + ); // Cursor caps a page at 1,000 events and takes about a second to answer one, so the pages // the first one's count implies are requested up to six ahead and consumed in page order. - const ahead: ReturnType[] = []; + // Pages still in flight when the read ends are interrupted with its scope. + const ahead: Fiber.Fiber< + Effect.Success>, + CursorAccountUsageInvalidError + >[] = []; for (let page = 1; ; page++) { while (ahead.length < Math.min(6, Math.ceil((total ?? 0) / pageSize) - page + 1)) { - const pending = readPage(page + ahead.length); - // Handled when its turn comes, or cancelled if the read ends first. - pending.catch(() => undefined); - ahead.push(pending); + ahead.push(yield* Effect.forkScoped(readPage(page + ahead.length))); } - const events = await (ahead.shift() ?? readPage(page)); + const next = ahead.shift(); + const events = yield* next === undefined ? readPage(page) : Fiber.join(next); // A rejected login comes back as the finished result instead of a page. if (!Array.isArray(events)) return events; pages.push(events); @@ -212,14 +285,17 @@ export async function readCursorAccountUsage( break; } } - if (!completed) throw new Error("Account usage page limit exceeded"); + if (!completed) return yield* invalid("Account usage page limit exceeded"); const rawCount = pages.reduce((sum, page) => sum + page.length, 0); - if (total !== undefined && rawCount < total) throw new Error("Incomplete account usage pages"); + if (total !== undefined && rawCount < total) + return yield* invalid("Incomplete account usage pages"); let removalsRemaining = total === undefined ? 0 : rawCount - total; let previousKeys: string[] = []; for (const events of pages) { const eventKeys = - removalsRemaining > 0 ? events.map((event) => accountHash(canonicalJson(event))) : []; + removalsRemaining > 0 + ? yield* Effect.forEach(events, (event) => accountHash(canonicalJson(event))) + : []; const removalCount = Math.min(removalsRemaining, boundaryOverlap(previousKeys, eventKeys)); removalsRemaining -= removalCount; previousKeys = eventKeys; @@ -239,7 +315,7 @@ export async function readCursorAccountUsage( value !== undefined && (typeof value !== "number" || !Number.isFinite(value) || value < 0) ) { - throw new Error("Invalid account usage totals"); + return yield* invalid("Invalid account usage totals"); } } const timestampMs = @@ -252,7 +328,7 @@ export async function readCursorAccountUsage( typeof event.model !== "string" || !event.model ) - throw new Error("Invalid account usage event"); + return yield* invalid("Invalid account usage event"); if (timestampMs < sinceMs || timestampMs > endDate) continue; const totals = { uncachedInputTokens: tokens(usage.inputTokens), @@ -265,8 +341,8 @@ export async function readCursorAccountUsage( typeof usage.totalCents === "number" ? usage.totalCents / 100 : null; const sessionId = typeof event.conversationId === "string" ? event.conversationId : ""; // No event ID is provided. Preserve identical billed rows with an occurrence index. - const key = accountHash( - JSON.stringify([timestampMs, event.model, sessionId, totals, reportedCostUsd]), + const key = yield* accountHash( + billedEventIdentity(timestampMs, event.model, sessionId, totals, reportedCostUsd), ); const occurrence = occurrences.get(key) ?? 0; occurrences.set(key, occurrence + 1); @@ -282,19 +358,29 @@ export async function readCursorAccountUsage( dedupeKey: `cursor-account:${accountKey}:${key}:${occurrence}`, }); } - await NodeTimersPromises.setImmediate(); + yield* Effect.yieldNow; } - if (removalsRemaining !== 0) throw new Error("Inconsistent account usage boundaries"); - return { accountKey, records, missing: false, error: null }; - } catch { + if (removalsRemaining !== 0) return yield* invalid("Inconsistent account usage boundaries"); return { accountKey, - records: [], + records, missing: false, - error: "Cursor account usage could not be read.", - }; - } finally { - // An early exit leaves the pages requested ahead of it in flight. - cancel.abort(); - } -} + error: null, + } satisfies CursorAccountUsageReadResult; + }); + return yield* read.pipe( + Effect.scoped, + Effect.timeoutOrElse({ + duration: Duration.seconds(60), + orElse: () => invalid("Account usage read timed out"), + }), + Effect.catch(() => + Effect.succeed({ + accountKey, + records: [], + missing: false, + error: "Cursor account usage could not be read.", + } satisfies CursorAccountUsageReadResult), + ), + ); +}); diff --git a/packages/provider-opencode/src/server/usage.test.ts b/packages/provider-opencode/src/server/usage.test.ts index 8389f85a5f3f4a83778514836a0b676c4ffbd40d..5a517e40c5cb9e72438a895e4f9ad48f936451d0 100644 --- a/packages/provider-opencode/src/server/usage.test.ts +++ b/packages/provider-opencode/src/server/usage.test.ts @@ -1,31 +1,31 @@ -// @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 * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as SqlClient from "effect/sql/SqlClient"; 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 }); -}); +/** A writable connection that stays open until the test's scope closes. */ +const openDatabase = (filename: string) => + Layer.build(NodeSqliteClient.layer({ filename })).pipe( + Effect.map(Context.get(SqlClient.SqlClient)), + ); 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)", - ); + it.effect("counts migrated OpenCode messages once and sees subsequent WAL writes", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const dir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "usage-reader-test-" }); + const db = yield* openDatabase(path.join(dir, "opencode.db")); + yield* db.unsafe("PRAGMA journal_mode = WAL"); + yield* db.unsafe("PRAGMA wal_autocheckpoint = 0"); + yield* db.unsafe("CREATE TABLE message (id TEXT, session_id TEXT, data TEXT)"); const message = { id: "msg-1", sessionID: "session-1", @@ -35,12 +35,13 @@ describe("readOpenCodeUsage", () => { 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); + const insert = (id: string, sessionId: string, data: string) => + db.unsafe("INSERT INTO message VALUES (?, ?, ?)", [id, sessionId, data]); + yield* insert(message.id, message.sessionID, JSON.stringify(message)); + const legacy = path.join(dir, "storage", "message", message.sessionID); + yield* fileSystem.makeDirectory(legacy, { recursive: true }); + yield* fileSystem.writeFileString(path.join(legacy, "msg-1.json"), JSON.stringify(message)); + const first = yield* readOpenCodeUsage(dir, 0); assert.isFalse(first.error); const records = first.files.flatMap((file) => file.records); assert.strictEqual(records.length, 1); @@ -52,20 +53,18 @@ describe("readOpenCodeUsage", () => { reasoningTokens: 5, }); assert.strictEqual(records[0]?.reportedCostUsd, 0.25); - insert.run( + yield* insert( "msg-2", message.sessionID, JSON.stringify({ ...message, id: "msg-2", time: { created: 1780000001000 } }), ); - const next = await readOpenCodeUsage(dir, 1780000001000); + const next = yield* 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(); - } - }); + assert.isAbove(Number((yield* fileSystem.stat(path.join(dir, "opencode.db-wal"))).size), 0); + }).pipe(Effect.provide(NodeServices.layer)), + ); }); diff --git a/packages/provider-opencode/src/server/usage.ts b/packages/provider-opencode/src/server/usage.ts index 9cc5ce2d09109a74c7ac016757835476c189fde8..f0005ad616089f3084b462ee3cca60bec0c55334 100644 --- a/packages/provider-opencode/src/server/usage.ts +++ b/packages/provider-opencode/src/server/usage.ts @@ -1,16 +1,10 @@ -// 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 { @@ -19,9 +13,13 @@ import { type UsageRecord, } from "@t3tools/provider-core/server/usage"; import { HostProcessEnvironment } from "@t3tools/shared/hostProcess"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; +import type * as PlatformError from "effect/PlatformError"; +import * as SqlClient from "effect/sql/SqlClient"; import type { OpenCodeSettings } from "../settings.ts"; @@ -93,11 +91,23 @@ export interface OpenCodeUsageReadResult { readonly error: boolean; } +const isNotFound = (cause: PlatformError.PlatformError) => cause.reason._tag === "NotFound"; + +/** A regular file or directory, never a symlink to one. */ +const entryType = Effect.fn("entryType")(function* (path: string) { + const fileSystem = yield* FileSystem.FileSystem; + const link = yield* Effect.exit(fileSystem.readLink(path)); + if (Exit.isSuccess(link)) return "SymbolicLink" as const; + return (yield* fileSystem.stat(path)).type; +}); + /** Reads current SQLite and pre-migration JSON stores without modifying either. */ -export async function readOpenCodeUsage( +export const readOpenCodeUsage = Effect.fn("readOpenCodeUsage")(function* ( root: string, sinceMs: number, -): Promise { +): Effect.fn.Return { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; const files: { path: string; records: UsageRecord[] }[] = []; const seen = new Set(); let found = false; @@ -111,49 +121,69 @@ export async function readOpenCodeUsage( 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; - } + const databases = yield* fileSystem.readDirectory(root).pipe( + Effect.flatMap((names) => + Effect.filter( + names.filter((name) => /^opencode(?:-[a-zA-Z0-9_-]+)?\.db$/.test(name)), + // One database removed between listing and stat must not hide the rest. + (name) => + entryType(path.join(root, name)).pipe( + Effect.map((type) => type === "File"), + Effect.catchTags({ + PlatformError: (cause) => + isNotFound(cause) ? Effect.succeed(false) : Effect.fail(cause), + }), + ), + ), + ), + Effect.map((names) => + names.sort((a, b) => + a === "opencode.db" ? -1 : b === "opencode.db" ? 1 : a.localeCompare(b), + ), + ), + Effect.catchTags({ + PlatformError: (cause) => { + if (!isNotFound(cause)) error = true; + return Effect.succeed([]); + }, + }), + ); for (const name of databases) { found = true; - const file = { path: NodePath.join(root, name), records: [] as UsageRecord[] }; + const file = { path: path.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 }); + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; // 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"); + yield* sql.unsafe("PRAGMA busy_timeout = 100"); const tables = new Set( - database - .prepare("SELECT name FROM sqlite_master WHERE type = 'table'") - .all() - .map((row) => row.name), + (yield* sql.unsafe<{ readonly name: unknown }>( + "SELECT name FROM sqlite_master WHERE type = 'table'", + )).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), + (yield* sql.unsafe<{ readonly name: unknown }>(`PRAGMA table_info(${table})`)).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( + const rows = yield* sql.unsafe<{ + readonly id: unknown; + readonly session_id: unknown; + readonly data: unknown; + readonly created: unknown; + }>( `SELECT id, session_id, data, ${timestamp} AS created FROM ${table}${where}`, + timestamp === "NULL" ? [] : [sinceMs], ); - let count = 0; - for (const row of statement.iterate(...(timestamp === "NULL" ? [] : [sinceMs]))) { + for (const [index, row] of rows.entries()) { append( file.records, parseOpenCodeMessage(text(row.data), { @@ -162,48 +192,64 @@ export async function readOpenCodeUsage( ...(typeof row.created === "number" ? { timestampMs: row.created } : {}), }), ); - if (++count % 256 === 0) await NodeTimersPromises.setImmediate(); + if (index % 256 === 255) yield* Effect.yieldNow; } } - } catch { - error = true; - } finally { - database?.close(); - } + }).pipe( + Effect.provide(NodeSqliteClient.layer({ filename: file.path, readonly: true })), + Effect.catchTags({ + SqlError: () => { + error = true; + return Effect.void; + }, + }), + ); } // 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")]; + const directories = [path.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")) { + yield* Effect.gen(function* () { + for (const name of yield* fileSystem.readDirectory(directory)) { + const entry = path.join(directory, name); + const type = yield* entryType(entry).pipe( + Effect.catchTags({ + PlatformError: (cause) => + isNotFound(cause) ? Effect.succeed(null) : Effect.fail(cause), + }), + ); + if (type === "Directory") { + directories.push(entry); + } else if (type === "File" && name.endsWith(".json")) { found = true; - const id = entry.name.slice(0, -5); + const id = name.slice(0, -5); if (seen.has(`opencode:${id}`)) continue; - const file = { path, records: [] as UsageRecord[] }; + const file = { path: entry, 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; - } + yield* fileSystem.readFileString(entry).pipe( + Effect.map((source) => append(file.records, parseOpenCodeMessage(source, { id }))), + Effect.catchTags({ + PlatformError: (cause) => { + if (!isNotFound(cause)) error = true; + return Effect.void; + }, + }), + ); } } - } catch (cause) { - if (object(cause).code !== "ENOENT") error = true; - } + }).pipe( + Effect.catchTags({ + PlatformError: (cause) => { + if (!isNotFound(cause)) error = true; + return Effect.void; + }, + }), + ); } return { files, missing: !found && !error, error }; -} +}); /** * The data directories to read: `OPENCODE_DATA_DIR` (comma-separated) or the @@ -244,7 +290,7 @@ export const openCodeUsageReader: ProviderUsageReader - Effect.promise(() => readOpenCodeUsage(dir, windowStartMs)).pipe( + readOpenCodeUsage(dir, windowStartMs).pipe( Effect.map((result) => ({ dir, files: result.missing && !result.error ? null : result.files,