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