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