1// @effect-diagnostics nodeBuiltinImport:off
2// Cold windows span ~1.4 GB/1,500 files; node:fs buffer streaming is ~10x cheaper than materializing files.
3import * as NodeFS from "node:fs";
4import * as NodeFSP from "node:fs/promises";
5import * as NodePath from "node:path";
6import * as NodeStringDecoder from "node:string_decoder";
7import * as NodeUtil from "node:util";
8import * as NodeZlib from "node:zlib";
9
10import type {
11 SelectedFields,
12 TranscriptUsageFormat,
13 UsageRecord,
14} from "@t3tools/provider-core/server/usage";
15import * as Option from "effect/Option";
16import * as Schema from "effect/Schema";
17
18import { createTranscriptJsonReader } from "../project/AgentSessionJson.ts";
19
20export interface TranscriptFile {
21 readonly path: string;
22 readonly size: number;
23 readonly mtimeMs: number;
24}
25
26/** Resume only if bytes before resumeOffset match the guard hash; rewritten files would corrupt totals. */
27export interface TranscriptParsePosition {
28 readonly resumeOffset: number;
29 readonly guardLength: number;
30 readonly guardHash: number;
31 readonly state: unknown;
32}
33
34export interface TranscriptParseResult {
35 readonly records: readonly UsageRecord[];
36 /** Unterminated tail records stay outside records/position so the next scan rereads them. */
37 readonly tailRecords: readonly UsageRecord[];
38 /** Includes tentative tail diagnostics; position.state only includes committed lines. */
39 readonly malformedRecords: number;
40 readonly position: TranscriptParsePosition;
41 readonly resumed: boolean;
42}
43
44export const GUARD_LENGTH = 64;
45const STREAMING_THRESHOLD_BYTES = 8 * 1024 * 1024;
46const READ_CHUNK_BYTES = 256 * 1024;
47const NEWLINE = 0x0a;
48const STAT_CONCURRENCY = 32;
49const WALK_CONCURRENCY = 16;
50// Measured about 3x faster than the promise API's stat across 20,000 transcripts.
51const statFile = NodeUtil.promisify(NodeFS.stat);
52const CARRIAGE_RETURN = 0x0d;
53
54function selectUsageFields(fields: SelectedFields) {
55 return (path: ReadonlyArray<string | number | null>): boolean => {
56 let selected: true | SelectedFields = fields;
57 for (const key of path) {
58 if (selected === true) return true;
59 if (typeof key !== "string" || !Object.hasOwn(selected, key)) return false;
60 selected = selected[key]!;
61 }
62 return true;
63 };
64}
65
66function fnv1a(buffer: Buffer): number {
67 let hash = 0x811c9dc5;
68 for (let index = 0; index < buffer.length; index += 1) {
69 hash ^= buffer[index]!;
70 hash = Math.imul(hash, 0x01000193);
71 }
72 return hash >>> 0;
73}
74
75/** Skips Grok logs without usage; preserves readdir order for first-seen deduplication. */
76export async function listTranscriptFiles(
77 root: string,
78 sinceMs: number,
79 options?: { readonly fileName?: string },
80): Promise<{
81 readonly files: readonly TranscriptFile[];
82 readonly walked: readonly string[];
83 readonly failedPaths: number;
84}> {
85 const fileName = options?.fileName;
86 let failedPaths = 0;
87 let reading = 0;
88 const waiting: Array<() => void> = [];
89 // Sibling directories read together; each returns its own list so the result keeps readdir order.
90 const walk = async (dir: string): Promise<readonly string[]> => {
91 if (reading < WALK_CONCURRENCY) reading += 1;
92 else await new Promise<void>((resume) => waiting.push(resume));
93 let entries;
94 try {
95 entries = await NodeFSP.readdir(dir, { withFileTypes: true });
96 } catch {
97 failedPaths += 1;
98 return [];
99 } finally {
100 // A waiting directory takes over this slot, so the count only drops when none waits.
101 const next = waiting.shift();
102 if (next === undefined) reading -= 1;
103 else next();
104 }
105 const found = await Promise.all(
106 entries.map((entry) => {
107 const child = NodePath.join(dir, entry.name);
108 if (entry.isDirectory()) return walk(child);
109 return (
110 fileName !== undefined
111 ? entry.name === fileName
112 : entry.name.endsWith(".jsonl") || entry.name.endsWith(".jsonl.zst")
113 )
114 ? [child]
115 : [];
116 }),
117 );
118 return found.flat();
119 };
120 const candidates = await walk(root);
121
122 const found: Array<TranscriptFile | undefined> = Array.from({ length: candidates.length });
123 const queue = candidates.entries();
124 const statQueued = async (): Promise<void> => {
125 for (const [index, path] of queue) {
126 try {
127 const stats = await statFile(path);
128 if (stats.mtimeMs >= sinceMs) {
129 found[index] = { path, size: stats.size, mtimeMs: stats.mtimeMs };
130 }
131 } catch (error) {
132 if (!(error instanceof Error && "code" in error && error.code === "ENOENT")) {
133 failedPaths += 1;
134 }
135 }
136 }
137 };
138 await Promise.all(
139 Array.from({ length: Math.min(STAT_CONCURRENCY, candidates.length) }, statQueued),
140 );
141 return { files: found.filter((file) => file !== undefined), walked: candidates, failedPaths };
142}
143
144export async function readDirectoryVolumeId(path: string): Promise<string> {
145 try {
146 const stats = await NodeFSP.stat(path);
147 return `${stats.dev}:${stats.ino}`;
148 } catch {
149 return "";
150 }
151}
152
153async function guardMatches(
154 handle: NodeFSP.FileHandle,
155 position: TranscriptParsePosition,
156): Promise<boolean> {
157 if (position.guardLength <= 0 || position.guardLength > GUARD_LENGTH) return false;
158 try {
159 const window = Buffer.alloc(position.guardLength);
160 const { bytesRead } = await handle.read(
161 window,
162 0,
163 position.guardLength,
164 position.resumeOffset - position.guardLength,
165 );
166 return bytesRead === position.guardLength && fnv1a(window) === position.guardHash;
167 } catch {
168 return false;
169 }
170}
171
172/** Plain files reuse one buffer, so a chunk is only valid until the next one is requested. */
173async function* readChunks(
174 handle: NodeFSP.FileHandle,
175 start: number,
176 compressed: boolean,
177): AsyncGenerator<Buffer> {
178 if (compressed) {
179 // Compressed snapshots cache a full parse; decoded byte offsets cannot
180 // resume the compressed input.
181 yield* handle
182 .createReadStream({ start, autoClose: true, highWaterMark: READ_CHUNK_BYTES })
183 .compose(NodeZlib.createZstdDecompress());
184 return;
185 }
186 const buffer = Buffer.allocUnsafe(READ_CHUNK_BYTES);
187 for (let position = start; ;) {
188 const { bytesRead } = await handle.read(buffer, 0, buffer.length, position);
189 if (bytesRead === 0) return;
190 position += bytesRead;
191 yield buffer.subarray(0, bytesRead);
192 }
193}
194
195// Match markers in one latin1 alternation; one character per byte preserves exact ASCII offsets.
196export function transcriptLineGate(markers: readonly [string, ...string[]]) {
197 if (markers.length === 1) {
198 const marker = Buffer.from(markers[0]);
199 return (line: Buffer) => line.includes(marker);
200 }
201 const pattern = new RegExp(
202 markers.map((marker) => marker.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")).join("|"),
203 );
204 return (line: Buffer) => pattern.test(line.toString("latin1"));
205}
206
207/** Returns null on read failure; caching it as empty would hide usage until size/mtime changes. */
208export async function readTranscriptRecords<State>(
209 filePath: string,
210 format: TranscriptUsageFormat<State>,
211 resumeFrom?: TranscriptParsePosition,
212 options?: { readonly streamingThresholdBytes?: number },
213): Promise<TranscriptParseResult | null> {
214 const streamingThresholdBytes = options?.streamingThresholdBytes ?? STREAMING_THRESHOLD_BYTES;
215 const compressed = filePath.endsWith(".jsonl.zst");
216 let handle: NodeFSP.FileHandle;
217 try {
218 handle = await NodeFSP.open(filePath, "r");
219 } catch {
220 return null;
221 }
222
223 try {
224 const codec = format.state;
225 const decodeState = codec === undefined ? undefined : Schema.decodeUnknownOption(codec.schema);
226 const encodeState = codec === undefined ? undefined : Schema.encodeSync(codec.schema);
227 let state = codec === undefined ? (undefined as State) : codec.initial();
228 let resumed = false;
229 let start = 0;
230 const savedState =
231 resumeFrom === undefined || decodeState === undefined || resumeFrom.state === null
232 ? Option.none()
233 : decodeState(resumeFrom.state);
234 if (
235 !compressed &&
236 resumeFrom !== undefined &&
237 resumeFrom.resumeOffset > 0 &&
238 (codec === undefined || Option.isSome(savedState)) &&
239 (await guardMatches(handle, resumeFrom))
240 ) {
241 if (Option.isSome(savedState)) state = savedState.value;
242 start = resumeFrom.resumeOffset;
243 resumed = true;
244 }
245
246 const toLineString = (lineBuffer: Buffer): string => {
247 const content =
248 lineBuffer.length > 0 && lineBuffer[lineBuffer.length - 1] === CARRIAGE_RETURN
249 ? lineBuffer.subarray(0, -1)
250 : lineBuffer;
251 return content.toString("utf8");
252 };
253
254 const mightCarryUsage = transcriptLineGate(format.usageMarkers);
255 const parseLine = (lineBuffer: Buffer, lineState: State, out: UsageRecord[]): void => {
256 if (!mightCarryUsage(lineBuffer)) return;
257 for (const record of format.parseLine(toLineString(lineBuffer), lineState)) out.push(record);
258 };
259
260 const records: UsageRecord[] = [];
261 let resumeOffset = start;
262 let scanOffset = start;
263 let pendingChunks: Buffer[] = [];
264 let pendingBytes = 0;
265 let streaming: ReturnType<typeof createTranscriptJsonReader> | undefined;
266 let decoder: NodeStringDecoder.StringDecoder | undefined;
267 const selectPath = selectUsageFields(format.selectFields);
268
269 const append = (segment: Buffer) => {
270 if (!streaming && pendingBytes + segment.length <= streamingThresholdBytes) {
271 // Copy: the next read overwrites the chunk this segment points into.
272 if (segment.length > 0) pendingChunks.push(Buffer.from(segment));
273 pendingBytes += segment.length;
274 return;
275 }
276 if (!streaming) {
277 // Usage has no import-history budget; retain selected fields regardless of surrounding tool payload size.
278 streaming = createTranscriptJsonReader(() => {}, selectPath, { maxDepth: Infinity });
279 decoder = new NodeStringDecoder.StringDecoder("utf8");
280 for (const pending of pendingChunks) streaming.write(decoder.write(pending));
281 pendingChunks = [];
282 pendingBytes = 0;
283 }
284 streaming.write(decoder!.write(segment));
285 };
286 const finish = (lineState: State, out: UsageRecord[]) => {
287 if (streaming) {
288 streaming.write(decoder!.end());
289 const projected = streaming.finish();
290 for (const record of format.parseProjected(projected, lineState)) out.push(record);
291 } else if (pendingBytes > 0) {
292 const line =
293 pendingChunks.length === 1
294 ? pendingChunks[0]!
295 : Buffer.concat(pendingChunks, pendingBytes);
296 parseLine(line, lineState, out);
297 }
298 pendingChunks = [];
299 pendingBytes = 0;
300 streaming = undefined;
301 decoder = undefined;
302 };
303 for await (const chunk of readChunks(handle, start, compressed)) {
304 let lineStart = 0;
305 while (lineStart < chunk.length) {
306 const newlineIndex = chunk.indexOf(NEWLINE, lineStart);
307 if (newlineIndex === -1) {
308 append(chunk.subarray(lineStart));
309 break;
310 }
311 if (!streaming && pendingBytes === 0) {
312 parseLine(chunk.subarray(lineStart, newlineIndex), state, records);
313 } else {
314 append(chunk.subarray(lineStart, newlineIndex));
315 finish(state, records);
316 }
317 lineStart = newlineIndex + 1;
318 resumeOffset = scanOffset + lineStart;
319 }
320 scanOffset += chunk.length;
321 }
322
323 // Parse unfinished tails against copied state so the next scan can replay from resumeOffset.
324 const encodedState = encodeState === undefined ? null : encodeState(state);
325 const tailRecords: UsageRecord[] = [];
326 const tailState =
327 decodeState === undefined ? state : Option.getOrThrow(decodeState(encodedState));
328 finish(tailState, tailRecords);
329
330 const guardLength = compressed ? 0 : Math.min(GUARD_LENGTH, resumeOffset);
331 let guardHash = 0;
332 if (guardLength > 0) {
333 const window = Buffer.alloc(guardLength);
334 await handle.read(window, 0, guardLength, resumeOffset - guardLength);
335 guardHash = fnv1a(window);
336 }
337
338 return {
339 records,
340 tailRecords,
341 malformedRecords: format.readMalformedRecords?.(tailState) ?? 0,
342 position: {
343 resumeOffset: compressed ? 0 : resumeOffset,
344 guardLength,
345 guardHash,
346 state: encodedState,
347 },
348 resumed,
349 };
350 } catch {
351 return null;
352 } finally {
353 await handle.close().catch(() => undefined);
354 }
355}