authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-09 15:26:20-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-09 15:26:20-07:00
logb1ec4b36876e7f3ad03a1e9e921286ec7d58fe95
tree5fd94c1bd2aa4e2514d06027d567bde5ab0a9a78
parent9a0766b0238f2c044a965dcba32c6cb7ea0b63cd
signature Signed by PGP key B5690EEEBB952194

refactor(usage): usage readers use Effect FileSystem and SqlClient (#17615)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

8 files changed, 1280 insertions(+), 968 deletions(-)

apps/server/src/provider/Drivers/AntigravityUsage.ts+3-2
......@@ -106,8 +106,9 @@ const make = Effect.gen(function* () {
106106 : resolvedRoot;
107107 conversationDirs.add(yield* fileSystem.realPath(dir).pipe(Effect.orElseSucceed(() => dir)));
108108 }
109 const result = yield* Effect.promise(() =>
110 readAntigravityUsage([...conversationDirs], windowStartMs, cache),
109 const result = yield* readAntigravityUsage([...conversationDirs], windowStartMs, cache).pipe(
110 Effect.provideService(FileSystem.FileSystem, fileSystem),
111 Effect.provideService(Path.Path, path),
111112 );
112113 const scanned: ProviderUsageScan[] = [];
113114 for (const dir of conversationDirs) {
apps/server/src/provider/Drivers/antigravityUsageReader.test.ts+368-311
......@@ -1,15 +1,43 @@
1// @effect-diagnostics nodeBuiltinImport:off - the reader under test opens real
2// SQLite databases, mirroring the reader's own deliberate node:sqlite usage.
1// @effect-diagnostics-next-line nodeBuiltinImport:off - Effect's File.Info has no ctime or sub-millisecond mtime, which the cache tests compare.
32import * as NodeFSP from "node:fs/promises";
4import * as NodeOS from "node:os";
5import * as NodePath from "node:path";
6import * as NodeSqlite from "node:sqlite";
73
8import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest";
4import * as NodeServices from "@effect/platform-node/NodeServices";
5import { assert, describe, it } from "@effect/vitest";
6import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
7import * as Context from "effect/Context";
8import * as Effect from "effect/Effect";
9import * as FileSystem from "effect/FileSystem";
10import * as Layer from "effect/Layer";
11import * as Path from "effect/Path";
12import * as SqlClient from "effect/sql/SqlClient";
913
1014import { makeAntigravityUsageCache, readAntigravityUsage } from "./antigravityUsageReader.ts";
1115
12let dir: string;
16/** A writable connection that stays open until the enclosing scope closes. */
17const openDatabase = Effect.fn("openDatabase")(function* (filename: string) {
18 const context = yield* Layer.build(NodeSqliteClient.layer({ filename }));
19 const sql = Context.get(context, SqlClient.SqlClient);
20 return {
21 exec: (statement: string) => sql.unsafe(statement),
22 run: (statement: string, params: ReadonlyArray<unknown>) => sql.unsafe(statement, params),
23 };
24});
25
26/** Seeds `filename` and closes it, as a provider that has finished writing. */
27const seedDatabase = <E>(
28 filename: string,
29 seed: (db: Effect.Success<ReturnType<typeof openDatabase>>) => Effect.Effect<void, E>,
30) =>
31 Effect.scoped(
32 Effect.gen(function* () {
33 yield* seed(yield* openDatabase(filename));
34 }),
35 );
36
37const tempDir = Effect.gen(function* () {
38 const fileSystem = yield* FileSystem.FileSystem;
39 return yield* fileSystem.makeTempDirectoryScoped({ prefix: "usage-reader-test-" });
40});
1341
1442function protoNumber(field: number, value: number): number[] {
1543 const varint = (number: number) => {
......@@ -34,224 +62,241 @@ function protoText(field: number, value: string): number[] {
3462 return protoBytes(field, [...Buffer.from(value)]);
3563}
3664
37beforeEach(async () => {
38 dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "usage-reader-test-"));
39});
65const TABLES =
66 "CREATE TABLE gen_metadata (idx INTEGER, data BLOB); CREATE TABLE steps (idx INTEGER, metadata BLOB)";
4067
41afterEach(async () => {
42 await NodeFSP.rm(dir, { recursive: true, force: true });
43});
68const createTables = (db: Effect.Success<ReturnType<typeof openDatabase>>) =>
69 Effect.forEach(TABLES.split("; "), db.exec, { discard: true });
4470
45describe("readAntigravityUsage", () => {
46 it("deduplicates Antigravity generation and step usage while preserving retry model and token buckets", async () => {
47 const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "session-1.db"));
48 const stamp = protoNumber(1, 1780000000);
49 const usage = [
50 ...protoNumber(2, 100),
51 ...protoNumber(3, 40),
52 ...protoNumber(4, 5),
53 ...protoNumber(5, 20),
54 ...protoNumber(9, 10),
55 ...protoText(11, "response-1"),
56 ];
57 const retry = [
58 ...protoNumber(1, 1026),
59 ...protoNumber(2, 12),
60 ...protoNumber(3, 3),
61 ...protoText(11, "retry-1"),
62 ];
63 const generation = protoBytes(1, [
64 ...protoBytes(4, usage),
65 ...protoText(19, "Gemini 3 Pro"),
66 ...protoBytes(9, protoBytes(4, stamp)),
67 ]);
68 const step = [
69 ...protoBytes(9, usage),
70 ...protoBytes(8, stamp),
71 ...protoBytes(28, protoBytes(2, retry)),
72 ];
73 try {
74 db.exec(
75 "CREATE TABLE gen_metadata (idx INTEGER, data BLOB); CREATE TABLE steps (idx INTEGER, metadata BLOB)",
76 );
77 db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run(0, new Uint8Array(generation));
78 db.prepare("INSERT INTO steps VALUES (?, ?)").run(0, new Uint8Array(step));
79 } finally {
80 db.close();
81 }
82 const result = await readAntigravityUsage(dir, 0);
83 assert.deepStrictEqual(result.errors, []);
84 const records = result.files.flatMap((file) => file.records);
85 assert.strictEqual(records.length, 2);
86 const main = records.find((record) => record.model === "gemini-3-pro");
87 assert.isDefined(main);
88 assert.strictEqual(main?.timestampMs, 1780000000000);
89 assert.strictEqual(main?.sessionId, "session-1");
90 assert.deepStrictEqual(main?.totals, {
91 uncachedInputTokens: 100,
92 cachedInputTokens: 20,
93 cacheCreationTokens: 5,
94 outputTokens: 40,
95 reasoningTokens: 10,
96 });
97 assert.strictEqual(
98 records.find((record) => record.model === "claude-opus-4-6")?.totals.uncachedInputTokens,
99 12,
100 );
101 assert.deepStrictEqual(
102 (await readAntigravityUsage(dir, 1780000000001)).files.flatMap((file) => file.records),
103 [],
104 );
105 });
71const records = (result: Effect.Success<ReturnType<typeof readAntigravityUsage>>) =>
72 result.files.flatMap((file) => file.records);
10673
107 it("uses the matching Antigravity generation model for each model-less step", async () => {
108 const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "model-switch.db"));
109 try {
110 db.exec(
111 "CREATE TABLE gen_metadata (idx INTEGER, data BLOB); CREATE TABLE steps (idx INTEGER, metadata BLOB)",
112 );
113 const generation = db.prepare("INSERT INTO gen_metadata VALUES (?, ?)");
114 const step = db.prepare("INSERT INTO steps VALUES (?, ?)");
115 for (const [idx, name] of ["Gemini 3 Pro", "Claude Opus 4.6"].entries()) {
116 generation.run(idx, new Uint8Array(protoBytes(1, protoText(19, name))));
117 step.run(idx, new Uint8Array(protoBytes(9, protoNumber(2, 10 + idx))));
118 }
119 } finally {
120 db.close();
121 }
122 const result = await readAntigravityUsage(dir, 0);
123 assert.deepStrictEqual(result.errors, []);
124 assert.deepStrictEqual(
125 result.files.flatMap((file) => file.records).map((record) => record.model),
126 ["gemini-3-pro", "claude-opus-4-6"],
127 );
128 });
74describe("readAntigravityUsage", () => {
75 it.effect(
76 "deduplicates Antigravity generation and step usage while preserving retry model and token buckets",
77 () =>
78 Effect.gen(function* () {
79 const path = yield* Path.Path;
80 const dir = yield* tempDir;
81 const stamp = protoNumber(1, 1780000000);
82 const usage = [
83 ...protoNumber(2, 100),
84 ...protoNumber(3, 40),
85 ...protoNumber(4, 5),
86 ...protoNumber(5, 20),
87 ...protoNumber(9, 10),
88 ...protoText(11, "response-1"),
89 ];
90 const retry = [
91 ...protoNumber(1, 1026),
92 ...protoNumber(2, 12),
93 ...protoNumber(3, 3),
94 ...protoText(11, "retry-1"),
95 ];
96 const generation = protoBytes(1, [
97 ...protoBytes(4, usage),
98 ...protoText(19, "Gemini 3 Pro"),
99 ...protoBytes(9, protoBytes(4, stamp)),
100 ]);
101 const step = [
102 ...protoBytes(9, usage),
103 ...protoBytes(8, stamp),
104 ...protoBytes(28, protoBytes(2, retry)),
105 ];
106 yield* seedDatabase(path.join(dir, "session-1.db"), (db) =>
107 Effect.gen(function* () {
108 yield* createTables(db);
109 yield* db.run("INSERT INTO gen_metadata VALUES (?, ?)", [
110 0,
111 new Uint8Array(generation),
112 ]);
113 yield* db.run("INSERT INTO steps VALUES (?, ?)", [0, new Uint8Array(step)]);
114 }),
115 );
116 const result = yield* readAntigravityUsage(dir, 0);
117 assert.deepStrictEqual(result.errors, []);
118 const read = records(result);
119 assert.strictEqual(read.length, 2);
120 const main = read.find((record) => record.model === "gemini-3-pro");
121 assert.isDefined(main);
122 assert.strictEqual(main?.timestampMs, 1780000000000);
123 assert.strictEqual(main?.sessionId, "session-1");
124 assert.deepStrictEqual(main?.totals, {
125 uncachedInputTokens: 100,
126 cachedInputTokens: 20,
127 cacheCreationTokens: 5,
128 outputTokens: 40,
129 reasoningTokens: 10,
130 });
131 assert.strictEqual(
132 read.find((record) => record.model === "claude-opus-4-6")?.totals.uncachedInputTokens,
133 12,
134 );
135 assert.deepStrictEqual(records(yield* readAntigravityUsage(dir, 1780000000001)), []);
136 }).pipe(Effect.provide(NodeServices.layer)),
137 );
129138
130 it("merges Antigravity aliases that bridge previously separate step records", async () => {
131 const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "bridge.db"));
132 try {
133 db.exec(
134 "CREATE TABLE gen_metadata (idx INTEGER, data BLOB); CREATE TABLE steps (idx INTEGER, metadata BLOB)",
139 it.effect("uses the matching Antigravity generation model for each model-less step", () =>
140 Effect.gen(function* () {
141 const path = yield* Path.Path;
142 const dir = yield* tempDir;
143 yield* seedDatabase(path.join(dir, "model-switch.db"), (db) =>
144 Effect.gen(function* () {
145 yield* createTables(db);
146 for (const [idx, name] of ["Gemini 3 Pro", "Claude Opus 4.6"].entries()) {
147 yield* db.run("INSERT INTO gen_metadata VALUES (?, ?)", [
148 idx,
149 new Uint8Array(protoBytes(1, protoText(19, name))),
150 ]);
151 yield* db.run("INSERT INTO steps VALUES (?, ?)", [
152 idx,
153 new Uint8Array(protoBytes(9, protoNumber(2, 10 + idx))),
154 ]);
155 }
156 }),
135157 );
136 const step = db.prepare("INSERT INTO steps VALUES (?, ?)");
137 step.run(
138 0,
139 new Uint8Array(protoBytes(9, [...protoNumber(2, 100), ...protoText(11, "response")])),
140 );
141 step.run(
142 1,
143 new Uint8Array(protoBytes(9, [...protoNumber(3, 40), ...protoText(12, "provider")])),
144 );
145 db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run(
146 0,
147 new Uint8Array(
148 protoBytes(1, [
149 ...protoText(19, "Gemini 3 Pro"),
150 ...protoBytes(4, [
151 ...protoNumber(2, 50),
152 ...protoNumber(5, 20),
153 ...protoText(11, "response"),
154 ...protoText(12, "provider"),
155 ]),
156 ]),
157 ),
158 const result = yield* readAntigravityUsage(dir, 0);
159 assert.deepStrictEqual(result.errors, []);
160 assert.deepStrictEqual(
161 records(result).map((record) => record.model),
162 ["gemini-3-pro", "claude-opus-4-6"],
158163 );
159 } finally {
160 db.close();
161 }
162 const result = await readAntigravityUsage(dir, 0);
163 assert.deepStrictEqual(result.errors, []);
164 const records = result.files.flatMap((file) => file.records);
165 assert.strictEqual(records.length, 1);
166 assert.deepStrictEqual(records[0]?.totals, {
167 uncachedInputTokens: 100,
168 cachedInputTokens: 20,
169 cacheCreationTokens: 0,
170 outputTokens: 40,
171 reasoningTokens: 0,
172 });
173 });
164 }).pipe(Effect.provide(NodeServices.layer)),
165 );
174166
175 it("merges Antigravity provider and message aliases across configured roots while keeping original ownership", async () => {
176 const roots = [NodePath.join(dir, "first"), NodePath.join(dir, "second")];
177 for (const [index, root] of roots.entries()) {
178 await NodeFSP.mkdir(root);
179 const db = new NodeSqlite.DatabaseSync(NodePath.join(root, `session-${index}.db`));
180 try {
181 db.exec("CREATE TABLE steps (idx INTEGER, metadata BLOB)");
182 for (const identity of [7, 12]) {
183 const usage = [
184 ...protoNumber(1, 246),
185 ...protoNumber(2, index === 0 ? 100 : 150),
186 ...protoText(11, `response-${index}-${identity}`),
187 ...protoText(identity, `shared-${identity}`),
188 ];
189 db.prepare("INSERT INTO steps VALUES (?, ?)").run(
190 identity,
191 new Uint8Array(protoBytes(9, usage)),
192 );
193 }
194 } finally {
195 db.close();
196 }
197 }
198 const result = await readAntigravityUsage(roots, 0);
199 assert.deepStrictEqual(result.errors, []);
200 assert.strictEqual(result.files.length, 2);
201 assert.strictEqual(result.files[0]?.root, roots[0]);
202 assert.strictEqual(result.files[0]?.records.length, 2);
203 assert.strictEqual(result.files[1]?.records.length, 0);
204 assert.deepStrictEqual(
205 result.files[0]?.records.map((record) => record.totals.uncachedInputTokens),
206 [150, 150],
207 );
208 assert.isTrue(result.files[0]?.records.every((record) => record.sessionId === "session-0"));
209 });
210
211 it("upgrades Antigravity fallback timestamps before applying the date window", async () => {
212 for (const fallback of ["mtime", "trajectory"]) {
213 const path = NodePath.join(dir, `${fallback}.db`);
214 const db = new NodeSqlite.DatabaseSync(path);
215 try {
216 db.exec(
217 "CREATE TABLE gen_metadata (idx INTEGER, data BLOB); CREATE TABLE steps (idx INTEGER, metadata BLOB)",
218 );
219 if (fallback === "trajectory") {
220 db.exec("CREATE TABLE trajectory_metadata_blob (data BLOB)");
221 db.prepare("INSERT INTO trajectory_metadata_blob VALUES (?)").run(
222 new Uint8Array(protoBytes(2, protoNumber(1, 1780000200))),
223 );
224 }
225 for (const [index, seconds] of [1780000000, 1780000200].entries()) {
226 const usage = [...protoNumber(2, 10), ...protoText(11, `${fallback}-${index}`)];
227 db.prepare("INSERT INTO steps VALUES (?, ?)").run(
228 index,
229 new Uint8Array(protoBytes(9, usage)),
230 );
231 db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run(
232 index,
167 it.effect("merges Antigravity aliases that bridge previously separate step records", () =>
168 Effect.gen(function* () {
169 const path = yield* Path.Path;
170 const dir = yield* tempDir;
171 yield* seedDatabase(path.join(dir, "bridge.db"), (db) =>
172 Effect.gen(function* () {
173 yield* createTables(db);
174 yield* db.run("INSERT INTO steps VALUES (?, ?)", [
175 0,
176 new Uint8Array(protoBytes(9, [...protoNumber(2, 100), ...protoText(11, "response")])),
177 ]);
178 yield* db.run("INSERT INTO steps VALUES (?, ?)", [
179 1,
180 new Uint8Array(protoBytes(9, [...protoNumber(3, 40), ...protoText(12, "provider")])),
181 ]);
182 yield* db.run("INSERT INTO gen_metadata VALUES (?, ?)", [
183 0,
233184 new Uint8Array(
234185 protoBytes(1, [
235 ...protoBytes(4, usage),
236 ...protoBytes(9, protoBytes(4, protoNumber(1, seconds))),
186 ...protoText(19, "Gemini 3 Pro"),
187 ...protoBytes(4, [
188 ...protoNumber(2, 50),
189 ...protoNumber(5, 20),
190 ...protoText(11, "response"),
191 ...protoText(12, "provider"),
192 ]),
237193 ]),
238194 ),
195 ]);
196 }),
197 );
198 const result = yield* readAntigravityUsage(dir, 0);
199 assert.deepStrictEqual(result.errors, []);
200 const read = records(result);
201 assert.strictEqual(read.length, 1);
202 assert.deepStrictEqual(read[0]?.totals, {
203 uncachedInputTokens: 100,
204 cachedInputTokens: 20,
205 cacheCreationTokens: 0,
206 outputTokens: 40,
207 reasoningTokens: 0,
208 });
209 }).pipe(Effect.provide(NodeServices.layer)),
210 );
211
212 it.effect(
213 "merges Antigravity provider and message aliases across configured roots while keeping original ownership",
214 () =>
215 Effect.gen(function* () {
216 const fileSystem = yield* FileSystem.FileSystem;
217 const path = yield* Path.Path;
218 const dir = yield* tempDir;
219 const roots = [path.join(dir, "first"), path.join(dir, "second")];
220 for (const [index, root] of roots.entries()) {
221 yield* fileSystem.makeDirectory(root);
222 yield* seedDatabase(path.join(root, `session-${index}.db`), (db) =>
223 Effect.gen(function* () {
224 yield* db.exec("CREATE TABLE steps (idx INTEGER, metadata BLOB)");
225 for (const identity of [7, 12]) {
226 const usage = [
227 ...protoNumber(1, 246),
228 ...protoNumber(2, index === 0 ? 100 : 150),
229 ...protoText(11, `response-${index}-${identity}`),
230 ...protoText(identity, `shared-${identity}`),
231 ];
232 yield* db.run("INSERT INTO steps VALUES (?, ?)", [
233 identity,
234 new Uint8Array(protoBytes(9, usage)),
235 ]);
236 }
237 }),
239238 );
240239 }
241 } finally {
242 db.close();
240 const result = yield* readAntigravityUsage(roots, 0);
241 assert.deepStrictEqual(result.errors, []);
242 assert.strictEqual(result.files.length, 2);
243 assert.strictEqual(result.files[0]?.root, roots[0]);
244 assert.strictEqual(result.files[0]?.records.length, 2);
245 assert.strictEqual(result.files[1]?.records.length, 0);
246 assert.deepStrictEqual(
247 result.files[0]?.records.map((record) => record.totals.uncachedInputTokens),
248 [150, 150],
249 );
250 assert.isTrue(result.files[0]?.records.every((record) => record.sessionId === "session-0"));
251 }).pipe(Effect.provide(NodeServices.layer)),
252 );
253
254 it.effect("upgrades Antigravity fallback timestamps before applying the date window", () =>
255 Effect.gen(function* () {
256 const fileSystem = yield* FileSystem.FileSystem;
257 const path = yield* Path.Path;
258 const dir = yield* tempDir;
259 for (const fallback of ["mtime", "trajectory"]) {
260 const file = path.join(dir, `${fallback}.db`);
261 yield* seedDatabase(file, (db) =>
262 Effect.gen(function* () {
263 yield* createTables(db);
264 if (fallback === "trajectory") {
265 yield* db.exec("CREATE TABLE trajectory_metadata_blob (data BLOB)");
266 yield* db.run("INSERT INTO trajectory_metadata_blob VALUES (?)", [
267 new Uint8Array(protoBytes(2, protoNumber(1, 1780000200))),
268 ]);
269 }
270 for (const [index, seconds] of [1780000000, 1780000200].entries()) {
271 const usage = [...protoNumber(2, 10), ...protoText(11, `${fallback}-${index}`)];
272 yield* db.run("INSERT INTO steps VALUES (?, ?)", [
273 index,
274 new Uint8Array(protoBytes(9, usage)),
275 ]);
276 yield* db.run("INSERT INTO gen_metadata VALUES (?, ?)", [
277 index,
278 new Uint8Array(
279 protoBytes(1, [
280 ...protoBytes(4, usage),
281 ...protoBytes(9, protoBytes(4, protoNumber(1, seconds))),
282 ]),
283 ),
284 ]);
285 }
286 }),
287 );
288 yield* fileSystem.utimes(file, 1780000000, 1780000000);
243289 }
244 await NodeFSP.utimes(path, 1780000000, 1780000000);
245 }
246 const result = await readAntigravityUsage(dir, 1780000100000);
247 assert.deepStrictEqual(result.errors, []);
248 const records = result.files.flatMap((file) => file.records);
249 assert.strictEqual(records.length, 2);
250 assert.deepStrictEqual(
251 records.map((record) => record.timestampMs),
252 [1780000200000, 1780000200000],
253 );
254 });
290 const result = yield* readAntigravityUsage(dir, 1780000100000);
291 assert.deepStrictEqual(result.errors, []);
292 const read = records(result);
293 assert.strictEqual(read.length, 2);
294 assert.deepStrictEqual(
295 read.map((record) => record.timestampMs),
296 [1780000200000, 1780000200000],
297 );
298 }).pipe(Effect.provide(NodeServices.layer)),
299 );
255300
256301 const antigravityGeneration = (responseId: string) =>
257302 new Uint8Array(
......@@ -266,126 +311,138 @@ describe("readAntigravityUsage", () => {
266311 ]),
267312 );
268313
269 it("reuses an unchanged Antigravity database instead of decoding it again", async () => {
270 const path = NodePath.join(dir, "session-1.db");
271 const db = new NodeSqlite.DatabaseSync(path);
272 try {
273 db.exec("CREATE TABLE gen_metadata (idx INTEGER, data BLOB)");
274 db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run(0, antigravityGeneration("r-1"));
314 it.effect("reuses an unchanged Antigravity database instead of decoding it again", () =>
315 Effect.gen(function* () {
316 const path = yield* Path.Path;
317 const dir = yield* tempDir;
318 const db = yield* openDatabase(path.join(dir, "session-1.db"));
319 yield* db.exec("CREATE TABLE gen_metadata (idx INTEGER, data BLOB)");
320 yield* db.run("INSERT INTO gen_metadata VALUES (?, ?)", [0, antigravityGeneration("r-1")]);
275321 const cache = makeAntigravityUsageCache();
276 assert.deepStrictEqual((await readAntigravityUsage(dir, 0, cache)).errors, []);
322 assert.deepStrictEqual((yield* readAntigravityUsage(dir, 0, cache)).errors, []);
277323
278324 // An exclusive lock makes a fresh read fail without touching the file, so
279325 // only a cache hit can still return the earlier records.
280 db.exec("BEGIN EXCLUSIVE");
281 assert.strictEqual((await readAntigravityUsage(dir, 0)).errors.length, 1);
282 const cached = await readAntigravityUsage(dir, 0, cache);
326 yield* db.exec("BEGIN EXCLUSIVE");
327 assert.strictEqual((yield* readAntigravityUsage(dir, 0)).errors.length, 1);
328 const cached = yield* readAntigravityUsage(dir, 0, cache);
283329 assert.deepStrictEqual(cached.errors, []);
284330 assert.deepStrictEqual(
285 cached.files.flatMap((file) => file.records).map((record) => record.dedupeKey),
331 records(cached).map((record) => record.dedupeKey),
286332 ["antigravity:11:r-1"],
287333 );
288 db.exec("ROLLBACK");
289 } finally {
290 db.close();
291 }
292 });
334 yield* db.exec("ROLLBACK");
335 }).pipe(Effect.provide(NodeServices.layer)),
336 );
293337
294 it("rereads an Antigravity database rewritten with its size and mtime restored", async () => {
295 const path = NodePath.join(dir, "session-1.db");
296 const db = new NodeSqlite.DatabaseSync(path);
297 try {
298 db.exec("CREATE TABLE gen_metadata (idx INTEGER, data BLOB)");
299 db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run(0, antigravityGeneration("r-1"));
300 } finally {
301 db.close();
302 }
303 await NodeFSP.utimes(path, 1780000000, 1780000000);
304 const cache = makeAntigravityUsageCache();
305 assert.deepStrictEqual((await readAntigravityUsage(dir, 0, cache)).errors, []);
338 it.effect("rereads an Antigravity database rewritten with its size and mtime restored", () =>
339 Effect.gen(function* () {
340 const fileSystem = yield* FileSystem.FileSystem;
341 const pathService = yield* Path.Path;
342 const dir = yield* tempDir;
343 const path = pathService.join(dir, "session-1.db");
344 yield* seedDatabase(path, (db) =>
345 Effect.gen(function* () {
346 yield* db.exec("CREATE TABLE gen_metadata (idx INTEGER, data BLOB)");
347 yield* db.run("INSERT INTO gen_metadata VALUES (?, ?)", [
348 0,
349 antigravityGeneration("r-1"),
350 ]);
351 }),
352 );
353 yield* fileSystem.utimes(path, 1780000000, 1780000000);
354 const cache = makeAntigravityUsageCache();
355 assert.deepStrictEqual((yield* readAntigravityUsage(dir, 0, cache)).errors, []);
306356
307 // ctime has the kernel's timestamp granularity, which can be a few
308 // milliseconds, so repeat the forged rewrite until it lands on a later tick
309 // than the cached read, as any real rewrite does.
310 const cached = await NodeFSP.stat(path);
311 do {
312 await NodeFSP.writeFile(path, Buffer.alloc(cached.size));
313 await NodeFSP.utimes(path, 1780000000, 1780000000);
314 } while ((await NodeFSP.stat(path)).ctimeMs === cached.ctimeMs);
315 const restored = await NodeFSP.stat(path);
316 assert.strictEqual(restored.size, cached.size);
317 assert.strictEqual(restored.mtimeMs, cached.mtimeMs);
318 const next = await readAntigravityUsage(dir, 0, cache);
319 assert.deepStrictEqual(next.errors, [path]);
320 assert.deepStrictEqual(
321 next.files.flatMap((file) => file.records),
322 [],
323 );
324 });
357 // ctime has the kernel's timestamp granularity, which can be a few
358 // milliseconds, so repeat the forged rewrite until it lands on a later tick
359 // than the cached read, as any real rewrite does.
360 const stat = () => Effect.promise(() => NodeFSP.stat(path));
361 const cached = yield* stat();
362 do {
363 yield* fileSystem.writeFile(path, new Uint8Array(cached.size));
364 yield* fileSystem.utimes(path, 1780000000, 1780000000);
365 } while ((yield* stat()).ctimeMs === cached.ctimeMs);
366 const restored = yield* stat();
367 assert.strictEqual(restored.size, cached.size);
368 assert.strictEqual(restored.mtimeMs, cached.mtimeMs);
369 const next = yield* readAntigravityUsage(dir, 0, cache);
370 assert.deepStrictEqual(next.errors, [path]);
371 assert.deepStrictEqual(records(next), []);
372 }).pipe(Effect.provide(NodeServices.layer)),
373 );
325374
326 it("rereads an Antigravity database when only its WAL changed", async () => {
327 const path = NodePath.join(dir, "session-1.db");
328 const db = new NodeSqlite.DatabaseSync(path);
329 try {
330 db.exec(
331 "PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0; CREATE TABLE gen_metadata (idx INTEGER, data BLOB)",
332 );
333 const insert = db.prepare("INSERT INTO gen_metadata VALUES (?, ?)");
334 insert.run(0, antigravityGeneration("r-1"));
375 it.effect("rereads an Antigravity database when only its WAL changed", () =>
376 Effect.gen(function* () {
377 const pathService = yield* Path.Path;
378 const dir = yield* tempDir;
379 const path = pathService.join(dir, "session-1.db");
380 const db = yield* openDatabase(path);
381 yield* db.exec("PRAGMA journal_mode = WAL");
382 yield* db.exec("PRAGMA wal_autocheckpoint = 0");
383 yield* db.exec("CREATE TABLE gen_metadata (idx INTEGER, data BLOB)");
384 const insert = (idx: number, responseId: string) =>
385 db.run("INSERT INTO gen_metadata VALUES (?, ?)", [idx, antigravityGeneration(responseId)]);
386 yield* insert(0, "r-1");
335387 const cache = makeAntigravityUsageCache();
336 const first = await readAntigravityUsage(dir, 0, cache);
337 assert.strictEqual(first.files.flatMap((file) => file.records).length, 1);
388 const first = yield* readAntigravityUsage(dir, 0, cache);
389 assert.strictEqual(records(first).length, 1);
338390
339 const before = await NodeFSP.stat(path);
340 insert.run(1, antigravityGeneration("r-2"));
341 const after = await NodeFSP.stat(path);
391 const before = yield* Effect.promise(() => NodeFSP.stat(path));
392 yield* insert(1, "r-2");
393 const after = yield* Effect.promise(() => NodeFSP.stat(path));
342394 assert.strictEqual(after.size, before.size);
343395 assert.strictEqual(after.mtimeMs, before.mtimeMs);
344396
345 const next = await readAntigravityUsage(dir, 0, cache);
397 const next = yield* readAntigravityUsage(dir, 0, cache);
346398 assert.deepStrictEqual(
347 next.files.flatMap((file) => file.records).map((record) => record.dedupeKey),
399 records(next).map((record) => record.dedupeKey),
348400 ["antigravity:11:r-1", "antigravity:11:r-2"],
349401 );
350 } finally {
351 db.close();
352 }
353 });
402 }).pipe(Effect.provide(NodeServices.layer)),
403 );
354404
355 it("reads Antigravity step-only stores and reports malformed databases", async () => {
356 const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "steps.db"));
357 try {
358 db.exec("CREATE TABLE steps (idx INTEGER, metadata BLOB)");
359 const usage = [...protoNumber(1, 246), ...protoNumber(2, 10), ...protoNumber(3, 5)];
360 db.prepare("INSERT INTO steps VALUES (?, ?)").run(
361 0,
362 new Uint8Array([...protoBytes(9, usage), ...protoBytes(8, protoNumber(1, 1780000000))]),
405 it.effect("reads Antigravity step-only stores and reports malformed databases", () =>
406 Effect.gen(function* () {
407 const fileSystem = yield* FileSystem.FileSystem;
408 const path = yield* Path.Path;
409 const dir = yield* tempDir;
410 yield* seedDatabase(path.join(dir, "steps.db"), (db) =>
411 Effect.gen(function* () {
412 yield* db.exec("CREATE TABLE steps (idx INTEGER, metadata BLOB)");
413 const usage = [...protoNumber(1, 246), ...protoNumber(2, 10), ...protoNumber(3, 5)];
414 yield* db.run("INSERT INTO steps VALUES (?, ?)", [
415 0,
416 new Uint8Array([...protoBytes(9, usage), ...protoBytes(8, protoNumber(1, 1780000000))]),
417 ]);
418 }),
363419 );
364 } finally {
365 db.close();
366 }
367 await NodeFSP.writeFile(NodePath.join(dir, "broken.db"), "not a sqlite database");
368 const result = await readAntigravityUsage(dir, 0);
369 assert.strictEqual(result.errors.length, 1);
370 assert.strictEqual(result.files.flatMap((file) => file.records)[0]?.model, "gemini-2.5-pro");
371 assert.strictEqual(result.files.flatMap((file) => file.records)[0]?.totals.outputTokens, 5);
372 });
420 yield* fileSystem.writeFileString(path.join(dir, "broken.db"), "not a sqlite database");
421 const result = yield* readAntigravityUsage(dir, 0);
422 assert.strictEqual(result.errors.length, 1);
423 assert.strictEqual(records(result)[0]?.model, "gemini-2.5-pro");
424 assert.strictEqual(records(result)[0]?.totals.outputTokens, 5);
425 }).pipe(Effect.provide(NodeServices.layer)),
426 );
373427
374 it("ignores large values in unused Antigravity protobuf fields", async () => {
375 const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "large-varint.db"));
376 try {
377 db.exec("CREATE TABLE steps (idx INTEGER, metadata BLOB)");
378 const unusedField = [...protoNumber(99, 0).slice(0, -1), ...Array(9).fill(0xff), 0x01];
379 const usage = [...protoNumber(1, 246), ...protoNumber(2, 10), ...unusedField];
380 db.prepare("INSERT INTO steps VALUES (?, ?)").run(0, new Uint8Array(protoBytes(9, usage)));
381 } finally {
382 db.close();
383 }
384 const result = await readAntigravityUsage(dir, 0);
385 assert.deepStrictEqual(result.errors, []);
386 assert.strictEqual(
387 result.files.flatMap((file) => file.records)[0]?.totals.uncachedInputTokens,
388 10,
389 );
390 });
428 it.effect("ignores large values in unused Antigravity protobuf fields", () =>
429 Effect.gen(function* () {
430 const path = yield* Path.Path;
431 const dir = yield* tempDir;
432 yield* seedDatabase(path.join(dir, "large-varint.db"), (db) =>
433 Effect.gen(function* () {
434 yield* db.exec("CREATE TABLE steps (idx INTEGER, metadata BLOB)");
435 const unusedField = [...protoNumber(99, 0).slice(0, -1), ...Array(9).fill(0xff), 0x01];
436 const usage = [...protoNumber(1, 246), ...protoNumber(2, 10), ...unusedField];
437 yield* db.run("INSERT INTO steps VALUES (?, ?)", [
438 0,
439 new Uint8Array(protoBytes(9, usage)),
440 ]);
441 }),
442 );
443 const result = yield* readAntigravityUsage(dir, 0);
444 assert.deepStrictEqual(result.errors, []);
445 assert.strictEqual(records(result)[0]?.totals.uncachedInputTokens, 10);
446 }).pipe(Effect.provide(NodeServices.layer)),
447 );
391448});
apps/server/src/provider/Drivers/antigravityUsageReader.ts+176-119
......@@ -1,11 +1,14 @@
1// node:sqlite reads live conversation databases while Node fs discovers them.
2// @effect-diagnostics nodeBuiltinImport:off
1// @effect-diagnostics-next-line nodeBuiltinImport:off - Effect's File.Info has no ctime, which the cache fingerprint needs.
32import * as NodeFSP from "node:fs/promises";
4import * as NodePath from "node:path";
5import * as NodeSqlite from "node:sqlite";
6import * as NodeTimersPromises from "node:timers/promises";
73
84import type { UsageRecord } from "@t3tools/provider-core/server/usage";
5import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
6import * as Effect from "effect/Effect";
7import * as Exit from "effect/Exit";
8import * as FileSystem from "effect/FileSystem";
9import * as Path from "effect/Path";
10import * as Schema from "effect/Schema";
11import * as SqlClient from "effect/sql/SqlClient";
912
1013type FieldValue = number | bigint | Uint8Array;
1114type Fields = Map<number, FieldValue[]>;
......@@ -158,103 +161,136 @@ interface UsageCandidate {
158161 timestampQuality: number;
159162}
160163
161async function readDatabase(path: string, fallbackTimestamp: number): Promise<UsageCandidate[]> {
162 const db = new NodeSqlite.DatabaseSync(path, { readOnly: true });
163 try {
164 db.exec("PRAGMA busy_timeout = 100; BEGIN");
164/** A conversation database whose usage tables are missing or undecodable. */
165class AntigravityDatabaseDecodeError extends Schema.TaggedError<AntigravityDatabaseDecodeError>()(
166 "AntigravityDatabaseDecodeError",
167 { path: Schema.String, cause: Schema.optional(Schema.Defect()) },
168) {
169 override get message(): string {
170 return `Antigravity usage could not be decoded from ${this.path}.`;
171 }
172}
173
174const readDatabase = Effect.fn("readDatabase")(function* (
175 path: string,
176 sessionId: string,
177 fallbackTimestamp: number,
178) {
179 const sql = yield* SqlClient.SqlClient;
180 const decode = <A>(evaluate: () => A) =>
181 Effect.try({
182 try: evaluate,
183 catch: (cause) => new AntigravityDatabaseDecodeError({ path, cause }),
184 });
185 yield* sql.unsafe("PRAGMA busy_timeout = 100");
186 return yield* Effect.gen(function* () {
165187 const tables = new Set(
166 db
167 .prepare("SELECT name FROM sqlite_master WHERE type = 'table'")
168 .all()
169 .map((row) => row.name),
188 (yield* sql.unsafe<{ readonly name: unknown }>(
189 "SELECT name FROM sqlite_master WHERE type = 'table'",
190 )).map((row) => row.name),
170191 );
171192 if (!tables.has("gen_metadata") && !tables.has("steps")) {
172 throw new Error("Missing Antigravity usage tables");
193 return yield* new AntigravityDatabaseDecodeError({ path });
173194 }
174 const readMetadata = async (query: string, column: string, step: boolean) => {
195 const readMetadata = Effect.fnUntraced(function* (
196 query: string,
197 column: string,
198 step: boolean,
199 ) {
175200 const entries: Array<{ idx: number; entry: Metadata }> = [];
176 for (const row of db.prepare(query).iterate()) {
177 if (typeof row.idx !== "number") throw new Error("Invalid Antigravity metadata index");
178 entries.push({ idx: row.idx, entry: metadata(blob(row[column]), step) });
179 if (entries.length % 256 === 0) await NodeTimersPromises.setImmediate();
201 for (const row of yield* sql.unsafe<Record<string, unknown>>(query)) {
202 const idx = row.idx;
203 if (typeof idx !== "number") {
204 return yield* new AntigravityDatabaseDecodeError({ path });
205 }
206 entries.push({ idx, entry: yield* decode(() => metadata(blob(row[column]), step)) });
207 if (entries.length % 256 === 0) yield* Effect.yieldNow;
180208 }
181209 return entries;
182 };
210 });
183211 const generations = tables.has("gen_metadata")
184 ? await readMetadata("SELECT idx, data FROM gen_metadata ORDER BY idx", "data", false)
212 ? yield* readMetadata("SELECT idx, data FROM gen_metadata ORDER BY idx", "data", false)
185213 : [];
186214 let trajectoryTimestamp: number | null = null;
187215 if (tables.has("trajectory_metadata_blob")) {
188 for (const row of db.prepare("SELECT data FROM trajectory_metadata_blob").iterate()) {
189 trajectoryTimestamp ??= timestamp(nested(fields(blob(row.data)), 2));
216 for (const row of yield* sql.unsafe<{ readonly data: unknown }>(
217 "SELECT data FROM trajectory_metadata_blob",
218 )) {
219 trajectoryTimestamp ??= yield* decode(() => timestamp(nested(fields(blob(row.data)), 2)));
190220 }
191221 }
192222 const steps = tables.has("steps")
193 ? await readMetadata(
223 ? yield* readMetadata(
194224 "SELECT idx, metadata FROM steps WHERE metadata IS NOT NULL ORDER BY idx",
195225 "metadata",
196226 true,
197227 )
198228 : [];
199 const sessionId = NodePath.basename(path, ".db");
200 const records: UsageCandidate[] = [];
201 const generationModels = new Map(generations.map(({ idx, entry }) => [idx, entry.model]));
202 for (const [source, entries] of [
203 ["step", steps],
204 ["generation", generations],
205 ] as const) {
206 for (const [index, { idx, entry }] of entries.entries()) {
207 for (const [usageIndex, usage] of entry.usages.entries()) {
208 const outputTokens = Math.max(
209 numberAt(usage, 3),
210 numberAt(usage, 9) + numberAt(usage, 10),
211 );
212 const totals = {
213 uncachedInputTokens: numberAt(usage, 2),
214 cachedInputTokens: numberAt(usage, 5),
215 cacheCreationTokens: numberAt(usage, 4),
216 outputTokens,
217 reasoningTokens: Math.min(outputTokens, numberAt(usage, 9)),
218 };
219 if (
220 totals.uncachedInputTokens +
221 totals.cachedInputTokens +
222 totals.cacheCreationTokens +
223 outputTokens ===
224 0
225 )
226 continue;
227 const keys = ([11, 12, 7] as const).flatMap((key) => {
228 const id = textAt(usage, key);
229 return id ? [`antigravity:${key}:${id}`] : [];
230 });
231 const record: UsageRecord = {
232 provider: "antigravity",
233 sessionId,
234 timestampMs: entry.timestampMs ?? trajectoryTimestamp ?? fallbackTimestamp,
235 model:
236 MODEL_IDS[numberAt(usage, 1)] ||
237 entry.model ||
238 (source === "step" ? generationModels.get(idx) : "") ||
239 modelName("", numberAt(usage, 1)) ||
240 "antigravity-unknown",
241 totals,
242 reportedCostUsd: null,
243 speed: "standard",
244 dedupeKey: keys[0] ?? `antigravity:${sessionId}:${source}:${index}:${usageIndex}`,
245 };
246 records.push({
247 record,
248 keys,
249 timestampQuality: entry.timestampMs !== null ? 2 : trajectoryTimestamp !== null ? 1 : 0,
250 });
251 }
229 return yield* decode(() =>
230 usageCandidates(sessionId, generations, steps, trajectoryTimestamp, fallbackTimestamp),
231 );
232 }).pipe(sql.withTransaction);
233});
234
235function usageCandidates(
236 sessionId: string,
237 generations: ReadonlyArray<{ idx: number; entry: Metadata }>,
238 steps: ReadonlyArray<{ idx: number; entry: Metadata }>,
239 trajectoryTimestamp: number | null,
240 fallbackTimestamp: number,
241): UsageCandidate[] {
242 const records: UsageCandidate[] = [];
243 const generationModels = new Map(generations.map(({ idx, entry }) => [idx, entry.model]));
244 for (const [source, entries] of [
245 ["step", steps],
246 ["generation", generations],
247 ] as const) {
248 for (const [index, { idx, entry }] of entries.entries()) {
249 for (const [usageIndex, usage] of entry.usages.entries()) {
250 const outputTokens = Math.max(numberAt(usage, 3), numberAt(usage, 9) + numberAt(usage, 10));
251 const totals = {
252 uncachedInputTokens: numberAt(usage, 2),
253 cachedInputTokens: numberAt(usage, 5),
254 cacheCreationTokens: numberAt(usage, 4),
255 outputTokens,
256 reasoningTokens: Math.min(outputTokens, numberAt(usage, 9)),
257 };
258 if (
259 totals.uncachedInputTokens +
260 totals.cachedInputTokens +
261 totals.cacheCreationTokens +
262 outputTokens ===
263 0
264 )
265 continue;
266 const keys = ([11, 12, 7] as const).flatMap((key) => {
267 const id = textAt(usage, key);
268 return id ? [`antigravity:${key}:${id}`] : [];
269 });
270 const record: UsageRecord = {
271 provider: "antigravity",
272 sessionId,
273 timestampMs: entry.timestampMs ?? trajectoryTimestamp ?? fallbackTimestamp,
274 model:
275 MODEL_IDS[numberAt(usage, 1)] ||
276 entry.model ||
277 (source === "step" ? generationModels.get(idx) : "") ||
278 modelName("", numberAt(usage, 1)) ||
279 "antigravity-unknown",
280 totals,
281 reportedCostUsd: null,
282 speed: "standard",
283 dedupeKey: keys[0] ?? `antigravity:${sessionId}:${source}:${index}:${usageIndex}`,
284 };
285 records.push({
286 record,
287 keys,
288 timestampQuality: entry.timestampMs !== null ? 2 : trajectoryTimestamp !== null ? 1 : 0,
289 });
252290 }
253291 }
254 return records;
255 } finally {
256 db.close();
257292 }
293 return records;
258294}
259295
260296interface CachedDatabase {
......@@ -275,11 +311,13 @@ export const makeAntigravityUsageCache = () => new Map<string, CachedDatabase>()
275311 * filtering. With a cache, databases unchanged since the previous read are not
276312 * decoded again.
277313 */
278export async function readAntigravityUsage(
314export const readAntigravityUsage = Effect.fn("readAntigravityUsage")(function* (
279315 conversationsDirectories: string | readonly string[],
280316 sinceMs: number,
281317 cache?: Map<string, CachedDatabase>,
282318) {
319 const fileSystem = yield* FileSystem.FileSystem;
320 const pathService = yield* Path.Path;
283321 const roots =
284322 typeof conversationsDirectories === "string"
285323 ? [conversationsDirectories]
......@@ -350,50 +388,69 @@ export async function readAntigravityUsage(
350388 }
351389 };
352390 const visited = new Set<string>();
353 const walk = async (directory: string, root: string): Promise<void> => {
354 let entries;
355 try {
356 entries = await NodeFSP.readdir(directory, { withFileTypes: true });
357 } catch (error) {
358 if ((error as NodeJS.ErrnoException).code !== "ENOENT") errors.push(directory);
359 return;
360 }
361 entries.sort((a, b) => a.name.localeCompare(b.name));
362 for (const entry of entries) {
363 const path = NodePath.join(directory, entry.name);
364 if (entry.isDirectory()) {
365 await walk(path, root);
366 } else if (entry.isFile() && entry.name.endsWith(".db")) {
367 try {
368 const canonical = await NodeFSP.realpath(path);
369 if (visited.has(canonical)) continue;
370 visited.add(canonical);
371 const stat = await NodeFSP.stat(path);
372 const wal = await NodeFSP.stat(`${path}-wal`).catch(() => null);
373 const fingerprint = [stat, wal]
374 .map((file) => (file ? `${file.size}:${file.mtimeMs}:${file.ctimeMs}` : "-"))
375 .join("/");
376 const cached = cache?.get(canonical);
377 let candidates: readonly UsageCandidate[];
378 if (cached?.fingerprint === fingerprint) {
379 candidates = cached.candidates;
380 } else {
381 candidates = await readDatabase(path, stat.mtimeMs);
382 cache?.set(canonical, { fingerprint, candidates });
383 }
391 const readFile = Effect.fn("readAntigravityUsage.readFile")(function* (path: string) {
392 const canonical = yield* fileSystem.realPath(path);
393 if (visited.has(canonical)) return null;
394 visited.add(canonical);
395 const fingerprintOf = (file: string) =>
396 Effect.tryPromise(() => NodeFSP.stat(file)).pipe(
397 Effect.map((stat) => ({ stat, part: `${stat.size}:${stat.mtimeMs}:${stat.ctimeMs}` })),
398 );
399 const { stat, part } = yield* fingerprintOf(path);
400 const wal = yield* fingerprintOf(`${path}-wal`).pipe(Effect.option);
401 const fingerprint = `${part}/${wal._tag === "Some" ? wal.value.part : "-"}`;
402 const cached = cache?.get(canonical);
403 if (cached?.fingerprint === fingerprint) return cached.candidates;
404 const candidates = yield* readDatabase(
405 path,
406 pathService.basename(path, ".db"),
407 stat.mtimeMs,
408 ).pipe(Effect.provide(NodeSqliteClient.layer({ filename: path, readonly: true })));
409 cache?.set(canonical, { fingerprint, candidates });
410 return candidates;
411 });
412 const walk = (directory: string, root: string): Effect.Effect<void> =>
413 Effect.gen(function* () {
414 const names = yield* fileSystem.readDirectory(directory).pipe(
415 Effect.catchTags({
416 PlatformError: (cause) => {
417 if (cause.reason._tag !== "NotFound") errors.push(directory);
418 return Effect.succeed(null);
419 },
420 }),
421 );
422 if (names === null) return;
423 names.sort((a, b) => a.localeCompare(b));
424 for (const name of names) {
425 const path = pathService.join(directory, name);
426 // Like a directory listing's entry types: symlinks are neither files nor directories.
427 const isLink = Exit.isSuccess(yield* Effect.exit(fileSystem.readLink(path)));
428 const type = isLink
429 ? "SymbolicLink"
430 : yield* fileSystem.stat(path).pipe(
431 Effect.map((info) => info.type),
432 Effect.orElseSucceed(() => null),
433 );
434 if (type === "Directory") {
435 yield* walk(path, root);
436 } else if ((type === "File" || type === null) && name.endsWith(".db")) {
437 const candidates = yield* readFile(path).pipe(
438 Effect.catch(() => {
439 errors.push(path);
440 return Effect.succeed(null);
441 }),
442 );
443 if (candidates === null) continue;
384444 const fileIndex = files.length;
385445 files.push({ root, path, records: [] });
386446 for (const [index, candidate] of candidates.entries()) {
387447 append(candidate, fileIndex);
388 if (index % 256 === 255) await NodeTimersPromises.setImmediate();
448 if (index % 256 === 255) yield* Effect.yieldNow;
389449 }
390 } catch {
391 errors.push(path);
392450 }
393451 }
394 }
395 };
396 for (const root of roots) await walk(root, root);
452 });
453 for (const root of roots) yield* walk(root, root);
397454 if (cache !== undefined) {
398455 for (const key of cache.keys()) if (!visited.has(key)) cache.delete(key);
399456 }
......@@ -403,4 +460,4 @@ export async function readAntigravityUsage(
403460 }
404461 }
405462 return { files, errors };
406}
463});
packages/provider-cursor/src/server/CursorAccountReader.ts+11-16
......@@ -5,9 +5,11 @@
55 * @module provider-cursor/server/CursorAccountReader
66 */
77import * as Context from "effect/Context";
8import * as Crypto from "effect/Crypto";
89import * as Effect from "effect/Effect";
10import * as FileSystem from "effect/FileSystem";
11import * as HttpClient from "effect/http/HttpClient";
912import * as Layer from "effect/Layer";
10import * as Result from "effect/Result";
1113
1214import type { CursorCredentialSource } from "./accountCache.ts";
1315import { readCursorAccountUsage, type CursorAccountUsageReadResult } from "./accountUsage.ts";
......@@ -29,23 +31,16 @@ export const layer = Layer.effect(
2931 CursorAccountReader,
3032 Effect.gen(function* () {
3133 const keychain = yield* CursorKeychain.CursorKeychain;
34 const context = yield* Effect.context<
35 FileSystem.FileSystem | Crypto.Crypto | HttpClient.HttpClient
36 >();
3237 return CursorAccountReader.of({
38 // Only a Keychain login asks the Keychain. The dashboard reader turns
39 // its outcome, a failure included, into the source's message.
3340 read: (credentialSource, sinceMs, untilMs) =>
34 Effect.gen(function* () {
35 // Only a Keychain login asks the Keychain. The dashboard reader
36 // turns its outcome, a failure included, into the source's message.
37 const token =
38 typeof credentialSource === "string"
39 ? Result.succeed(null)
40 : yield* Effect.result(keychain.accessToken);
41 return yield* Effect.promise(() =>
42 readCursorAccountUsage(credentialSource, sinceMs, untilMs, globalThis.fetch, () =>
43 Result.isSuccess(token)
44 ? Promise.resolve(token.success)
45 : Promise.reject(token.failure),
46 ),
47 );
48 }),
41 readCursorAccountUsage(credentialSource, sinceMs, untilMs, keychain.accessToken).pipe(
42 Effect.provideContext(context),
43 ),
4944 });
5045 }),
5146);
packages/provider-cursor/src/server/accountUsage.test.ts+379-308
......@@ -1,340 +1,411 @@
1// @effect-diagnostics nodeBuiltinImport:off - the reader under test reads a
2// real CLI credential file from disk.
3import * as NodeFSP from "node:fs/promises";
4import * as NodeOS from "node:os";
5import * as NodePath from "node:path";
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, describe, it } from "@effect/vitest";
3import * as Context from "effect/Context";
4import * as Effect from "effect/Effect";
5import * as FileSystem from "effect/FileSystem";
6import * as FetchHttpClient from "effect/http/FetchHttpClient";
7import * as HttpClient from "effect/http/HttpClient";
8import * as HttpClientError from "effect/http/HttpClientError";
9import * as HttpClientRequest from "effect/http/HttpClientRequest";
10import * as HttpClientResponse from "effect/http/HttpClientResponse";
11import * as Path from "effect/Path";
612
7import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest";
8
9import { CursorKeychainTimeoutError } from "./CursorKeychain.ts";
13import * as CursorKeychain from "./CursorKeychain.ts";
1014import { readCursorAccountUsage } from "./accountUsage.ts";
1115
12let dir: string;
13
14beforeEach(async () => {
15 dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "usage-reader-test-"));
16});
16/** A scoped temporary directory holding `auth.json` with `accessToken`. */
17const writeAuth = (accessToken: string) =>
18 Effect.gen(function* () {
19 const fileSystem = yield* FileSystem.FileSystem;
20 const path = yield* Path.Path;
21 const dir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "usage-reader-test-" });
22 const authPath = path.join(dir, "auth.json");
23 yield* fileSystem.writeFileString(authPath, JSON.stringify({ accessToken }));
24 return { dir, authPath };
25 });
1726
18afterEach(async () => {
19 await NodeFSP.rm(dir, { recursive: true, force: true });
20});
27/** One dashboard request as Cursor sees it: the URL, the fetch options and the decoded page. */
28interface DashboardRequest {
29 readonly url: string;
30 readonly init: RequestInit;
31 readonly page: number;
32 readonly headers: Headers;
33}
2134
22describe("Cursor account history", () => {
23 it("reads Cursor account history with the default macOS Keychain login", async () => {
24 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
25 let keychainReads = 0;
26 const result = await readCursorAccountUsage(
27 { kind: "keychain" },
28 0,
29 1781000000000,
30 async (_url, init) => {
31 assert.include(new Headers(init.headers).get("cookie") ?? "", "demo%3A%3A");
32 return Response.json({ totalUsageEventsCount: 0, usageEventsDisplay: [] });
33 },
34 async () => {
35 keychainReads++;
36 return accessToken;
35/** Stands in for Cursor's dashboard: `answer` replies to each page request. */
36const dashboard = (answer: (request: DashboardRequest) => Response | Promise<Response>) =>
37 HttpClient.make((request, url, signal, fiber) =>
38 Effect.tryPromise({
39 try: async () => {
40 const init = Context.getOrUndefined(fiber.context, FetchHttpClient.RequestInit);
41 const body =
42 request.body._tag === "Uint8Array" ? new TextDecoder().decode(request.body.body) : "";
43 const response = await answer({
44 url: url.toString(),
45 init: { ...init, signal, body },
46 page: JSON.parse(body).page,
47 headers: new Headers(request.headers),
48 });
49 return HttpClientResponse.fromWeb(request, response);
3750 },
38 );
39 assert.strictEqual(keychainReads, 1);
40 assert.isNull(result.error);
41 assert.isFalse(result.missing);
42 assert.isNotNull(result.accountKey);
43 });
51 catch: (cause) =>
52 new HttpClientError.HttpClientError({
53 reason: new HttpClientError.TransportError({ request, cause }),
54 }),
55 }),
56 );
4457
45 it("reads paginated Cursor account history including headless calls with separate cache tokens", async () => {
46 const authPath = NodePath.join(dir, "auth.json");
47 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo", exp: 4102444800 })).toString("base64url")}.signature`;
48 await NodeFSP.writeFile(authPath, JSON.stringify({ accessToken }));
49 const pages: number[] = [];
50 const signals: AbortSignal[] = [];
51 const request = async (url: string, init: RequestInit) => {
52 assert.strictEqual(String(url), "https://cursor.com/api/dashboard/get-filtered-usage-events");
53 assert.strictEqual(init?.redirect, "error");
54 const headers = new Headers(init?.headers);
55 assert.strictEqual(headers.get("origin"), "https://cursor.com");
56 assert.include(headers.get("cookie") ?? "", "WorkosCursorSessionToken=demo%3A%3A");
57 const body = JSON.parse(String(init?.body));
58 pages.push(body.page);
59 if (init.signal) signals.push(init.signal);
60 return Response.json({
61 totalUsageEventsCount: 1001,
62 usageEventsDisplay: Array.from({ length: body.page === 1 ? 1000 : 1 }, (_, index) => ({
63 timestamp: String(1780000000000 + ((body.page - 1) * 1000 + index) * 1000),
64 model: "claude-sonnet-4-5",
65 conversationId: `conversation-${body.page}`,
66 isHeadless: body.page === 2,
67 chargedCents: 0,
68 tokenUsage: {
69 inputTokens: 10,
70 outputTokens: 5,
71 cacheReadTokens: 30,
72 cacheWriteTokens: 2,
73 totalCents: 25,
74 },
75 })),
76 });
77 };
78 const result = await readCursorAccountUsage(authPath, 0, 1781000000000, request);
79 assert.isNull(result.error);
80 assert.deepStrictEqual(pages, [1, 2]);
81 assert.lengthOf(signals, 2);
82 assert.notStrictEqual(signals[0], signals[1]);
83 assert.strictEqual(result.records.length, 1001);
84 assert.strictEqual(result.records.at(-1)?.sessionId, "conversation-2");
85 assert.deepStrictEqual(result.records[0]?.totals, {
86 uncachedInputTokens: 10,
87 cachedInputTokens: 30,
88 cacheCreationTokens: 2,
89 outputTokens: 5,
90 reasoningTokens: 0,
91 });
92 assert.strictEqual(result.records[0]?.reportedCostUsd, 0.25);
93 assert.isFalse(result.accountKey?.includes("demo") ?? true);
94 });
95
96 it("reads Cursor account history beyond 100 pages", async () => {
97 const authPath = NodePath.join(dir, "auth.json");
98 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
99 await NodeFSP.writeFile(authPath, JSON.stringify({ accessToken }));
100 const fullPage = Array.from({ length: 1000 }, () => ({ tokenUsage: null }));
101 let requests = 0;
102 const result = await readCursorAccountUsage(authPath, 0, 1781000000000, async () => {
103 requests += 1;
104 return Response.json({
105 totalUsageEventsCount: 100_001,
106 usageEventsDisplay: requests <= 100 ? fullPage : [{ tokenUsage: null }],
107 });
108 });
109 assert.isNull(result.error);
110 assert.strictEqual(requests, 101);
111 assert.deepStrictEqual(result.records, []);
112 });
58/** Runs `readCursorAccountUsage` against `answer` standing in for Cursor. */
59const readAccountUsage = (
60 credentialSource: Parameters<typeof readCursorAccountUsage>[0],
61 sinceMs: number,
62 endDate: number,
63 answer: (request: DashboardRequest) => Response | Promise<Response>,
64 keychainToken?: Parameters<typeof readCursorAccountUsage>[3],
65) =>
66 readCursorAccountUsage(credentialSource, sinceMs, endDate, keychainToken).pipe(
67 Effect.provideService(HttpClient.HttpClient, dashboard(answer)),
68 );
11369
114 it("accepts confirmed empty Cursor usage but rejects error envelopes", async () => {
115 const authPath = NodePath.join(dir, "auth.json");
116 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
117 await NodeFSP.writeFile(authPath, JSON.stringify({ accessToken }));
118 for (const body of [
119 {},
120 { totalUsageEventsCount: 0 },
121 { totalUsageEventsCount: 0, usageEventsDisplay: [] },
122 ]) {
123 const result = await readCursorAccountUsage(authPath, 0, 1781000000000, async () =>
124 Response.json(body),
70describe("Cursor account history", () => {
71 it.effect("reads Cursor account history with the default macOS Keychain login", () =>
72 Effect.gen(function* () {
73 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
74 let keychainReads = 0;
75 const result = yield* readAccountUsage(
76 { kind: "keychain" },
77 0,
78 1781000000000,
79 ({ headers }) => {
80 assert.include(headers.get("cookie") ?? "", "demo%3A%3A");
81 return Response.json({ totalUsageEventsCount: 0, usageEventsDisplay: [] });
82 },
83 Effect.sync(() => {
84 keychainReads++;
85 return accessToken;
86 }),
12587 );
88 assert.strictEqual(keychainReads, 1);
12689 assert.isNull(result.error);
127 assert.deepStrictEqual(result.records, []);
12890 assert.isFalse(result.missing);
129 }
130 for (const body of [
131 { error: "upstream error" },
132 { detail: "unknown error envelope" },
133 { totalUsageEventsCount: 0, error: "upstream error" },
134 null,
135 [],
136 "invalid",
137 0,
138 ]) {
139 const result = await readCursorAccountUsage(authPath, 0, 1781000000000, async () =>
140 Response.json(body),
141 );
142 assert.isNotNull(result.error);
143 assert.deepStrictEqual(result.records, []);
144 }
145 });
91 assert.isNotNull(result.accountKey);
92 }).pipe(Effect.provide(NodeServices.layer)),
93 );
14694
147 it("requires a terminal Cursor page after a full page reaches the reported count", async () => {
148 const authPath = NodePath.join(dir, "auth.json");
149 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
150 await NodeFSP.writeFile(authPath, JSON.stringify({ accessToken }));
151 let requests = 0;
152 const result = await readCursorAccountUsage(authPath, 0, 1781000000000, async () => {
153 requests++;
154 return Response.json(
155 requests === 1
156 ? {
157 totalUsageEventsCount: 1000,
158 usageEventsDisplay: Array.from({ length: 1000 }, (_, index) => ({
159 timestamp: String(1780000000000 + index),
160 model: "gpt-5",
161 tokenUsage: { inputTokens: 10, outputTokens: 5 },
162 })),
163 }
164 : { totalUsageEventsCount: 1000 },
165 );
166 });
167 assert.isNull(result.error);
168 assert.strictEqual(result.records.length, 1000);
169 assert.strictEqual(requests, 2);
170 });
95 it.effect(
96 "reads paginated Cursor account history including headless calls with separate cache tokens",
97 () =>
98 Effect.gen(function* () {
99 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo", exp: 4102444800 })).toString("base64url")}.signature`;
100 const { authPath } = yield* writeAuth(accessToken);
101 const pages: number[] = [];
102 const signals: AbortSignal[] = [];
103 const request = ({ url, init, headers, page }: DashboardRequest) => {
104 assert.strictEqual(url, "https://cursor.com/api/dashboard/get-filtered-usage-events");
105 assert.strictEqual(init.redirect, "error");
106 assert.strictEqual(headers.get("origin"), "https://cursor.com");
107 assert.strictEqual(headers.get("content-type"), "application/json");
108 assert.include(headers.get("cookie") ?? "", "WorkosCursorSessionToken=demo%3A%3A");
109 const body = { page };
110 pages.push(body.page);
111 if (init.signal) signals.push(init.signal);
112 return Response.json({
113 totalUsageEventsCount: 1001,
114 usageEventsDisplay: Array.from({ length: body.page === 1 ? 1000 : 1 }, (_, index) => ({
115 timestamp: String(1780000000000 + ((body.page - 1) * 1000 + index) * 1000),
116 model: "claude-sonnet-4-5",
117 conversationId: `conversation-${body.page}`,
118 isHeadless: body.page === 2,
119 chargedCents: 0,
120 tokenUsage: {
121 inputTokens: 10,
122 outputTokens: 5,
123 cacheReadTokens: 30,
124 cacheWriteTokens: 2,
125 totalCents: 25,
126 },
127 })),
128 });
129 };
130 const result = yield* readAccountUsage(authPath, 0, 1781000000000, request);
131 assert.isNull(result.error);
132 assert.deepStrictEqual(pages, [1, 2]);
133 assert.lengthOf(signals, 2);
134 assert.notStrictEqual(signals[0], signals[1]);
135 assert.strictEqual(result.records.length, 1001);
136 assert.strictEqual(result.records.at(-1)?.sessionId, "conversation-2");
137 assert.deepStrictEqual(result.records[0]?.totals, {
138 uncachedInputTokens: 10,
139 cachedInputTokens: 30,
140 cacheCreationTokens: 2,
141 outputTokens: 5,
142 reasoningTokens: 0,
143 });
144 assert.strictEqual(result.records[0]?.reportedCostUsd, 0.25);
145 assert.isFalse(result.accountKey?.includes("demo") ?? true);
146 }).pipe(Effect.provide(NodeServices.layer)),
147 );
171148
172 it("removes only count-proven Cursor boundary copies and preserves identical billed events", async () => {
173 const authPath = NodePath.join(dir, "auth.json");
174 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
175 await NodeFSP.writeFile(authPath, JSON.stringify({ accessToken }));
176 const event = (index: number) => ({
177 timestamp: String(1780000000000 + index),
178 model: "gpt-5",
179 tokenUsage: { inputTokens: 10, outputTokens: 5, totalCents: 1 },
180 });
181 for (const total of [2000, 2001]) {
149 it.effect("reads Cursor account history beyond 100 pages", () =>
150 Effect.gen(function* () {
151 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
152 const { authPath } = yield* writeAuth(accessToken);
153 const fullPage = Array.from({ length: 1000 }, () => ({ tokenUsage: null }));
182154 let requests = 0;
183 const result = await readCursorAccountUsage(authPath, 0, 1781000000000, async () => {
184 requests++;
155 const result = yield* readAccountUsage(authPath, 0, 1781000000000, async () => {
156 requests += 1;
185157 return Response.json({
186 totalUsageEventsCount: total,
187 usageEventsDisplay:
188 requests === 1
189 ? Array.from({ length: 1000 }, (_, index) => event(index))
190 : requests === 2
191 ? Array.from({ length: 1000 }, (_, index) => event(999 + index))
192 : [event(1999)],
158 totalUsageEventsCount: 100_001,
159 usageEventsDisplay: requests <= 100 ? fullPage : [{ tokenUsage: null }],
193160 });
194161 });
195162 assert.isNull(result.error);
196 assert.strictEqual(result.records.length, total);
197 assert.strictEqual(requests, 3);
198 assert.strictEqual(result.records.at(-1)?.timestampMs, 1780000001999);
199 assert.strictEqual(
200 result.records.filter((record) => record.timestampMs === 1780000000999).length,
201 total === 2000 ? 1 : 2,
202 );
203 assert.strictEqual(new Set(result.records.map((record) => record.dedupeKey)).size, total);
204 }
205 let requests = 0;
206 const inconsistent = await readCursorAccountUsage(authPath, 0, 1781000000000, async () => {
207 requests++;
208 return Response.json({
209 totalUsageEventsCount: 1001,
210 usageEventsDisplay:
163 assert.strictEqual(requests, 101);
164 assert.deepStrictEqual(result.records, []);
165 }).pipe(Effect.provide(NodeServices.layer)),
166 );
167
168 it.effect("accepts confirmed empty Cursor usage but rejects error envelopes", () =>
169 Effect.gen(function* () {
170 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
171 const { authPath } = yield* writeAuth(accessToken);
172 for (const body of [
173 {},
174 { totalUsageEventsCount: 0 },
175 { totalUsageEventsCount: 0, usageEventsDisplay: [] },
176 ]) {
177 const result = yield* readAccountUsage(authPath, 0, 1781000000000, async () =>
178 Response.json(body),
179 );
180 assert.isNull(result.error);
181 assert.deepStrictEqual(result.records, []);
182 assert.isFalse(result.missing);
183 }
184 for (const body of [
185 { error: "upstream error" },
186 { detail: "unknown error envelope" },
187 { totalUsageEventsCount: 0, error: "upstream error" },
188 null,
189 [],
190 "invalid",
191 0,
192 ]) {
193 const result = yield* readAccountUsage(authPath, 0, 1781000000000, async () =>
194 Response.json(body),
195 );
196 assert.isNotNull(result.error);
197 assert.deepStrictEqual(result.records, []);
198 }
199 }).pipe(Effect.provide(NodeServices.layer)),
200 );
201
202 it.effect("requires a terminal Cursor page after a full page reaches the reported count", () =>
203 Effect.gen(function* () {
204 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
205 const { authPath } = yield* writeAuth(accessToken);
206 let requests = 0;
207 const result = yield* readAccountUsage(authPath, 0, 1781000000000, async () => {
208 requests++;
209 return Response.json(
211210 requests === 1
212 ? Array.from({ length: 1000 }, (_, index) => event(index))
213 : [event(500), event(1000)],
211 ? {
212 totalUsageEventsCount: 1000,
213 usageEventsDisplay: Array.from({ length: 1000 }, (_, index) => ({
214 timestamp: String(1780000000000 + index),
215 model: "gpt-5",
216 tokenUsage: { inputTokens: 10, outputTokens: 5 },
217 })),
218 }
219 : { totalUsageEventsCount: 1000 },
220 );
214221 });
215 });
216 assert.isNotNull(inconsistent.error);
217 assert.deepStrictEqual(inconsistent.records, []);
218 });
222 assert.isNull(result.error);
223 assert.strictEqual(result.records.length, 1000);
224 assert.strictEqual(requests, 2);
225 }).pipe(Effect.provide(NodeServices.layer)),
226 );
219227
220 it("does not present truncated Cursor account pages or authentication failures as complete history", async () => {
221 const authPath = NodePath.join(dir, "auth.json");
222 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo", exp: 4102444800 })).toString("base64url")}.signature`;
223 await NodeFSP.writeFile(authPath, JSON.stringify({ accessToken }));
224 const truncated = await readCursorAccountUsage(authPath, 0, 1781000000000, async () =>
225 Response.json({ totalUsageEventsCount: 101, usageEventsDisplay: [] }),
226 );
227 assert.isNotNull(truncated.error);
228 assert.deepStrictEqual(truncated.records, []);
229 const denied = await readCursorAccountUsage(
230 authPath,
231 0,
232 1781000000000,
233 async () => new Response(accessToken, { status: 401 }),
234 );
235 assert.isNotNull(denied.error);
236 assert.isFalse(denied.error?.includes(accessToken) ?? true);
237 assert.deepStrictEqual(denied.records, []);
238 let requested = false;
239 const missing = await readCursorAccountUsage(
240 NodePath.join(dir, "missing.json"),
241 0,
242 1781000000000,
243 async () => {
244 requested = true;
245 return Response.json({});
246 },
247 );
248 assert.isTrue(missing.missing);
249 assert.isFalse(requested);
250 });
228 it.effect(
229 "removes only count-proven Cursor boundary copies and preserves identical billed events",
230 () =>
231 Effect.gen(function* () {
232 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
233 const { authPath } = yield* writeAuth(accessToken);
234 const event = (index: number) => ({
235 timestamp: String(1780000000000 + index),
236 model: "gpt-5",
237 tokenUsage: { inputTokens: 10, outputTokens: 5, totalCents: 1 },
238 });
239 for (const total of [2000, 2001]) {
240 let requests = 0;
241 const result = yield* readAccountUsage(authPath, 0, 1781000000000, async () => {
242 requests++;
243 return Response.json({
244 totalUsageEventsCount: total,
245 usageEventsDisplay:
246 requests === 1
247 ? Array.from({ length: 1000 }, (_, index) => event(index))
248 : requests === 2
249 ? Array.from({ length: 1000 }, (_, index) => event(999 + index))
250 : [event(1999)],
251 });
252 });
253 assert.isNull(result.error);
254 assert.strictEqual(result.records.length, total);
255 assert.strictEqual(requests, 3);
256 assert.strictEqual(result.records.at(-1)?.timestampMs, 1780000001999);
257 assert.strictEqual(
258 result.records.filter((record) => record.timestampMs === 1780000000999).length,
259 total === 2000 ? 1 : 2,
260 );
261 assert.strictEqual(new Set(result.records.map((record) => record.dedupeKey)).size, total);
262 }
263 let requests = 0;
264 const inconsistent = yield* readAccountUsage(authPath, 0, 1781000000000, async () => {
265 requests++;
266 return Response.json({
267 totalUsageEventsCount: 1001,
268 usageEventsDisplay:
269 requests === 1
270 ? Array.from({ length: 1000 }, (_, index) => event(index))
271 : [event(500), event(1000)],
272 });
273 });
274 assert.isNotNull(inconsistent.error);
275 assert.deepStrictEqual(inconsistent.records, []);
276 }).pipe(Effect.provide(NodeServices.layer)),
277 );
278
279 it.effect(
280 "does not present truncated Cursor account pages or authentication failures as complete history",
281 () =>
282 Effect.gen(function* () {
283 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo", exp: 4102444800 })).toString("base64url")}.signature`;
284 const { dir, authPath } = yield* writeAuth(accessToken);
285 const truncated = yield* readAccountUsage(authPath, 0, 1781000000000, async () =>
286 Response.json({ totalUsageEventsCount: 101, usageEventsDisplay: [] }),
287 );
288 assert.isNotNull(truncated.error);
289 assert.deepStrictEqual(truncated.records, []);
290 const denied = yield* readAccountUsage(
291 authPath,
292 0,
293 1781000000000,
294 async () => new Response(accessToken, { status: 401 }),
295 );
296 assert.isNotNull(denied.error);
297 assert.isFalse(denied.error?.includes(accessToken) ?? true);
298 assert.deepStrictEqual(denied.records, []);
299 let requested = false;
300 const missing = yield* readAccountUsage(
301 (yield* Path.Path).join(dir, "missing.json"),
302 0,
303 1781000000000,
304 async () => {
305 requested = true;
306 return Response.json({});
307 },
308 );
309 assert.isTrue(missing.missing);
310 assert.isFalse(requested);
311 }).pipe(Effect.provide(NodeServices.layer)),
312 );
251313});
252314
253315describe("readCursorAccountUsage", () => {
254 it("asks for Keychain approval when the prompt goes unanswered", async () => {
255 const result = await readCursorAccountUsage(
256 { kind: "keychain" },
257 0,
258 1,
259 () => Promise.reject(new Error("no network expected")),
260 () => Promise.reject(new CursorKeychainTimeoutError()),
261 );
262 assert.deepStrictEqual(result, {
263 accountKey: null,
264 records: [],
265 missing: false,
266 error: "Allow Keychain access on the Mac running T3 Code, then refresh.",
267 });
268 });
269
270 it("reads the pages behind the first together and keeps them in page order", async () => {
271 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
272 let inFlight = 0;
273 let mostInFlight = 0;
274 const result = await readCursorAccountUsage(
275 { kind: "keychain" },
276 0,
277 1781000000000,
278 async (_url, init) => {
279 const page: number = JSON.parse(String(init.body)).page;
280 inFlight++;
281 mostInFlight = Math.max(mostInFlight, inFlight);
282 // Later pages answer first, so record order cannot come from arrival order.
283 for (let turn = page; turn < 9; turn++) await Promise.resolve();
284 inFlight--;
285 return Response.json({
286 totalUsageEventsCount: 8001,
287 usageEventsDisplay: Array.from({ length: page === 9 ? 1 : 1000 }, (_, index) => ({
288 timestamp: String(1780000000000 + (page - 1) * 1000 + index),
289 model: "gpt-5",
290 tokenUsage: { inputTokens: 1 },
291 })),
292 });
293 },
294 async () => accessToken,
295 );
296 assert.isNull(result.error);
297 assert.strictEqual(mostInFlight, 6);
298 assert.deepStrictEqual(
299 result.records.map((record) => record.timestampMs),
300 Array.from({ length: 8001 }, (_, index) => 1780000000000 + index),
301 );
302 });
316 it.effect("asks for Keychain approval when the prompt goes unanswered", () =>
317 Effect.gen(function* () {
318 const result = yield* readAccountUsage(
319 { kind: "keychain" },
320 0,
321 1,
322 () => {
323 throw new Error("no network expected");
324 },
325 Effect.fail(new CursorKeychain.CursorKeychainTimeoutError()),
326 );
327 assert.deepStrictEqual(result, {
328 accountKey: null,
329 records: [],
330 missing: false,
331 error: "Allow Keychain access on the Mac running T3 Code, then refresh.",
332 });
333 }).pipe(Effect.provide(NodeServices.layer)),
334 );
303335
304 it("ends like a page-by-page read when a page fails, leaving no request open", async () => {
305 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
306 const fullPage = {
307 totalUsageEventsCount: 8001,
308 usageEventsDisplay: Array.from({ length: 1000 }, () => ({ tokenUsage: null })),
309 };
310 // Page 3 fails while the pages after it are still waiting on Cursor.
311 for (const [secondPage, error] of [
312 [() => Response.json(fullPage), "Cursor account usage could not be read."],
313 [() => new Response(null, { status: 401 }), "Sign in to Cursor again to read account usage."],
314 ] as const) {
315 let open = 0;
316 const result = await readCursorAccountUsage(
336 it.effect("reads the pages behind the first together and keeps them in page order", () =>
337 Effect.gen(function* () {
338 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
339 let inFlight = 0;
340 let mostInFlight = 0;
341 const result = yield* readAccountUsage(
317342 { kind: "keychain" },
318343 0,
319344 1781000000000,
320 (_url, init) => {
321 const page: number = JSON.parse(String(init.body)).page;
322 if (page === 1) return Promise.resolve(Response.json(fullPage));
323 if (page === 2) return Promise.resolve(secondPage());
324 if (page === 3) return Promise.reject(new Error("connection reset"));
325 open++;
326 return new Promise((_resolve, reject) => {
327 init.signal?.addEventListener("abort", () => {
328 open--;
329 reject(init.signal?.reason);
330 });
345 async ({ page }) => {
346 inFlight++;
347 mostInFlight = Math.max(mostInFlight, inFlight);
348 // Later pages answer first, so record order cannot come from arrival order.
349 for (let turn = page; turn < 9; turn++) await Promise.resolve();
350 inFlight--;
351 return Response.json({
352 totalUsageEventsCount: 8001,
353 usageEventsDisplay: Array.from({ length: page === 9 ? 1 : 1000 }, (_, index) => ({
354 timestamp: String(1780000000000 + (page - 1) * 1000 + index),
355 model: "gpt-5",
356 tokenUsage: { inputTokens: 1 },
357 })),
331358 });
332359 },
333 async () => accessToken,
360 Effect.succeed(accessToken),
334361 );
335 assert.strictEqual(result.error, error);
336 assert.deepStrictEqual(result.records, []);
337 assert.strictEqual(open, 0);
338 }
339 });
362 assert.isNull(result.error);
363 assert.strictEqual(mostInFlight, 6);
364 assert.deepStrictEqual(
365 result.records.map((record) => record.timestampMs),
366 Array.from({ length: 8001 }, (_, index) => 1780000000000 + index),
367 );
368 }).pipe(Effect.provide(NodeServices.layer)),
369 );
370
371 it.effect("ends like a page-by-page read when a page fails, leaving no request open", () =>
372 Effect.gen(function* () {
373 const accessToken = `header.${Buffer.from(JSON.stringify({ sub: "auth|demo" })).toString("base64url")}.signature`;
374 const fullPage = {
375 totalUsageEventsCount: 8001,
376 usageEventsDisplay: Array.from({ length: 1000 }, () => ({ tokenUsage: null })),
377 };
378 // Page 3 fails while the pages after it are still waiting on Cursor.
379 for (const [secondPage, error] of [
380 [() => Response.json(fullPage), "Cursor account usage could not be read."],
381 [
382 () => new Response(null, { status: 401 }),
383 "Sign in to Cursor again to read account usage.",
384 ],
385 ] as const) {
386 let open = 0;
387 const result = yield* readAccountUsage(
388 { kind: "keychain" },
389 0,
390 1781000000000,
391 ({ init, page }) => {
392 if (page === 1) return Promise.resolve(Response.json(fullPage));
393 if (page === 2) return Promise.resolve(secondPage());
394 if (page === 3) return Promise.reject(new Error("connection reset"));
395 open++;
396 return new Promise((_resolve, reject) => {
397 init.signal?.addEventListener("abort", () => {
398 open--;
399 reject(init.signal?.reason);
400 });
401 });
402 },
403 Effect.succeed(accessToken),
404 );
405 assert.strictEqual(result.error, error);
406 assert.deepStrictEqual(result.records, []);
407 assert.strictEqual(open, 0);
408 }
409 }).pipe(Effect.provide(NodeServices.layer)),
410 );
340411});
packages/provider-cursor/src/server/accountUsage.ts+201-115
......@@ -1,5 +1,3 @@
1// Node fs reads CLI credentials, and crypto hashes account IDs for deduplication.
2// @effect-diagnostics nodeBuiltinImport:off
31/**
42 * Cursor account usage from its dashboard API, read with the saved CLI or
53 * Keychain login. The dashboard covers CLI, desktop and headless usage from
......@@ -7,13 +5,22 @@
75 *
86 * @module provider-cursor/server/accountUsage
97 */
10import * as NodeFSP from "node:fs/promises";
11import * as NodeCrypto from "node:crypto";
12import * as NodeTimersPromises from "node:timers/promises";
13
148import type { UsageRecord } from "@t3tools/provider-core/server/usage";
9import * as Crypto from "effect/Crypto";
10import * as Duration from "effect/Duration";
11import * as Effect from "effect/Effect";
12import * as Base64Url from "effect/encoding/Base64Url";
13import * as Hex from "effect/encoding/Hex";
14import * as Fiber from "effect/Fiber";
15import * as FileSystem from "effect/FileSystem";
16import * as FetchHttpClient from "effect/http/FetchHttpClient";
17import * as HttpClient from "effect/http/HttpClient";
18import * as HttpClientRequest from "effect/http/HttpClientRequest";
19import * as PlatformError from "effect/PlatformError";
20import * as Result from "effect/Result";
21import * as Schema from "effect/Schema";
1522
16import { isCursorKeychainTimeoutError } from "./CursorKeychain.ts";
23import * as CursorKeychain from "./CursorKeychain.ts";
1724
1825function object(value: unknown): Record<string, unknown> {
1926 return typeof value === "object" && value !== null && !Array.isArray(value)
......@@ -44,7 +51,27 @@ export interface CursorAccountUsageReadResult {
4451 readonly error: string | null;
4552}
4653
47const accountHash = (value: string) => NodeCrypto.createHash("sha256").update(value).digest("hex");
54/** Why a dashboard read was abandoned; the caller only sees the generic message. */
55class CursorAccountUsageInvalidError extends Schema.TaggedError<CursorAccountUsageInvalidError>()(
56 "CursorAccountUsageInvalidError",
57 { reason: Schema.String },
58) {}
59
60const invalid = (reason: string) => Effect.fail(new CursorAccountUsageInvalidError({ reason }));
61
62const decodeJson = Schema.decodeEffect(Schema.fromJsonString(Schema.Unknown));
63
64const pageRequestBody = (page: number, pageSize: number, sinceMs: number, endDate: number) =>
65 JSON.stringify({ page, pageSize, startDate: String(sinceMs), endDate: String(endDate) });
66
67const billedEventIdentity = (...fields: readonly unknown[]) => JSON.stringify(fields);
68
69const textEncoder = new TextEncoder();
70
71const accountHash = Effect.fn("accountHash")(function* (value: string) {
72 const crypto = yield* Crypto.Crypto;
73 return Hex.encode(yield* crypto.digest("SHA-256", textEncoder.encode(value)));
74});
4875
4976function canonicalJson(value: unknown): string {
5077 if (Array.isArray(value)) return `[${value.map(canonicalJson).join(",")}]`;
......@@ -70,26 +97,54 @@ function boundaryOverlap(previous: readonly string[], current: readonly string[]
7097 return lengths.at(-1) ?? 0;
7198}
7299
100const DASHBOARD_USAGE_URL = "https://cursor.com/api/dashboard/get-filtered-usage-events";
101
102/** The session cookie must never follow a redirect off cursor.com. */
103const NO_REDIRECT: RequestInit = { redirect: "error" };
104
73105/**
74106 * Dashboard usage includes headless agents and reports fresh input separately
75 * from cache reads. `keychainToken` is only called for a Keychain login.
107 * from cache reads. `keychainToken` is only run for a Keychain login.
76108 */
77export async function readCursorAccountUsage(
109export const readCursorAccountUsage = Effect.fn("readCursorAccountUsage")(function* (
78110 credentialSource: string | { readonly kind: "keychain" },
79111 sinceMs: number,
80112 endDate: number,
81 request: (url: string, init: RequestInit) => Promise<Response> = globalThis.fetch,
82 keychainToken: () => Promise<string | null> = () =>
83 Promise.reject(new Error("No Keychain reader")),
84): Promise<CursorAccountUsageReadResult> {
85 let accessToken: unknown;
86 try {
87 accessToken =
88 typeof credentialSource === "string"
89 ? object(JSON.parse(await NodeFSP.readFile(credentialSource, "utf8"))).accessToken
90 : await keychainToken();
91 } catch (cause) {
92 const missing = typeof credentialSource === "string" && object(cause).code === "ENOENT";
113 keychainToken: Effect.Effect<
114 string | null,
115 | CursorKeychain.CursorKeychainTimeoutError
116 | CursorKeychain.CursorKeychainReadError
117 | CursorAccountUsageInvalidError
118 > = invalid("No Keychain reader"),
119): Effect.fn.Return<
120 CursorAccountUsageReadResult,
121 never,
122 FileSystem.FileSystem | Crypto.Crypto | HttpClient.HttpClient
123> {
124 const fileSystem = yield* FileSystem.FileSystem;
125 const httpClient = yield* HttpClient.HttpClient;
126 const credential = yield* Effect.result<
127 unknown,
128 | PlatformError.PlatformError
129 | Schema.SchemaError
130 | CursorKeychain.CursorKeychainTimeoutError
131 | CursorKeychain.CursorKeychainReadError
132 | CursorAccountUsageInvalidError,
133 never
134 >(
135 typeof credentialSource === "string"
136 ? fileSystem.readFileString(credentialSource).pipe(
137 Effect.flatMap(decodeJson),
138 Effect.map((credentials) => object(credentials).accessToken),
139 )
140 : keychainToken,
141 );
142 if (Result.isFailure(credential)) {
143 const cause = credential.failure;
144 const missing =
145 typeof credentialSource === "string" &&
146 PlatformError.isPlatformError(cause) &&
147 cause.reason._tag === "NotFound";
93148 return {
94149 accountKey: null,
95150 records: [],
......@@ -98,11 +153,12 @@ export async function readCursorAccountUsage(
98153 ? null
99154 : typeof credentialSource === "string"
100155 ? "Cursor credentials could not be read."
101 : isCursorKeychainTimeoutError(cause)
156 : CursorKeychain.isCursorKeychainTimeoutError(cause)
102157 ? "Allow Keychain access on the Mac running T3 Code, then refresh."
103158 : "Cursor Keychain credentials could not be read.",
104159 };
105160 }
161 const accessToken = credential.success;
106162 if (typeof accessToken !== "string" || !accessToken) {
107163 return {
108164 accountKey: null,
......@@ -115,95 +171,112 @@ export async function readCursorAccountUsage(
115171 };
116172 }
117173 let accountKey: string | null = null;
118 const cancel = new AbortController();
119 try {
120 const payload = accessToken.split(".")[1];
174 const read = Effect.gen(function* () {
175 const payload = Base64Url.decodeString(accessToken.split(".")[1] ?? "");
176 if (Result.isFailure(payload)) return yield* invalid("Invalid authentication");
121177 const subject = object(
122 JSON.parse(Buffer.from(payload ?? "", "base64url").toString("utf8")),
178 yield* decodeJson(payload.success).pipe(
179 Effect.catch(() => invalid("Invalid authentication")),
180 ),
123181 ).sub;
124 if (typeof subject !== "string" || !subject) throw new Error("Invalid authentication");
182 if (typeof subject !== "string" || !subject) return yield* invalid("Invalid authentication");
125183 const userId = subject.split("|").at(-1);
126 if (!userId) throw new Error("Invalid authentication");
127 accountKey = accountHash(subject);
184 if (!userId) return yield* invalid("Invalid authentication");
185 accountKey = yield* accountHash(subject);
128186 if (!Number.isFinite(sinceMs) || !Number.isFinite(endDate) || sinceMs < 0 || sinceMs > endDate)
129 throw new Error("Invalid date window");
130 const deadline = AbortSignal.any([cancel.signal, AbortSignal.timeout(60_000)]);
187 return yield* invalid("Invalid date window");
131188 const records: UsageRecord[] = [];
132189 const occurrences = new Map<string, number>();
133190 const pages: unknown[][] = [];
134191 let completed = false;
135192 const pageSize = 1000;
136193 let total: number | undefined;
137 const readPage = async (page: number) => {
138 // A count can include overlapping page boundaries. Allow room to
139 // reconcile them without imposing a fixed account-size limit.
140 if (page > (total === undefined ? 1000 : Math.ceil(total / pageSize) * 2 + 1)) {
141 throw new Error("Account usage page limit exceeded");
142 }
143 const response = await request("https://cursor.com/api/dashboard/get-filtered-usage-events", {
144 method: "POST",
145 redirect: "error",
146 signal: AbortSignal.any([deadline, AbortSignal.timeout(10_000)]),
147 headers: {
148 "Content-Type": "application/json",
149 Origin: "https://cursor.com",
150 Cookie: `WorkosCursorSessionToken=${encodeURIComponent(`${userId}::${accessToken}`)}`,
151 },
152 body: JSON.stringify({
153 page,
154 pageSize,
155 startDate: String(sinceMs),
156 endDate: String(endDate),
194 const readPage = (page: number) =>
195 Effect.gen(function* () {
196 // A count can include overlapping page boundaries. Allow room to
197 // reconcile them without imposing a fixed account-size limit.
198 if (page > (total === undefined ? 1000 : Math.ceil(total / pageSize) * 2 + 1)) {
199 return yield* invalid("Account usage page limit exceeded");
200 }
201 const response = yield* httpClient
202 .execute(
203 HttpClientRequest.post(DASHBOARD_USAGE_URL).pipe(
204 HttpClientRequest.setHeaders({
205 Origin: "https://cursor.com",
206 Cookie: `WorkosCursorSessionToken=${encodeURIComponent(`${userId}::${accessToken}`)}`,
207 }),
208 HttpClientRequest.bodyText(
209 pageRequestBody(page, pageSize, sinceMs, endDate),
210 "application/json",
211 ),
212 ),
213 )
214 .pipe(
215 Effect.provideService(FetchHttpClient.RequestInit, NO_REDIRECT),
216 Effect.mapError(
217 () => new CursorAccountUsageInvalidError({ reason: "Account usage request failed" }),
218 ),
219 );
220 if (response.status === 401 || response.status === 403) {
221 return {
222 accountKey,
223 records: [],
224 missing: false,
225 error: "Sign in to Cursor again to read account usage.",
226 } satisfies CursorAccountUsageReadResult;
227 }
228 if (response.status < 200 || response.status >= 300)
229 return yield* invalid("Account usage request failed");
230 const parsed: unknown = yield* response.json.pipe(
231 Effect.mapError(
232 () => new CursorAccountUsageInvalidError({ reason: "Invalid account usage page" }),
233 ),
234 );
235 if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
236 return yield* invalid("Invalid account usage page");
237 }
238 const body = object(parsed);
239 const keys = Object.keys(body);
240 if ("error" in body || "message" in body || "code" in body)
241 return yield* invalid("Account usage error response");
242 const count = keys.length === 0 ? 0 : body.totalUsageEventsCount;
243 const events =
244 keys.length === 0 || (keys.length === 1 && keys[0] === "totalUsageEventsCount")
245 ? []
246 : body.usageEventsDisplay;
247 if (
248 (count !== undefined &&
249 (typeof count !== "number" ||
250 !Number.isSafeInteger(count) ||
251 count < 0 ||
252 (total !== undefined && count !== total))) ||
253 !Array.isArray(events) ||
254 events.length > pageSize ||
255 (count === undefined && !Array.isArray(body.usageEventsDisplay))
256 ) {
257 return yield* invalid("Inconsistent account usage page");
258 }
259 if (typeof count === "number") total = count;
260 return events as unknown[];
261 }).pipe(
262 Effect.timeoutOrElse({
263 duration: Duration.seconds(10),
264 orElse: () => invalid("Account usage request timed out"),
157265 }),
158 });
159 if (response.status === 401 || response.status === 403) {
160 return {
161 accountKey,
162 records: [],
163 missing: false,
164 error: "Sign in to Cursor again to read account usage.",
165 };
166 }
167 if (!response.ok) throw new Error("Account usage request failed");
168 const parsed: unknown = await response.json();
169 if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
170 throw new Error("Invalid account usage page");
171 }
172 const body = object(parsed);
173 const keys = Object.keys(body);
174 if ("error" in body || "message" in body || "code" in body)
175 throw new Error("Account usage error response");
176 const count = keys.length === 0 ? 0 : body.totalUsageEventsCount;
177 const events =
178 keys.length === 0 || (keys.length === 1 && keys[0] === "totalUsageEventsCount")
179 ? []
180 : body.usageEventsDisplay;
181 if (
182 (count !== undefined &&
183 (typeof count !== "number" ||
184 !Number.isSafeInteger(count) ||
185 count < 0 ||
186 (total !== undefined && count !== total))) ||
187 !Array.isArray(events) ||
188 events.length > pageSize ||
189 (count === undefined && !Array.isArray(body.usageEventsDisplay))
190 ) {
191 throw new Error("Inconsistent account usage page");
192 }
193 if (typeof count === "number") total = count;
194 return events;
195 };
266 );
196267 // Cursor caps a page at 1,000 events and takes about a second to answer one, so the pages
197268 // the first one's count implies are requested up to six ahead and consumed in page order.
198 const ahead: ReturnType<typeof readPage>[] = [];
269 // Pages still in flight when the read ends are interrupted with its scope.
270 const ahead: Fiber.Fiber<
271 Effect.Success<ReturnType<typeof readPage>>,
272 CursorAccountUsageInvalidError
273 >[] = [];
199274 for (let page = 1; ; page++) {
200275 while (ahead.length < Math.min(6, Math.ceil((total ?? 0) / pageSize) - page + 1)) {
201 const pending = readPage(page + ahead.length);
202 // Handled when its turn comes, or cancelled if the read ends first.
203 pending.catch(() => undefined);
204 ahead.push(pending);
276 ahead.push(yield* Effect.forkScoped(readPage(page + ahead.length)));
205277 }
206 const events = await (ahead.shift() ?? readPage(page));
278 const next = ahead.shift();
279 const events = yield* next === undefined ? readPage(page) : Fiber.join(next);
207280 // A rejected login comes back as the finished result instead of a page.
208281 if (!Array.isArray(events)) return events;
209282 pages.push(events);
......@@ -212,14 +285,17 @@ export async function readCursorAccountUsage(
212285 break;
213286 }
214287 }
215 if (!completed) throw new Error("Account usage page limit exceeded");
288 if (!completed) return yield* invalid("Account usage page limit exceeded");
216289 const rawCount = pages.reduce((sum, page) => sum + page.length, 0);
217 if (total !== undefined && rawCount < total) throw new Error("Incomplete account usage pages");
290 if (total !== undefined && rawCount < total)
291 return yield* invalid("Incomplete account usage pages");
218292 let removalsRemaining = total === undefined ? 0 : rawCount - total;
219293 let previousKeys: string[] = [];
220294 for (const events of pages) {
221295 const eventKeys =
222 removalsRemaining > 0 ? events.map((event) => accountHash(canonicalJson(event))) : [];
296 removalsRemaining > 0
297 ? yield* Effect.forEach(events, (event) => accountHash(canonicalJson(event)))
298 : [];
223299 const removalCount = Math.min(removalsRemaining, boundaryOverlap(previousKeys, eventKeys));
224300 removalsRemaining -= removalCount;
225301 previousKeys = eventKeys;
......@@ -239,7 +315,7 @@ export async function readCursorAccountUsage(
239315 value !== undefined &&
240316 (typeof value !== "number" || !Number.isFinite(value) || value < 0)
241317 ) {
242 throw new Error("Invalid account usage totals");
318 return yield* invalid("Invalid account usage totals");
243319 }
244320 }
245321 const timestampMs =
......@@ -252,7 +328,7 @@ export async function readCursorAccountUsage(
252328 typeof event.model !== "string" ||
253329 !event.model
254330 )
255 throw new Error("Invalid account usage event");
331 return yield* invalid("Invalid account usage event");
256332 if (timestampMs < sinceMs || timestampMs > endDate) continue;
257333 const totals = {
258334 uncachedInputTokens: tokens(usage.inputTokens),
......@@ -265,8 +341,8 @@ export async function readCursorAccountUsage(
265341 typeof usage.totalCents === "number" ? usage.totalCents / 100 : null;
266342 const sessionId = typeof event.conversationId === "string" ? event.conversationId : "";
267343 // No event ID is provided. Preserve identical billed rows with an occurrence index.
268 const key = accountHash(
269 JSON.stringify([timestampMs, event.model, sessionId, totals, reportedCostUsd]),
344 const key = yield* accountHash(
345 billedEventIdentity(timestampMs, event.model, sessionId, totals, reportedCostUsd),
270346 );
271347 const occurrence = occurrences.get(key) ?? 0;
272348 occurrences.set(key, occurrence + 1);
......@@ -282,19 +358,29 @@ export async function readCursorAccountUsage(
282358 dedupeKey: `cursor-account:${accountKey}:${key}:${occurrence}`,
283359 });
284360 }
285 await NodeTimersPromises.setImmediate();
361 yield* Effect.yieldNow;
286362 }
287 if (removalsRemaining !== 0) throw new Error("Inconsistent account usage boundaries");
288 return { accountKey, records, missing: false, error: null };
289 } catch {
363 if (removalsRemaining !== 0) return yield* invalid("Inconsistent account usage boundaries");
290364 return {
291365 accountKey,
292 records: [],
366 records,
293367 missing: false,
294 error: "Cursor account usage could not be read.",
295 };
296 } finally {
297 // An early exit leaves the pages requested ahead of it in flight.
298 cancel.abort();
299 }
300}
368 error: null,
369 } satisfies CursorAccountUsageReadResult;
370 });
371 return yield* read.pipe(
372 Effect.scoped,
373 Effect.timeoutOrElse({
374 duration: Duration.seconds(60),
375 orElse: () => invalid("Account usage read timed out"),
376 }),
377 Effect.catch(() =>
378 Effect.succeed({
379 accountKey,
380 records: [],
381 missing: false,
382 error: "Cursor account usage could not be read.",
383 } satisfies CursorAccountUsageReadResult),
384 ),
385 );
386});
packages/provider-opencode/src/server/usage.test.ts+35-36
......@@ -1,31 +1,31 @@
1// @effect-diagnostics nodeBuiltinImport:off - the reader under test opens real
2// SQLite databases and walks a real legacy JSON store.
3import * as NodeFSP from "node:fs/promises";
4import * as NodeOS from "node:os";
5import * as NodePath from "node:path";
6import * as NodeSqlite from "node:sqlite";
7
8import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest";
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, describe, it } from "@effect/vitest";
3import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
4import * as Context from "effect/Context";
5import * as Effect from "effect/Effect";
6import * as FileSystem from "effect/FileSystem";
7import * as Layer from "effect/Layer";
8import * as Path from "effect/Path";
9import * as SqlClient from "effect/sql/SqlClient";
910
1011import { readOpenCodeUsage } from "./usage.ts";
1112
12let dir: string;
13
14beforeEach(async () => {
15 dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "usage-reader-test-"));
16});
17
18afterEach(async () => {
19 await NodeFSP.rm(dir, { recursive: true, force: true });
20});
13/** A writable connection that stays open until the test's scope closes. */
14const openDatabase = (filename: string) =>
15 Layer.build(NodeSqliteClient.layer({ filename })).pipe(
16 Effect.map(Context.get(SqlClient.SqlClient)),
17 );
2118
2219describe("readOpenCodeUsage", () => {
23 it("counts migrated OpenCode messages once and sees subsequent WAL writes", async () => {
24 const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "opencode.db"));
25 try {
26 db.exec(
27 "PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0; CREATE TABLE message (id TEXT, session_id TEXT, data TEXT)",
28 );
20 it.effect("counts migrated OpenCode messages once and sees subsequent WAL writes", () =>
21 Effect.gen(function* () {
22 const fileSystem = yield* FileSystem.FileSystem;
23 const path = yield* Path.Path;
24 const dir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "usage-reader-test-" });
25 const db = yield* openDatabase(path.join(dir, "opencode.db"));
26 yield* db.unsafe("PRAGMA journal_mode = WAL");
27 yield* db.unsafe("PRAGMA wal_autocheckpoint = 0");
28 yield* db.unsafe("CREATE TABLE message (id TEXT, session_id TEXT, data TEXT)");
2929 const message = {
3030 id: "msg-1",
3131 sessionID: "session-1",
......@@ -35,12 +35,13 @@ describe("readOpenCodeUsage", () => {
3535 cost: 0.25,
3636 tokens: { input: 100, output: 20, reasoning: 5, cache: { read: 30, write: 10 } },
3737 };
38 const insert = db.prepare("INSERT INTO message VALUES (?, ?, ?)");
39 insert.run(message.id, message.sessionID, JSON.stringify(message));
40 const legacy = NodePath.join(dir, "storage", "message", message.sessionID);
41 await NodeFSP.mkdir(legacy, { recursive: true });
42 await NodeFSP.writeFile(NodePath.join(legacy, "msg-1.json"), JSON.stringify(message));
43 const first = await readOpenCodeUsage(dir, 0);
38 const insert = (id: string, sessionId: string, data: string) =>
39 db.unsafe("INSERT INTO message VALUES (?, ?, ?)", [id, sessionId, data]);
40 yield* insert(message.id, message.sessionID, JSON.stringify(message));
41 const legacy = path.join(dir, "storage", "message", message.sessionID);
42 yield* fileSystem.makeDirectory(legacy, { recursive: true });
43 yield* fileSystem.writeFileString(path.join(legacy, "msg-1.json"), JSON.stringify(message));
44 const first = yield* readOpenCodeUsage(dir, 0);
4445 assert.isFalse(first.error);
4546 const records = first.files.flatMap((file) => file.records);
4647 assert.strictEqual(records.length, 1);
......@@ -52,20 +53,18 @@ describe("readOpenCodeUsage", () => {
5253 reasoningTokens: 5,
5354 });
5455 assert.strictEqual(records[0]?.reportedCostUsd, 0.25);
55 insert.run(
56 yield* insert(
5657 "msg-2",
5758 message.sessionID,
5859 JSON.stringify({ ...message, id: "msg-2", time: { created: 1780000001000 } }),
5960 );
60 const next = await readOpenCodeUsage(dir, 1780000001000);
61 const next = yield* readOpenCodeUsage(dir, 1780000001000);
6162 assert.isFalse(next.error);
6263 assert.deepStrictEqual(
6364 next.files.flatMap((file) => file.records).map((record) => record.dedupeKey),
6465 ["opencode:msg-2"],
6566 );
66 assert.isAbove((await NodeFSP.stat(NodePath.join(dir, "opencode.db-wal"))).size, 0);
67 } finally {
68 db.close();
69 }
70 });
67 assert.isAbove(Number((yield* fileSystem.stat(path.join(dir, "opencode.db-wal"))).size), 0);
68 }).pipe(Effect.provide(NodeServices.layer)),
69 );
7170});
packages/provider-opencode/src/server/usage.ts+107-61
......@@ -1,16 +1,10 @@
1// node:sqlite reads live OpenCode databases; Node fs walks legacy JSON history.
2// @effect-diagnostics nodeBuiltinImport:off
31/**
42 * Usage history for OpenCode, read from its SQLite databases and legacy JSON
53 * message store under each data directory.
64 *
75 * @module provider-opencode/server/usage
86 */
9import * as NodeFSP from "node:fs/promises";
107import * as NodeOS from "node:os";
11import * as NodePath from "node:path";
12import * as NodeSqlite from "node:sqlite";
13import * as NodeTimersPromises from "node:timers/promises";
148
159import { expandHomePath } from "@t3tools/provider-core/server/pathExpansion";
1610import {
......@@ -19,9 +13,13 @@ import {
1913 type UsageRecord,
2014} from "@t3tools/provider-core/server/usage";
2115import { HostProcessEnvironment } from "@t3tools/shared/hostProcess";
16import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
2217import * as Effect from "effect/Effect";
18import * as Exit from "effect/Exit";
2319import * as FileSystem from "effect/FileSystem";
2420import * as Path from "effect/Path";
21import type * as PlatformError from "effect/PlatformError";
22import * as SqlClient from "effect/sql/SqlClient";
2523
2624import type { OpenCodeSettings } from "../settings.ts";
2725
......@@ -93,11 +91,23 @@ export interface OpenCodeUsageReadResult {
9391 readonly error: boolean;
9492}
9593
94const isNotFound = (cause: PlatformError.PlatformError) => cause.reason._tag === "NotFound";
95
96/** A regular file or directory, never a symlink to one. */
97const entryType = Effect.fn("entryType")(function* (path: string) {
98 const fileSystem = yield* FileSystem.FileSystem;
99 const link = yield* Effect.exit(fileSystem.readLink(path));
100 if (Exit.isSuccess(link)) return "SymbolicLink" as const;
101 return (yield* fileSystem.stat(path)).type;
102});
103
96104/** Reads current SQLite and pre-migration JSON stores without modifying either. */
97export async function readOpenCodeUsage(
105export const readOpenCodeUsage = Effect.fn("readOpenCodeUsage")(function* (
98106 root: string,
99107 sinceMs: number,
100): Promise<OpenCodeUsageReadResult> {
108): Effect.fn.Return<OpenCodeUsageReadResult, never, FileSystem.FileSystem | Path.Path> {
109 const fileSystem = yield* FileSystem.FileSystem;
110 const path = yield* Path.Path;
101111 const files: { path: string; records: UsageRecord[] }[] = [];
102112 const seen = new Set<string>();
103113 let found = false;
......@@ -111,49 +121,69 @@ export async function readOpenCodeUsage(
111121 records.push(record);
112122 };
113123
114 let databases: string[] = [];
115 try {
116 databases = (await NodeFSP.readdir(root, { withFileTypes: true }))
117 .filter((entry) => entry.isFile() && /^opencode(?:-[a-zA-Z0-9_-]+)?\.db$/.test(entry.name))
118 .map((entry) => entry.name)
119 .sort((a, b) => (a === "opencode.db" ? -1 : b === "opencode.db" ? 1 : a.localeCompare(b)));
120 } catch (cause) {
121 if (object(cause).code !== "ENOENT") error = true;
122 }
124 const databases = yield* fileSystem.readDirectory(root).pipe(
125 Effect.flatMap((names) =>
126 Effect.filter(
127 names.filter((name) => /^opencode(?:-[a-zA-Z0-9_-]+)?\.db$/.test(name)),
128 // One database removed between listing and stat must not hide the rest.
129 (name) =>
130 entryType(path.join(root, name)).pipe(
131 Effect.map((type) => type === "File"),
132 Effect.catchTags({
133 PlatformError: (cause) =>
134 isNotFound(cause) ? Effect.succeed(false) : Effect.fail(cause),
135 }),
136 ),
137 ),
138 ),
139 Effect.map((names) =>
140 names.sort((a, b) =>
141 a === "opencode.db" ? -1 : b === "opencode.db" ? 1 : a.localeCompare(b),
142 ),
143 ),
144 Effect.catchTags({
145 PlatformError: (cause) => {
146 if (!isNotFound(cause)) error = true;
147 return Effect.succeed([]);
148 },
149 }),
150 );
123151 for (const name of databases) {
124152 found = true;
125 const file = { path: NodePath.join(root, name), records: [] as UsageRecord[] };
153 const file = { path: path.join(root, name), records: [] as UsageRecord[] };
126154 files.push(file);
127 let database: NodeSqlite.DatabaseSync | undefined;
128 try {
129 database = new NodeSqlite.DatabaseSync(NodePath.join(root, name), { readOnly: true });
155 yield* Effect.gen(function* () {
156 const sql = yield* SqlClient.SqlClient;
130157 // A busy live provider should fail this source promptly rather than
131158 // stalling the server while SQLite waits for its writer.
132 database.exec("PRAGMA busy_timeout = 100");
159 yield* sql.unsafe("PRAGMA busy_timeout = 100");
133160 const tables = new Set(
134 database
135 .prepare("SELECT name FROM sqlite_master WHERE type = 'table'")
136 .all()
137 .map((row) => row.name),
161 (yield* sql.unsafe<{ readonly name: unknown }>(
162 "SELECT name FROM sqlite_master WHERE type = 'table'",
163 )).map((row) => row.name),
138164 );
139165 if (!tables.has("message") && !tables.has("session_message")) error = true;
140166 for (const table of ["message", "session_message"] as const) {
141167 if (!tables.has(table)) continue;
142168 const columns = new Set(
143 database
144 .prepare(`PRAGMA table_info(${table})`)
145 .all()
146 .map((row) => row.name),
169 (yield* sql.unsafe<{ readonly name: unknown }>(`PRAGMA table_info(${table})`)).map(
170 (row) => row.name,
171 ),
147172 );
148173 const timestamp = columns.has("time_created") ? "time_created" : "NULL";
149174 const predicates = table === "session_message" ? ["type = 'assistant'"] : [];
150175 if (timestamp !== "NULL") predicates.push("time_created >= ?");
151176 const where = predicates.length > 0 ? ` WHERE ${predicates.join(" AND ")}` : "";
152 const statement = database.prepare(
177 const rows = yield* sql.unsafe<{
178 readonly id: unknown;
179 readonly session_id: unknown;
180 readonly data: unknown;
181 readonly created: unknown;
182 }>(
153183 `SELECT id, session_id, data, ${timestamp} AS created FROM ${table}${where}`,
184 timestamp === "NULL" ? [] : [sinceMs],
154185 );
155 let count = 0;
156 for (const row of statement.iterate(...(timestamp === "NULL" ? [] : [sinceMs]))) {
186 for (const [index, row] of rows.entries()) {
157187 append(
158188 file.records,
159189 parseOpenCodeMessage(text(row.data), {
......@@ -162,48 +192,64 @@ export async function readOpenCodeUsage(
162192 ...(typeof row.created === "number" ? { timestampMs: row.created } : {}),
163193 }),
164194 );
165 if (++count % 256 === 0) await NodeTimersPromises.setImmediate();
195 if (index % 256 === 255) yield* Effect.yieldNow;
166196 }
167197 }
168 } catch {
169 error = true;
170 } finally {
171 database?.close();
172 }
198 }).pipe(
199 Effect.provide(NodeSqliteClient.layer({ filename: file.path, readonly: true })),
200 Effect.catchTags({
201 SqlError: () => {
202 error = true;
203 return Effect.void;
204 },
205 }),
206 );
173207 }
174208
175209 // Do not follow symlinks, including cycles. Database records win over their
176210 // old JSON copies when OpenCode has migrated a store in place.
177 const directories = [NodePath.join(root, "storage", "message")];
211 const directories = [path.join(root, "storage", "message")];
178212 while (directories.length > 0) {
179213 const directory = directories.pop()!;
180 try {
181 for (const entry of await NodeFSP.readdir(directory, { withFileTypes: true })) {
182 const path = NodePath.join(directory, entry.name);
183 if (entry.isDirectory()) {
184 directories.push(path);
185 } else if (entry.isFile() && entry.name.endsWith(".json")) {
214 yield* Effect.gen(function* () {
215 for (const name of yield* fileSystem.readDirectory(directory)) {
216 const entry = path.join(directory, name);
217 const type = yield* entryType(entry).pipe(
218 Effect.catchTags({
219 PlatformError: (cause) =>
220 isNotFound(cause) ? Effect.succeed(null) : Effect.fail(cause),
221 }),
222 );
223 if (type === "Directory") {
224 directories.push(entry);
225 } else if (type === "File" && name.endsWith(".json")) {
186226 found = true;
187 const id = entry.name.slice(0, -5);
227 const id = name.slice(0, -5);
188228 if (seen.has(`opencode:${id}`)) continue;
189 const file = { path, records: [] as UsageRecord[] };
229 const file = { path: entry, records: [] as UsageRecord[] };
190230 files.push(file);
191 try {
192 append(
193 file.records,
194 parseOpenCodeMessage(await NodeFSP.readFile(path, "utf8"), { id }),
195 );
196 } catch (cause) {
197 if (object(cause).code !== "ENOENT") error = true;
198 }
231 yield* fileSystem.readFileString(entry).pipe(
232 Effect.map((source) => append(file.records, parseOpenCodeMessage(source, { id }))),
233 Effect.catchTags({
234 PlatformError: (cause) => {
235 if (!isNotFound(cause)) error = true;
236 return Effect.void;
237 },
238 }),
239 );
199240 }
200241 }
201 } catch (cause) {
202 if (object(cause).code !== "ENOENT") error = true;
203 }
242 }).pipe(
243 Effect.catchTags({
244 PlatformError: (cause) => {
245 if (!isNotFound(cause)) error = true;
246 return Effect.void;
247 },
248 }),
249 );
204250 }
205251 return { files, missing: !found && !error, error };
206}
252});
207253
208254/**
209255 * The data directories to read: `OPENCODE_DATA_DIR` (comma-separated) or the
......@@ -244,7 +290,7 @@ export const openCodeUsageReader: ProviderUsageReader<OpenCodeSettings, OpenCode
244290 return yield* Effect.forEach(
245291 roots,
246292 (dir) =>
247 Effect.promise(() => readOpenCodeUsage(dir, windowStartMs)).pipe(
293 readOpenCodeUsage(dir, windowStartMs).pipe(
248294 Effect.map((result) => ({
249295 dir,
250296 files: result.missing && !result.error ? null : result.files,