authorgravatar for 91240327+shivamhwp@users.noreply.github.comshivam <91240327+shivamhwp@users.noreply.github.com> 2026-09-29 06:30:52+05:30
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-29 01:00:52+00:00
logadfc9240eab7c5db376a87591b497d8144289257
tree082cc8494b41229d4055d0587ac50a5037f27331
parent38f3c62304bfe79ea6c9f75a17afb732b555f48d
signature Signed by PGP key B5690EEEBB952194

fix(server): preserve usage in oversized transcript records (#13650)

6 files changed, 636 insertions(+), 37 deletions(-)

apps/server/src/project/AgentSessionJson.test.ts created+60
......@@ -0,0 +1,60 @@
1import { describe, expect, it } from "@effect/vitest";
2
3import { createTranscriptJsonReader, TranscriptJsonLimitError } from "./AgentSessionJson.ts";
4
5function read(
6 json: string,
7 size: number,
8 select: (path: ReadonlyArray<string | number | null>) => boolean = () => true,
9) {
10 const reader = createTranscriptJsonReader(() => {}, select);
11 for (let offset = 0; offset < json.length; offset += size)
12 reader.write(json.slice(offset, offset + size));
13 return reader.finish();
14}
15
16describe("transcript JSON projection", () => {
17 it.each([1, 2, 7, 64, 1024])("matches JSON.parse across %i-character boundaries", (size) => {
18 for (const text of [
19 '{"a":1,"a":2,"b":"before","b":"after"}',
20 '{"a":{"x":1},"a":{"y":2},"b":[],"c":{}}',
21 '{"a":[null,true,false,1,-2.3e4,"😀\\u0061\\\\\\\"",{},[],[1,2]]}',
22 '{"a":"s","a":null,"b":null,"b":"s","c":0,"c":false}',
23 '{"__proto__":{"polluted":true},"constructor":1,"__proto__":2}',
24 ])
25 expect(read(text, size)).toEqual(JSON.parse(text));
26 });
27
28 it("projects siblings and array elements without merging repeated parent objects", () => {
29 const text =
30 '{"message":{"usage":{"input":100},"content":"large"},"message":{"usage":{"output":5},"content":[1,2]},"rows":[{"keep":1,"drop":2},{"keep":3}],"drop":{"keep":4}}';
31 const projected = read(text, 1, (path) => {
32 if (path[0] === "drop") return false;
33 return !path.includes("content") && !path.includes("drop");
34 });
35 expect(projected).toEqual({
36 message: { usage: { output: 5 } },
37 rows: [{ keep: 1 }, { keep: 3 }],
38 });
39 });
40
41 it.each(['{"a":', '{"a":1} trailing', '{"a":1}{"a":2}', '{"a":"bad\\x"}', '{"a":[1,]}'])(
42 "rejects malformed input %s",
43 (text) => {
44 expect(read(text, 1)).toBeUndefined();
45 },
46 );
47
48 it("retains the import allocation and depth limits", () => {
49 const limited = createTranscriptJsonReader(
50 () => {
51 throw new TranscriptJsonLimitError("budget");
52 },
53 () => true,
54 );
55 expect(() => limited.write('{"a":1}')).toThrow(TranscriptJsonLimitError);
56 expect(() => read("[".repeat(129) + "0" + "]".repeat(129), 10)).toThrow(
57 TranscriptJsonLimitError,
58 );
59 });
60});
apps/server/src/project/AgentSessionJson.ts+59-13
......@@ -1,7 +1,6 @@
11import * as SchemaAST from "effect/SchemaAST";
2import { isMany, none, type Many } from "stream-chain/defs.js";
2import { none, type Many } from "stream-chain/defs.js";
33import { Assembler } from "stream-json/core/assembler.js";
4import { filter } from "stream-json/core/filters/filter.js";
54import * as StreamJson from "stream-json/core/parser.js";
65import type { ParserOptions, Token } from "stream-json/core/parser.js";
76
......@@ -55,6 +54,7 @@ export class TranscriptJsonLimitError extends Error {}
5554export function createTranscriptJsonReader(
5655 reserve: (bytes: number) => void,
5756 selectPath: (path: JsonPath) => boolean,
57 options?: { readonly maxDepth?: number },
5858) {
5959 // The synchronous tokenizer is exported at runtime in 3.6.0, but omitted
6060 // from its bundled types. Unlike parser(), it does not wrap tokens in an
......@@ -65,9 +65,6 @@ export function createTranscriptJsonReader(
6565 ) => (input: string | typeof none) => Many<Token> | typeof none;
6666 };
6767 const tokenize = jsonParser({ packValues: false });
68 const select = filter({ filter: selectPath, streamKeys: false }) as (
69 input: Token | typeof none,
70 ) => Token | Many<Token> | typeof none;
7168 const assembler = new Assembler();
7269 let key: string | null = null;
7370 let value = "";
......@@ -100,13 +97,62 @@ export function createTranscriptJsonReader(
10097 assembler.consume(token);
10198 }
10299 };
100 // Forward actual selected keys instead of reconstructing them from path
101 // changes: adjacent duplicate keys have the same path but JSON.parse keeps
102 // the last value. Reconstructing paths can silently retain the first value.
103 const stack: Array<{ path: JsonPath; key: string | number | null; selected: boolean }> = [];
104 let selectedValue = false;
105 const startValue = () => {
106 const parent = stack.at(-1);
107 const path = parent?.selected ? [...parent.path, parent.key] : [];
108 const selected = (parent?.selected ?? true) && selectPath(path);
109 if (selected && typeof parent?.key === "string") {
110 assemble({ name: "keyValue", value: parent.key });
111 }
112 return { path, selected };
113 };
114 const endValue = () => {
115 const parent = stack.at(-1);
116 if (parent && typeof parent.key === "number") parent.key++;
117 };
103118 const selectToken = (token: Token | typeof none) => {
104 const selected = select(token);
105 if (selected === none) return;
106 if (isMany(selected)) {
107 for (const item of selected.values) assemble(item);
108 } else {
109 assemble(selected);
119 if (token === none) return;
120 switch (token.name) {
121 case "keyValue": {
122 const parent = stack.at(-1);
123 if (parent) parent.key = token.value;
124 return;
125 }
126 case "startObject":
127 case "startArray": {
128 const frame = startValue();
129 stack.push({ ...frame, key: token.name === "startArray" ? 0 : null });
130 if (frame.selected) assemble(token);
131 return;
132 }
133 case "endObject":
134 case "endArray":
135 if (stack.pop()?.selected) assemble(token);
136 endValue();
137 return;
138 case "startString":
139 case "startNumber":
140 selectedValue = startValue().selected;
141 if (selectedValue) assemble(token);
142 return;
143 case "endString":
144 case "endNumber":
145 if (selectedValue) assemble(token);
146 endValue();
147 return;
148 case "nullValue":
149 case "trueValue":
150 case "falseValue":
151 if (startValue().selected) assemble(token);
152 endValue();
153 return;
154 default:
155 if (selectedValue) assemble(token);
110156 }
111157 };
112158 const consume = (input: string | typeof none) => {
......@@ -116,8 +162,8 @@ export function createTranscriptJsonReader(
116162 if (tokens === none) return;
117163 for (const token of tokens.values) {
118164 if (token.name === "startObject" || token.name === "startArray") {
119 if (++depth > 128)
120 throw new TranscriptJsonLimitError("Transcript JSON nesting exceeds 128 levels");
165 if (++depth > (options?.maxDepth ?? 128))
166 throw new TranscriptJsonLimitError("Transcript JSON nesting exceeds the depth limit");
121167 } else if (token.name === "endObject" || token.name === "endArray") {
122168 if (--depth === 0) complete = true;
123169 }
apps/server/src/usage/UsageService.test.ts+53
......@@ -645,6 +645,59 @@ describe("UsageService", () => {
645645 }).pipe(Effect.scoped),
646646 );
647647
648 it.live(
649 "keeps large-record totals and costs exact through append, dedupe, restart and cleanup",
650 () =>
651 Effect.gen(function* () {
652 const { transcript, settings, home } = yield* setup;
653 const large = claudeLine(1, 9900).replace(
654 '"message":',
655 '"padding":' + encodeUnknownJsonString("x".repeat(9 * 1024 * 1024)) + ',"message":',
656 );
657 yield* Effect.promise(() => NodeFSP.writeFile(transcript, large));
658 yield* Effect.gen(function* () {
659 const service = yield* UsageService.make;
660 const first = yield* service.readSummary(WINDOW);
661 assert.strictEqual(totalOutputTokens(first), 9900);
662 assert.closeTo(
663 first.buckets.reduce((sum, bucket) => sum + bucket.costUsd, 0),
664 0.4951,
665 1e-12,
666 );
667 const warm = yield* service.readSummary(WINDOW);
668 assert.deepStrictEqual(warm.buckets, first.buckets);
669 // The repeated content block has the same message/request identity.
670 yield* Effect.promise(() => NodeFSP.appendFile(transcript, large + claudeLine(2, 100)));
671 const appended = yield* service.readSummary(WINDOW);
672 assert.strictEqual(totalOutputTokens(appended), 10000);
673 assert.strictEqual(
674 appended.buckets.reduce((sum, bucket) => sum + bucket.totals.uncachedInputTokens, 0),
675 20,
676 );
677 const restarted = yield* UsageService.make;
678 const restored = yield* restarted.readSummary(WINDOW);
679 assert.deepStrictEqual(restored.buckets, appended.buckets);
680 yield* Effect.promise(() => NodeFSP.rm(transcript));
681 const afterCleanup = yield* UsageService.make;
682 assert.deepStrictEqual(
683 (yield* afterCleanup.readSummary(WINDOW)).buckets,
684 appended.buckets,
685 );
686 }).pipe(
687 Effect.provide(
688 serviceLayers({
689 prefix: "usage-service-large-record-test",
690 home,
691 settings,
692 ratesDocument: {
693 "claude-fable-5": { input_cost_per_token: 1e-5, output_cost_per_token: 5e-5 },
694 },
695 }),
696 ),
697 );
698 }).pipe(Effect.scoped),
699 );
700
648701 it.live("preserves saved tokens, costs and sessions after transcript cleanup and restart", () =>
649702 Effect.gen(function* () {
650703 const { transcript, settings, home } = yield* setup;
apps/server/src/usage/usageTranscriptReader.ts+129-24
......@@ -17,15 +17,21 @@
1717 */
1818import * as NodeFSP from "node:fs/promises";
1919import * as NodePath from "node:path";
20import * as NodeStringDecoder from "node:string_decoder";
2021
2122import type { UsageProviderKind } from "@t3tools/contracts";
2223
24import { createTranscriptJsonReader } from "../project/AgentSessionJson.ts";
25
2326import {
2427 initialCodexScanState,
2528 mightCarryUsage,
2629 parseClaudeLine,
30 parseClaudeRecord,
2731 parseCodexLine,
32 parseCodexRecord,
2833 parseGrokLine,
34 parseGrokRecord,
2935 type CodexScanState,
3036 type UsageRecord,
3137} from "./usageTranscripts.ts";
......@@ -74,9 +80,62 @@ export interface TranscriptParseResult {
7480
7581/** 64 bytes of JSONL tail is ample to distinguish a replaced file. */
7682export const GUARD_LENGTH = 64;
83// Native parsing is faster for common 1–4 MiB context/tool records. Above
84// 8 MiB, project usage without allocating the whole record. This switches
85// readers; it never discards a record because of its size.
86const STREAMING_THRESHOLD_BYTES = 8 * 1024 * 1024;
7787const NEWLINE = 0x0a;
7888const CARRIAGE_RETURN = 0x0d;
7989
90type SelectedFields = { readonly [key: string]: true | SelectedFields };
91
92// Keep the fields consumed by usageTranscripts, including reducer state and
93// dedupe/cost metadata. A selected subtree (usage) keeps future token fields.
94const USAGE_FIELDS: Record<"claude" | "codex" | "grok", SelectedFields> = {
95 claude: {
96 type: true,
97 timestamp: true,
98 requestId: true,
99 sessionId: true,
100 costUSD: true,
101 message: { id: true, model: true, usage: true },
102 },
103 codex: {
104 type: true,
105 timestamp: true,
106 payload: {
107 type: true,
108 id: true,
109 session_id: true,
110 model: true,
111 forked_from_id: true,
112 source: { subagent: { thread_spawn: { parent_thread_id: true } } },
113 info: { last_token_usage: true },
114 },
115 },
116 grok: {
117 timestamp: true,
118 params: {
119 sessionId: true,
120 _meta: { agentTimestampMs: true },
121 update: { sessionUpdate: true, prompt_id: true, usage: true },
122 },
123 },
124};
125
126function selectUsageFields(provider: UsageProviderKind) {
127 const fields = USAGE_FIELDS[provider === "codex" || provider === "grok" ? provider : "claude"];
128 return (path: ReadonlyArray<string | number | null>): boolean => {
129 let selected: true | SelectedFields = fields;
130 for (const key of path) {
131 if (selected === true) return true;
132 if (typeof key !== "string" || !Object.hasOwn(selected, key)) return false;
133 selected = selected[key]!;
134 }
135 return true;
136 };
137}
138
80139function fnv1a(buffer: Buffer): number {
81140 let hash = 0x811c9dc5;
82141 for (let index = 0; index < buffer.length; index += 1) {
......@@ -194,7 +253,9 @@ export async function readTranscriptRecords(
194253 filePath: string,
195254 provider: UsageProviderKind,
196255 resumeFrom?: TranscriptParsePosition,
256 options?: { readonly streamingThresholdBytes?: number },
197257): Promise<TranscriptParseResult | null> {
258 const streamingThresholdBytes = options?.streamingThresholdBytes ?? STREAMING_THRESHOLD_BYTES;
198259 let handle: NodeFSP.FileHandle;
199260 try {
200261 handle = await NodeFSP.open(filePath, "r");
......@@ -248,43 +309,87 @@ export async function readTranscriptRecords(
248309 };
249310
250311 const records: UsageRecord[] = [];
251 // Buffer-level line splitting rather than `readline`, because resuming
252 // needs byte-exact offsets and decoded strings cannot provide them.
253 // Newline-free chunks are collected rather than concatenated as they
254 // arrive, so a single huge line costs one copy instead of one per chunk.
312 // Byte offsets remain independent of UTF-8 decoding. Only complete lines
313 // commit the resume point; an unfinished tail is replayed on the next scan.
255314 let resumeOffset = start;
315 let scanOffset = start;
256316 let pendingChunks: Buffer[] = [];
317 let pendingBytes = 0;
318 let streaming: ReturnType<typeof createTranscriptJsonReader> | undefined;
319 let decoder: NodeStringDecoder.StringDecoder | undefined;
320 const selectPath = selectUsageFields(provider);
321
322 const append = (segment: Buffer) => {
323 if (!streaming && pendingBytes + segment.length <= streamingThresholdBytes) {
324 if (segment.length > 0) pendingChunks.push(segment);
325 pendingBytes += segment.length;
326 return;
327 }
328 if (!streaming) {
329 // Usage has no import-history budget: retain all selected usage fields,
330 // regardless of the size of the surrounding unselected tool content.
331 streaming = createTranscriptJsonReader(() => {}, selectPath, { maxDepth: Infinity });
332 decoder = new NodeStringDecoder.StringDecoder("utf8");
333 for (const pending of pendingChunks) streaming.write(decoder.write(pending));
334 pendingChunks = [];
335 pendingBytes = 0;
336 }
337 streaming.write(decoder!.write(segment));
338 };
339 const finish = (state: CodexScanState, out: UsageRecord[]) => {
340 if (streaming) {
341 streaming.write(decoder!.end());
342 const projected = streaming.finish();
343 if (provider === "grok") {
344 out.push(...parseGrokRecord(projected));
345 } else {
346 const record =
347 provider === "codex"
348 ? parseCodexRecord(projected, state)
349 : parseClaudeRecord(projected);
350 if (record !== null) out.push(record);
351 }
352 } else if (pendingBytes > 0) {
353 const line =
354 pendingChunks.length === 1
355 ? pendingChunks[0]!
356 : Buffer.concat(pendingChunks, pendingBytes);
357 parseLine(toLineString(line), state, out);
358 }
359 pendingChunks = [];
360 pendingBytes = 0;
361 streaming = undefined;
362 decoder = undefined;
363 };
257364 const stream = handle.createReadStream({
258365 start,
259366 autoClose: false,
367 highWaterMark: 256 * 1024,
260368 }) as AsyncIterable<Buffer>;
261369 for await (const chunk of stream) {
262 if (!chunk.includes(NEWLINE)) {
263 pendingChunks.push(chunk);
264 continue;
265 }
266 const buffer: Buffer =
267 pendingChunks.length === 0 ? chunk : Buffer.concat([...pendingChunks, chunk]);
268 pendingChunks = [];
269370 let lineStart = 0;
270 for (;;) {
271 const newlineIndex = buffer.indexOf(NEWLINE, lineStart);
272 if (newlineIndex === -1) break;
273 parseLine(toLineString(buffer.subarray(lineStart, newlineIndex)), codexState, records);
371 while (lineStart < chunk.length) {
372 const newlineIndex = chunk.indexOf(NEWLINE, lineStart);
373 if (newlineIndex === -1) {
374 append(chunk.subarray(lineStart));
375 break;
376 }
377 // Most lines fit in the current chunk. Avoid buffering/streaming
378 // machinery on this hot path.
379 if (!streaming && pendingBytes === 0) {
380 parseLine(toLineString(chunk.subarray(lineStart, newlineIndex)), codexState, records);
381 } else {
382 append(chunk.subarray(lineStart, newlineIndex));
383 finish(codexState, records);
384 }
274385 lineStart = newlineIndex + 1;
386 resumeOffset = scanOffset + lineStart;
275387 }
276 resumeOffset += lineStart;
277 if (lineStart < buffer.length) pendingChunks.push(buffer.subarray(lineStart));
388 scanOffset += chunk.length;
278389 }
279390
280 // A trailing segment without its newline is parsed for this result but not
281 // consumed: a writer may still be appending to it, and counting a half
282 // record now and its full form later would double count.
283391 const tailRecords: UsageRecord[] = [];
284 if (pendingChunks.length > 0) {
285 const pending = pendingChunks.length === 1 ? pendingChunks[0]! : Buffer.concat(pendingChunks);
286 if (pending.length > 0) parseLine(toLineString(pending), { ...codexState }, tailRecords);
287 }
392 finish({ ...codexState }, tailRecords);
288393
289394 const guardLength = Math.min(GUARD_LENGTH, resumeOffset);
290395 let guardHash = 0;
apps/server/src/usage/usageTranscriptStreaming.test.ts created+323
......@@ -0,0 +1,323 @@
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 {
9 readTranscriptRecords as readWithDefaultThreshold,
10 type TranscriptParsePosition,
11} from "./usageTranscriptReader.ts";
12
13// Exercise the same transition with compact fixtures. UsageService tests and
14// external 65/517 MiB fixtures also exercise the production threshold.
15const readTranscriptRecords = (
16 path: string,
17 provider: "claude" | "codex" | "grok",
18 position?: TranscriptParsePosition,
19) => readWithDefaultThreshold(path, provider, position, { streamingThresholdBytes: 256 * 1024 });
20
21let dir: string;
22beforeEach(async () => {
23 dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "usage-stream-"));
24});
25afterEach(async () => {
26 await NodeFSP.rm(dir, { recursive: true, force: true });
27});
28
29const timestamp = "2026-08-01T10:00:00Z";
30const content = '工具 output \\" usage token_count '.repeat(40_000);
31const claude = (id = "m1", output = 99) => ({
32 type: "assistant",
33 timestamp,
34 sessionId: "s1",
35 requestId: `r-${id}`,
36 costUSD: 0.25,
37 message: {
38 content: [{ type: "tool_use", input: { text: content } }],
39 id,
40 model: "claude-fable-5",
41 usage: {
42 input_tokens: 100,
43 output_tokens: output,
44 cache_read_input_tokens: 20,
45 cache_creation_input_tokens: 5,
46 speed: "fast",
47 },
48 },
49});
50const codex = [
51 { type: "session_meta", timestamp, payload: { id: "s1" } },
52 { type: "turn_context", timestamp, payload: { model: "gpt-5.6-sol" } },
53 {
54 type: "event_msg",
55 timestamp,
56 payload: {
57 type: "token_count",
58 info: {
59 last_token_usage: {
60 input_tokens: 100,
61 output_tokens: 99,
62 cached_input_tokens: 20,
63 cache_write_input_tokens: 5,
64 reasoning_output_tokens: 10,
65 },
66 },
67 },
68 },
69];
70const grok = {
71 timestamp: 1785578400,
72 params: {
73 sessionId: "s1",
74 _meta: { agentTimestampMs: 1785578400123 },
75 update: {
76 sessionUpdate: "turn_completed",
77 prompt_id: "p1",
78 usage: {
79 inputTokens: 100,
80 outputTokens: 99,
81 costUsdTicks: 2500000000,
82 modelUsage: {
83 "grok-4.5-build": {
84 inputTokens: 100,
85 outputTokens: 99,
86 cachedReadTokens: 20,
87 cacheCreationTokens: 5,
88 reasoningTokens: 10,
89 },
90 },
91 },
92 },
93 },
94};
95
96async function scan(
97 lines: readonly unknown[],
98 provider: "claude" | "codex" | "grok",
99 name = "history",
100) {
101 const path = NodePath.join(dir, `${name}.jsonl`);
102 await NodeFSP.writeFile(path, lines.map((line) => JSON.stringify(line)).join("\n") + "\n");
103 const result = await readTranscriptRecords(path, provider);
104 expect(result).not.toBeNull();
105 return result!;
106}
107
108describe("large usage records", () => {
109 it("keeps usage after large Claude tool input, including fast-mode cost and dedupe metadata", async () => {
110 const result = await scan([claude()], "claude");
111 expect(result.records).toEqual([
112 {
113 provider: "claude",
114 timestampMs: Date.parse(timestamp),
115 sessionId: "s1",
116 model: "claude-fable-5",
117 totals: {
118 uncachedInputTokens: 100,
119 outputTokens: 99,
120 cachedInputTokens: 20,
121 cacheCreationTokens: 5,
122 reasoningTokens: 0,
123 },
124 reportedCostUsd: 0.25,
125 fast: true,
126 dedupeKey: "m1:r-m1",
127 },
128 ]);
129 });
130
131 it.each(["claude", "codex", "grok"] as const)(
132 "matches ordinary %s records with large irrelevant fields in either order",
133 async (provider) => {
134 const small =
135 provider === "codex"
136 ? codex
137 : provider === "grok"
138 ? [grok]
139 : [{ ...claude(), message: { ...claude().message, content: [] } }];
140 const expected = await scan(small, provider, "small");
141 for (const first of [true, false]) {
142 const large = small.map((record) =>
143 first ? { padding: content, ...record } : { ...record, padding: content },
144 );
145 const actual = await scan(large, provider);
146 expect(actual.records).toEqual(expected.records);
147 expect(actual.position.codexState).toEqual(expected.position.codexState);
148 }
149 },
150 );
151
152 it("preserves Grok model allocation, precise timestamp and prompt identity", async () => {
153 const result = await scan([{ padding: content, ...grok }], "grok");
154 expect(result.records[0]).toMatchObject({
155 timestampMs: 1785578400123,
156 model: "grok-4.5-build",
157 sessionId: "s1",
158 totals: {
159 uncachedInputTokens: 75,
160 cachedInputTokens: 20,
161 cacheCreationTokens: 5,
162 outputTokens: 99,
163 reasoningTokens: 10,
164 },
165 dedupeKey: "s1:p1:grok-4.5-build",
166 reportedCostUsd: 0.25,
167 });
168 });
169
170 it("does not count usage-looking text or nested objects inside tool output", async () => {
171 const result = await scan(
172 [
173 { type: "user", padding: content, toolOutput: claude() },
174 { padding: content, message: { content: JSON.stringify(claude()) } },
175 { type: "assistant", timestamp, message: { model: "claude-fable-5", content: [claude()] } },
176 ],
177 "claude",
178 );
179 expect(result.records).toEqual([]);
180 });
181
182 it("replays partial UTF-8 and JSON tails, commits exact CRLF offsets, and replaces the tail once", async () => {
183 const path = NodePath.join(dir, "tail.jsonl");
184 const first = JSON.stringify(claude("first", 5)) + "\r\n";
185 const second = Buffer.from(JSON.stringify(claude("second", 7)));
186 const split = second.lastIndexOf(Buffer.from("工具")) + 1;
187 await NodeFSP.writeFile(path, Buffer.concat([Buffer.from(first), second.subarray(0, split)]));
188 const partial = await readTranscriptRecords(path, "claude");
189 expect(partial?.records.map((r) => r.totals.outputTokens)).toEqual([5]);
190 expect(partial?.tailRecords).toEqual([]);
191 expect(partial?.position.resumeOffset).toBe(Buffer.byteLength(first));
192 await NodeFSP.appendFile(path, second.subarray(split));
193 const complete = await readTranscriptRecords(path, "claude", partial!.position);
194 expect(complete?.resumed).toBe(true);
195 expect(complete?.records).toEqual([]);
196 expect(complete?.tailRecords.map((r) => r.totals.outputTokens)).toEqual([7]);
197 expect(complete?.position.resumeOffset).toBe(Buffer.byteLength(first));
198 await NodeFSP.appendFile(path, "\r\n" + JSON.stringify(claude("third", 11)) + "\n");
199 const appended = await readTranscriptRecords(path, "claude", complete!.position);
200 expect(appended?.records.map((r) => r.totals.outputTokens)).toEqual([7, 11]);
201 expect(appended?.tailRecords).toEqual([]);
202 expect(appended?.position.resumeOffset).toBe((await NodeFSP.stat(path)).size);
203 const full = await readTranscriptRecords(path, "claude");
204 expect([...partial!.records, ...appended!.records]).toEqual(full?.records);
205 });
206
207 it("preserves Codex model switches and duplicate suppression across streaming resumes", async () => {
208 const path = NodePath.join(dir, "history.jsonl");
209 const first = await scan(
210 codex.map((record) => ({ padding: content, ...record })),
211 "codex",
212 );
213 await NodeFSP.appendFile(
214 path,
215 [
216 { padding: content, ...codex[2] },
217 { type: "turn_context", payload: { padding: content, model: "gpt-6" } },
218 {
219 type: "event_msg",
220 timestamp,
221 payload: {
222 type: "token_count",
223 padding: content,
224 info: { last_token_usage: { input_tokens: 200, output_tokens: 101 } },
225 },
226 },
227 ]
228 .map((line) => JSON.stringify(line))
229 .join("\n") + "\n",
230 );
231 const result = await readTranscriptRecords(path, "codex", first.position);
232 expect(result?.resumed).toBe(true);
233 expect(result?.records).toHaveLength(1);
234 expect(result?.records[0]).toMatchObject({
235 model: "gpt-6",
236 sessionId: "s1",
237 totals: { outputTokens: 101 },
238 });
239 const full = await readTranscriptRecords(path, "codex");
240 expect([...first.records, ...result!.records]).toEqual(full?.records);
241 });
242
243 it.each(["forked_from_id", "source"])(
244 "preserves Codex fork-copy suppression from %s",
245 async (field) => {
246 const fork =
247 field === "source"
248 ? { source: { subagent: { thread_spawn: { parent_thread_id: "parent" } } } }
249 : { forked_from_id: "parent" };
250 const lines = [
251 { type: "session_meta", timestamp, payload: { padding: content, id: "child", ...fork } },
252 codex[1],
253 codex[2],
254 {
255 ...codex[2],
256 timestamp: "2026-08-01T10:00:05Z",
257 payload: {
258 type: "token_count",
259 info: { last_token_usage: { input_tokens: 100, output_tokens: 101 } },
260 },
261 },
262 ];
263 const result = await scan(lines, "codex");
264 expect(result.records).toHaveLength(1);
265 expect(result.records[0]).toMatchObject({
266 sessionId: "child",
267 totals: { outputTokens: 101 },
268 });
269 },
270 );
271
272 it("rejects malformed large lines without accepting partial usage or losing following lines", async () => {
273 const valid = JSON.stringify(claude());
274 const path = NodePath.join(dir, "broken.jsonl");
275 await NodeFSP.writeFile(
276 path,
277 [valid + " junk", valid.slice(0, -1), valid + valid, JSON.stringify(claude("good", 7))].join(
278 "\n",
279 ) + "\n",
280 );
281 const result = await readTranscriptRecords(path, "claude");
282 expect(result?.records.map((r) => r.totals.outputTokens)).toEqual([7]);
283 });
284
285 it("matches JSON.parse for reordered and escaped keys, duplicate fields, arrays and unknown key names", async () => {
286 const path = NodePath.join(dir, "odd.jsonl");
287 const small = JSON.stringify({ ...claude(), message: { ...claude().message, content: [] } });
288 const variants = [
289 small.replace('"message":', '"mess\\u0061ge":'),
290 small.replace('"costUSD":0.25', '"costUSD":5,"costUSD":0.25'),
291 small.replace('"type":"assistant"', '"type":"user","type":"assistant"'),
292 small.replace('"message":', '"__proto__":{"type":"user"},"message":'),
293 small.replace('"message":', '"message.usage":{"output_tokens":1234},"message":'),
294 ];
295 for (const line of variants) {
296 await NodeFSP.writeFile(path, line + "\n");
297 const expected = await readTranscriptRecords(path, "claude");
298 await NodeFSP.writeFile(
299 path,
300 '{"padding":' + JSON.stringify(content) + "," + line.slice(1) + "\n",
301 );
302 const actual = await readTranscriptRecords(path, "claude");
303 expect(actual?.records).toEqual(expected?.records);
304 }
305 });
306 it("reads usage after deeply nested discarded tool content", async () => {
307 const path = NodePath.join(dir, "deep.jsonl");
308 const record = JSON.stringify(claude());
309 const nested = "[".repeat(300) + "0" + "]".repeat(300);
310 await NodeFSP.writeFile(path, '{"toolOutput":' + nested + "," + record.slice(1) + "\n");
311 const result = await readTranscriptRecords(path, "claude");
312 expect(result?.records[0]?.totals.outputTokens).toBe(99);
313 });
314
315 it("keeps JSON number semantics for non-finite values without serializing them into null", async () => {
316 const path = NodePath.join(dir, "numbers.jsonl");
317 const record = JSON.stringify(claude()).replace('"costUSD":0.25', '"costUSD":1e400');
318 await NodeFSP.writeFile(path, record + "\n");
319 const result = await readTranscriptRecords(path, "claude");
320 expect(result?.records[0]?.reportedCostUsd).toBeNull();
321 expect(result?.records[0]?.totals.outputTokens).toBe(99);
322 });
323});
apps/server/src/usage/usageTranscripts.ts+12
......@@ -113,6 +113,10 @@ export function parseClaudeLine(line: string): UsageRecord | null {
113113 } catch {
114114 return null;
115115 }
116 return parseClaudeRecord(parsed);
117}
118
119export function parseClaudeRecord(parsed: unknown): UsageRecord | null {
116120 if (typeof parsed !== "object" || parsed === null) return null;
117121
118122 const record = parsed as Record<string, unknown>;
......@@ -228,6 +232,10 @@ export function parseCodexLine(line: string, state: CodexScanState): UsageRecord
228232 } catch {
229233 return null;
230234 }
235 return parseCodexRecord(parsed, state);
236}
237
238export function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord | null {
231239 if (typeof parsed !== "object" || parsed === null) return null;
232240
233241 const record = parsed as Record<string, unknown>;
......@@ -385,6 +393,10 @@ export function parseGrokLine(line: string): readonly UsageRecord[] {
385393 } catch {
386394 return [];
387395 }
396 return parseGrokRecord(parsed);
397}
398
399export function parseGrokRecord(parsed: unknown): readonly UsageRecord[] {
388400 if (typeof parsed !== "object" || parsed === null) return [];
389401
390402 const record = parsed as Record<string, unknown>;