1// @effect-diagnostics nodeBuiltinImport:off - exercise the real byte reader.
2import * as NodeFSP from "node:fs/promises";
3import * as NodeOS from "node:os";
4import * as NodePath from "node:path";
5
6import { afterEach, beforeEach, describe, expect, it } from "@effect/vitest";
7
8import { TEST_FORMATS } from "./usageTestFormats.ts";
9import {
10 readTranscriptRecords as readWithDefaultThreshold,
11 type TranscriptParsePosition,
12} from "./usageTranscriptReader.ts";
13
14const readTranscriptRecords = (
15 path: string,
16 provider: keyof typeof TEST_FORMATS,
17 position?: TranscriptParsePosition,
18) =>
19 readWithDefaultThreshold(path, TEST_FORMATS[provider], position, {
20 streamingThresholdBytes: 256 * 1024,
21 });
22
23let dir: string;
24beforeEach(async () => {
25 dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "usage-stream-"));
26});
27afterEach(async () => {
28 await NodeFSP.rm(dir, { recursive: true, force: true });
29});
30
31const timestamp = "2026-08-01T10:00:00Z";
32const content = '工具 output \\" usage token_count '.repeat(40_000);
33const claude = (id = "m1", output = 99) => ({
34 type: "assistant",
35 timestamp,
36 sessionId: "s1",
37 requestId: `r-${id}`,
38 costUSD: 0.25,
39 message: {
40 content: [{ type: "tool_use", input: { text: content } }],
41 id,
42 model: "claude-fable-5",
43 usage: {
44 input_tokens: 100,
45 output_tokens: output,
46 cache_read_input_tokens: 20,
47 cache_creation_input_tokens: 5,
48 speed: "fast",
49 },
50 },
51});
52const codex = [
53 { type: "session_meta", timestamp, payload: { id: "s1", model_provider: "openai" } },
54 { type: "turn_context", timestamp, payload: { model: "gpt-5.6-sol", turn_id: "initial-turn" } },
55 {
56 type: "event_msg",
57 timestamp,
58 payload: {
59 type: "token_count",
60 info: {
61 last_token_usage: {
62 input_tokens: 100,
63 output_tokens: 99,
64 cached_input_tokens: 20,
65 cache_write_input_tokens: 5,
66 reasoning_output_tokens: 10,
67 },
68 },
69 },
70 },
71];
72const grok = {
73 timestamp: 1785578400,
74 params: {
75 sessionId: "s1",
76 _meta: { agentTimestampMs: 1785578400123 },
77 update: {
78 sessionUpdate: "turn_completed",
79 prompt_id: "p1",
80 usage: {
81 inputTokens: 100,
82 outputTokens: 99,
83 costUsdTicks: 2500000000,
84 modelUsage: {
85 "grok-4.5-build": {
86 inputTokens: 100,
87 outputTokens: 99,
88 cachedReadTokens: 20,
89 cacheCreationTokens: 5,
90 reasoningTokens: 10,
91 },
92 },
93 },
94 },
95 },
96};
97
98async function scan(
99 lines: readonly unknown[],
100 provider: keyof typeof TEST_FORMATS,
101 name = "history",
102) {
103 const path = NodePath.join(dir, `${name}.jsonl`);
104 await NodeFSP.writeFile(path, lines.map((line) => JSON.stringify(line)).join("\n") + "\n");
105 const result = await readTranscriptRecords(path, provider);
106 expect(result).not.toBeNull();
107 return result!;
108}
109
110describe("large usage records", () => {
111 it("streams Muse completion envelopes and resumes a partial appended event", async () => {
112 const entry = {
113 schema_version: 1,
114 record_type: "event",
115 payload_type: "runtime.session",
116 payload_schema_version: 3,
117 id: "envelope-1",
118 recorded_at: 1_785_578_400_123_456,
119 stream: { kind: "session", id: "parent" },
120 payload: {
121 source_run_record_id: "inference-1",
122 event: {
123 kind: "automated_review_completed",
124 model: { model_id: "review-model" },
125 usage: {
126 input_tokens: 100,
127 cached_input_tokens: 60,
128 output_tokens: 20,
129 reasoning_tokens: 8,
130 },
131 },
132 },
133 };
134 const small = await scan([entry], "muse", "small-muse");
135 expect(small.records[0]?.totals).toMatchObject({
136 uncachedInputTokens: 40,
137 cachedInputTokens: 60,
138 });
139 const path = NodePath.join(dir, "large-muse.jsonl");
140 await NodeFSP.writeFile(path, JSON.stringify({ ...entry, discarded: { content } }) + "\n");
141 const large = await readTranscriptRecords(path, "muse");
142 expect(large?.records).toEqual(small.records);
143 const appended = JSON.stringify({
144 ...entry,
145 id: "envelope-2",
146 payload: { ...entry.payload, source_run_record_id: "inference-2" },
147 });
148 await NodeFSP.appendFile(path, appended.slice(0, -7));
149 const partial = await readTranscriptRecords(path, "muse", large!.position);
150 expect(partial?.records).toEqual([]);
151 await NodeFSP.appendFile(path, appended.slice(-7) + "\n");
152 const completed = await readTranscriptRecords(path, "muse", partial!.position);
153 expect(completed?.records).toHaveLength(1);
154 expect(completed?.records[0]?.dedupeKey).not.toBe(large?.records[0]?.dedupeKey);
155 expect(completed?.records[0]?.totals).toEqual(small.records[0]?.totals);
156 });
157
158 it("keeps Pi model, cost, and fork identity after large assistant content", async () => {
159 const entry = {
160 type: "message",
161 id: "a123abcd",
162 timestamp,
163 message: {
164 role: "assistant",
165 provider: "anthropic",
166 model: "requested",
167 responseModel: "actual",
168 content: [{ type: "text", text: content }],
169 usage: {
170 input: 100,
171 output: 99,
172 cacheRead: 20,
173 cacheWrite: 5,
174 reasoning: 10,
175 cost: { total: 0.25 },
176 },
177 },
178 };
179 const result = await scan([{ type: "session", id: "pi-session" }, entry], "pi");
180 expect(result.records).toEqual([
181 {
182 provider: "pi",
183 model: "anthropic/actual",
184 rateModel: "actual",
185 sessionId: "pi-session",
186 timestampMs: Date.parse(timestamp),
187 totals: {
188 uncachedInputTokens: 100,
189 outputTokens: 99,
190 cachedInputTokens: 20,
191 cacheCreationTokens: 5,
192 reasoningTokens: 10,
193 },
194 reportedCostUsd: 0.25,
195 speed: "standard",
196 dedupeKey: JSON.stringify(["pi", entry.id, timestamp, "anthropic/actual"]),
197 },
198 ]);
199 });
200
201 it("keeps usage after large Claude tool input, including fast-mode cost and dedupe metadata", async () => {
202 const result = await scan([claude()], "claude");
203 expect(result.records).toEqual([
204 {
205 provider: "claude",
206 timestampMs: Date.parse(timestamp),
207 sessionId: "s1",
208 model: "claude-fable-5",
209 totals: {
210 uncachedInputTokens: 100,
211 outputTokens: 99,
212 cachedInputTokens: 20,
213 cacheCreationTokens: 5,
214 reasoningTokens: 0,
215 },
216 reportedCostUsd: 0.25,
217 speed: "fast",
218 dedupeKey: "m1:r-m1",
219 },
220 ]);
221 });
222
223 it.each(["claude", "codex", "grok"] as const)(
224 "matches ordinary %s records with large irrelevant fields in either order",
225 async (provider) => {
226 const small =
227 provider === "codex"
228 ? codex
229 : provider === "grok"
230 ? [grok]
231 : [{ ...claude(), message: { ...claude().message, content: [] } }];
232 const expected = await scan(small, provider, "small");
233 for (const first of [true, false]) {
234 const large = small.map((record) =>
235 first ? { padding: content, ...record } : { ...record, padding: content },
236 );
237 const actual = await scan(large, provider);
238 expect(actual.records).toEqual(expected.records);
239 expect(actual.position.state).toEqual(expected.position.state);
240 }
241 },
242 );
243
244 it("preserves Grok model allocation, precise timestamp and prompt identity", async () => {
245 const result = await scan([{ padding: content, ...grok }], "grok");
246 expect(result.records[0]).toMatchObject({
247 timestampMs: 1785578400123,
248 model: "grok-4.5-build",
249 sessionId: "s1",
250 totals: {
251 uncachedInputTokens: 75,
252 cachedInputTokens: 20,
253 cacheCreationTokens: 5,
254 outputTokens: 99,
255 reasoningTokens: 10,
256 },
257 dedupeKey: "s1:p1:grok-4.5-build",
258 reportedCostUsd: 0.25,
259 });
260 });
261
262 it("does not count usage-looking text or nested objects inside tool output", async () => {
263 const result = await scan(
264 [
265 { type: "user", padding: content, toolOutput: claude() },
266 { padding: content, message: { content: JSON.stringify(claude()) } },
267 { type: "assistant", timestamp, message: { model: "claude-fable-5", content: [claude()] } },
268 ],
269 "claude",
270 );
271 expect(result.records).toEqual([]);
272 });
273
274 it("replays partial UTF-8 and JSON tails, commits exact CRLF offsets, and replaces the tail once", async () => {
275 const path = NodePath.join(dir, "tail.jsonl");
276 const first = JSON.stringify(claude("first", 5)) + "\r\n";
277 const second = Buffer.from(JSON.stringify(claude("second", 7)));
278 const split = second.lastIndexOf(Buffer.from("工具")) + 1;
279 await NodeFSP.writeFile(path, Buffer.concat([Buffer.from(first), second.subarray(0, split)]));
280 const partial = await readTranscriptRecords(path, "claude");
281 expect(partial?.records.map((r) => r.totals.outputTokens)).toEqual([5]);
282 expect(partial?.tailRecords).toEqual([]);
283 expect(partial?.position.resumeOffset).toBe(Buffer.byteLength(first));
284 await NodeFSP.appendFile(path, second.subarray(split));
285 const complete = await readTranscriptRecords(path, "claude", partial!.position);
286 expect(complete?.resumed).toBe(true);
287 expect(complete?.records).toEqual([]);
288 expect(complete?.tailRecords.map((r) => r.totals.outputTokens)).toEqual([7]);
289 expect(complete?.position.resumeOffset).toBe(Buffer.byteLength(first));
290 await NodeFSP.appendFile(path, "\r\n" + JSON.stringify(claude("third", 11)) + "\n");
291 const appended = await readTranscriptRecords(path, "claude", complete!.position);
292 expect(appended?.records.map((r) => r.totals.outputTokens)).toEqual([7, 11]);
293 expect(appended?.tailRecords).toEqual([]);
294 expect(appended?.position.resumeOffset).toBe((await NodeFSP.stat(path)).size);
295 const full = await readTranscriptRecords(path, "claude");
296 expect([...partial!.records, ...appended!.records]).toEqual(full?.records);
297 });
298
299 it("preserves Codex model switches and duplicate suppression across streaming resumes", async () => {
300 const path = NodePath.join(dir, "history.jsonl");
301 const first = await scan(
302 codex.map((record) => ({ padding: "x".repeat(9 * 1024 * 1024), ...record })),
303 "codex",
304 );
305 expect(first.records[0]).toMatchObject({
306 sessionId: "s1",
307 model: "gpt-5.6-sol",
308 unresolvedServiceTierTurnId: "initial-turn",
309 });
310 await NodeFSP.appendFile(
311 path,
312 [
313 { padding: content, ...codex[2] },
314 { type: "turn_context", payload: { padding: content, model: "gpt-6" } },
315 {
316 type: "event_msg",
317 timestamp,
318 payload: {
319 type: "token_count",
320 padding: content,
321 info: { last_token_usage: { input_tokens: 200, output_tokens: 101 } },
322 },
323 },
324 ]
325 .map((line) => JSON.stringify(line))
326 .join("\n") + "\n",
327 );
328 const result = await readTranscriptRecords(path, "codex", first.position);
329 expect(result?.resumed).toBe(true);
330 expect(result?.records).toHaveLength(1);
331 expect(result?.records[0]).toMatchObject({
332 model: "gpt-6",
333 sessionId: "s1",
334 totals: { outputTokens: 101 },
335 });
336 expect(result?.records[0]).not.toHaveProperty("unresolvedServiceTierTurnId");
337 const full = await readTranscriptRecords(path, "codex");
338 expect([...first.records, ...result!.records]).toEqual(full?.records);
339 });
340
341 it.each(["forked_from_id", "source"])(
342 "preserves Codex fork-copy suppression from %s",
343 async (field) => {
344 const fork =
345 field === "source"
346 ? { source: { subagent: { thread_spawn: { parent_thread_id: "parent" } } } }
347 : { forked_from_id: "parent" };
348 const lines = [
349 { type: "session_meta", timestamp, payload: { padding: content, id: "child", ...fork } },
350 codex[1],
351 codex[2],
352 {
353 ...codex[2],
354 timestamp: "2026-08-01T10:00:05Z",
355 payload: {
356 type: "token_count",
357 info: { last_token_usage: { input_tokens: 100, output_tokens: 101 } },
358 },
359 },
360 ];
361 const result = await scan(lines, "codex");
362 expect(result.records).toHaveLength(1);
363 expect(result.records[0]).toMatchObject({
364 sessionId: "child",
365 totals: { outputTokens: 101 },
366 });
367 },
368 );
369
370 it("rejects malformed large lines without accepting partial usage or losing following lines", async () => {
371 const valid = JSON.stringify(claude());
372 const path = NodePath.join(dir, "broken.jsonl");
373 await NodeFSP.writeFile(
374 path,
375 [valid + " junk", valid.slice(0, -1), valid + valid, JSON.stringify(claude("good", 7))].join(
376 "\n",
377 ) + "\n",
378 );
379 const result = await readTranscriptRecords(path, "claude");
380 expect(result?.records.map((r) => r.totals.outputTokens)).toEqual([7]);
381 });
382
383 it("matches JSON.parse for reordered and escaped keys, duplicate fields, arrays and unknown key names", async () => {
384 const path = NodePath.join(dir, "odd.jsonl");
385 const small = JSON.stringify({ ...claude(), message: { ...claude().message, content: [] } });
386 const variants = [
387 small.replace('"message":', '"mess\\u0061ge":'),
388 small.replace('"costUSD":0.25', '"costUSD":5,"costUSD":0.25'),
389 small.replace('"type":"assistant"', '"type":"user","type":"assistant"'),
390 small.replace('"message":', '"__proto__":{"type":"user"},"message":'),
391 small.replace('"message":', '"message.usage":{"output_tokens":1234},"message":'),
392 ];
393 for (const line of variants) {
394 await NodeFSP.writeFile(path, line + "\n");
395 const expected = await readTranscriptRecords(path, "claude");
396 await NodeFSP.writeFile(
397 path,
398 '{"padding":' + JSON.stringify(content) + "," + line.slice(1) + "\n",
399 );
400 const actual = await readTranscriptRecords(path, "claude");
401 expect(actual?.records).toEqual(expected?.records);
402 }
403 });
404 it("reads usage after deeply nested discarded tool content", async () => {
405 const path = NodePath.join(dir, "deep.jsonl");
406 const record = JSON.stringify(claude());
407 const nested = "[".repeat(300) + "0" + "]".repeat(300);
408 await NodeFSP.writeFile(path, '{"toolOutput":' + nested + "," + record.slice(1) + "\n");
409 const result = await readTranscriptRecords(path, "claude");
410 expect(result?.records[0]?.totals.outputTokens).toBe(99);
411 });
412
413 it("keeps JSON number semantics for non-finite values without serializing them into null", async () => {
414 const path = NodePath.join(dir, "numbers.jsonl");
415 const record = JSON.stringify(claude()).replace('"costUSD":0.25', '"costUSD":1e400');
416 await NodeFSP.writeFile(path, record + "\n");
417 const result = await readTranscriptRecords(path, "claude");
418 expect(result?.records[0]?.reportedCostUsd).toBeNull();
419 expect(result?.records[0]?.totals.outputTokens).toBe(99);
420 });
421});