diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 8326b565..6013abac 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed opening and listing very large JSONL session files by reading session entries line-by-line instead of materializing the full file as one string ([#5231](https://github.com/earendil-works/pi/issues/5231)). + ## [0.78.0] - 2026-05-29 ### New Features diff --git a/packages/coding-agent/src/core/session-manager.ts b/packages/coding-agent/src/core/session-manager.ts index 914da08d..c2bae164 100644 --- a/packages/coding-agent/src/core/session-manager.ts +++ b/packages/coding-agent/src/core/session-manager.ts @@ -4,17 +4,19 @@ import { randomUUID } from "crypto"; import { appendFileSync, closeSync, + createReadStream, existsSync, mkdirSync, openSync, readdirSync, - readFileSync, readSync, statSync, writeFileSync, } from "fs"; -import { readdir, readFile, stat } from "fs/promises"; +import { readdir, stat } from "fs/promises"; import { join, resolve } from "path"; +import { createInterface } from "readline"; +import { StringDecoder } from "string_decoder"; import { getAgentDir as getDefaultAgentDir, getSessionsDir } from "../config.ts"; import { normalizePath, resolvePath } from "../utils/paths.ts"; import { @@ -448,29 +450,57 @@ export function getDefaultSessionDir(cwd: string, agentDir: string = getDefaultA return sessionDir; } +const SESSION_READ_BUFFER_SIZE = 1024 * 1024; + +function parseSessionEntryLine(line: string): FileEntry | null { + if (!line.trim()) return null; + try { + return JSON.parse(line) as FileEntry; + } catch { + // Skip malformed lines + return null; + } +} + /** Exported for testing */ export function loadEntriesFromFile(filePath: string): FileEntry[] { const resolvedFilePath = normalizePath(filePath); if (!existsSync(resolvedFilePath)) return []; - const content = readFileSync(resolvedFilePath, "utf8"); const entries: FileEntry[] = []; - const lines = content.trim().split("\n"); + const fd = openSync(resolvedFilePath, "r"); + try { + const decoder = new StringDecoder("utf8"); + const buffer = Buffer.allocUnsafe(SESSION_READ_BUFFER_SIZE); + let pending = ""; - for (const line of lines) { - if (!line.trim()) continue; - try { - const entry = JSON.parse(line) as FileEntry; - entries.push(entry); - } catch { - // Skip malformed lines + while (true) { + const bytesRead = readSync(fd, buffer, 0, buffer.length, null); + if (bytesRead === 0) break; + + pending += decoder.write(buffer.subarray(0, bytesRead)); + let lineStart = 0; + let newlineIndex = pending.indexOf("\n", lineStart); + while (newlineIndex !== -1) { + const entry = parseSessionEntryLine(pending.slice(lineStart, newlineIndex)); + if (entry) entries.push(entry); + lineStart = newlineIndex + 1; + newlineIndex = pending.indexOf("\n", lineStart); + } + pending = pending.slice(lineStart); } + + pending += decoder.end(); + const finalEntry = parseSessionEntryLine(pending); + if (finalEntry) entries.push(finalEntry); + } finally { + closeSync(fd); } // Validate session header if (entries.length === 0) return entries; const header = entries[0]; - if (header.type !== "session" || typeof (header as any).id !== "string") { + if (header.type !== "session" || typeof (header as { id?: unknown }).id !== "string") { return []; } @@ -542,80 +572,59 @@ function extractTextContent(message: Message): string { .join(" "); } -function getLastActivityTime(entries: FileEntry[]): number | undefined { - let lastActivityTime: number | undefined; +function getMessageActivityTime(entry: SessionMessageEntry): number | undefined { + const message = entry.message; + if (!isMessageWithContent(message)) return undefined; + if (message.role !== "user" && message.role !== "assistant") return undefined; - for (const entry of entries) { - if (entry.type !== "message") continue; - - const message = (entry as SessionMessageEntry).message; - if (!isMessageWithContent(message)) continue; - if (message.role !== "user" && message.role !== "assistant") continue; - - const msgTimestamp = (message as { timestamp?: number }).timestamp; - if (typeof msgTimestamp === "number") { - lastActivityTime = Math.max(lastActivityTime ?? 0, msgTimestamp); - continue; - } - - const entryTimestamp = (entry as SessionEntryBase).timestamp; - if (typeof entryTimestamp === "string") { - const t = new Date(entryTimestamp).getTime(); - if (!Number.isNaN(t)) { - lastActivityTime = Math.max(lastActivityTime ?? 0, t); - } - } + const msgTimestamp = (message as { timestamp?: number }).timestamp; + if (typeof msgTimestamp === "number") { + return msgTimestamp; } - return lastActivityTime; -} - -function getSessionModifiedDate(entries: FileEntry[], header: SessionHeader, statsMtime: Date): Date { - const lastActivityTime = getLastActivityTime(entries); - if (typeof lastActivityTime === "number" && lastActivityTime > 0) { - return new Date(lastActivityTime); - } - - const headerTime = typeof header.timestamp === "string" ? new Date(header.timestamp).getTime() : NaN; - return !Number.isNaN(headerTime) ? new Date(headerTime) : statsMtime; + const t = new Date(entry.timestamp).getTime(); + return Number.isNaN(t) ? undefined : t; } async function buildSessionInfo(filePath: string): Promise { try { - const content = await readFile(filePath, "utf8"); - const entries: FileEntry[] = []; - const lines = content.trim().split("\n"); - - for (const line of lines) { - if (!line.trim()) continue; - try { - entries.push(JSON.parse(line) as FileEntry); - } catch { - // Skip malformed lines - } - } - - if (entries.length === 0) return null; - const header = entries[0]; - if (header.type !== "session") return null; - const stats = await stat(filePath); + let header: SessionHeader | null = null; let messageCount = 0; let firstMessage = ""; const allMessages: string[] = []; let name: string | undefined; + let lastActivityTime: number | undefined; + + const rl = createInterface({ + input: createReadStream(filePath, { encoding: "utf8" }), + crlfDelay: Infinity, + }); + + for await (const line of rl) { + const entry = parseSessionEntryLine(line); + if (!entry) continue; + + if (!header) { + if (entry.type !== "session") return null; + header = entry; + continue; + } - for (const entry of entries) { // Extract session name (use latest, including explicit clears) if (entry.type === "session_info") { - const infoEntry = entry as SessionInfoEntry; - name = infoEntry.name?.trim() || undefined; + name = entry.name?.trim() || undefined; } if (entry.type !== "message") continue; messageCount++; - const message = (entry as SessionMessageEntry).message; + const activityTime = getMessageActivityTime(entry); + if (typeof activityTime === "number") { + lastActivityTime = Math.max(lastActivityTime ?? 0, activityTime); + } + + const message = entry.message; if (!isMessageWithContent(message)) continue; if (message.role !== "user" && message.role !== "assistant") continue; @@ -628,18 +637,25 @@ async function buildSessionInfo(filePath: string): Promise { } } - const cwd = typeof (header as SessionHeader).cwd === "string" ? (header as SessionHeader).cwd : ""; - const parentSessionPath = (header as SessionHeader).parentSession; + if (!header) return null; - const modified = getSessionModifiedDate(entries, header as SessionHeader, stats.mtime); + const cwd = typeof header.cwd === "string" ? header.cwd : ""; + const parentSessionPath = header.parentSession; + const headerTime = typeof header.timestamp === "string" ? new Date(header.timestamp).getTime() : NaN; + const modified = + typeof lastActivityTime === "number" && lastActivityTime > 0 + ? new Date(lastActivityTime) + : !Number.isNaN(headerTime) + ? new Date(headerTime) + : stats.mtime; return { path: filePath, - id: (header as SessionHeader).id, + id: header.id, cwd, name, parentSessionPath, - created: new Date((header as SessionHeader).timestamp), + created: new Date(header.timestamp), modified, messageCount, firstMessage: firstMessage || "(no messages)", @@ -855,8 +871,14 @@ export class SessionManager { private _rewriteFile(): void { if (!this.persist || !this.sessionFile) return; - const content = `${this.fileEntries.map((e) => JSON.stringify(e)).join("\n")}\n`; - writeFileSync(this.sessionFile, content); + const fd = openSync(this.sessionFile, "w"); + try { + for (const entry of this.fileEntries) { + writeFileSync(fd, `${JSON.stringify(entry)}\n`); + } + } finally { + closeSync(fd); + } } isPersisted(): boolean { diff --git a/packages/coding-agent/test/session-manager/file-operations.test.ts b/packages/coding-agent/test/session-manager/file-operations.test.ts index fd771ef9..626dd52c 100644 --- a/packages/coding-agent/test/session-manager/file-operations.test.ts +++ b/packages/coding-agent/test/session-manager/file-operations.test.ts @@ -1,4 +1,5 @@ -import { mkdirSync, readFileSync, rmSync, writeFileSync } from "fs"; +import { constants as bufferConstants } from "buffer"; +import { appendFileSync, closeSync, mkdirSync, openSync, readFileSync, rmSync, writeFileSync, writeSync } from "fs"; import { tmpdir } from "os"; import { join } from "path"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; @@ -63,6 +64,35 @@ describe("loadEntriesFromFile", () => { const entries = loadEntriesFromFile(file); expect(entries).toHaveLength(2); }); + + it("opens session files larger than Node's max string length", () => { + const file = join(tempDir, "large.jsonl"); + writeFileSync( + file, + '{"type":"session","version":3,"id":"abc","timestamp":"2025-01-01T00:00:00Z","cwd":"/tmp"}\n', + ); + + const fd = openSync(file, "r+"); + try { + const newline = Buffer.from("\n"); + const stride = 16 * 1024 * 1024; + for (let offset = stride; offset <= bufferConstants.MAX_STRING_LENGTH + stride; offset += stride) { + writeSync(fd, newline, 0, newline.length, offset); + } + } finally { + closeSync(fd); + } + + appendFileSync( + file, + '{"type":"message","id":"1","parentId":null,"timestamp":"2025-01-01T00:00:01Z","message":{"role":"user","content":"hi","timestamp":1}}\n', + ); + + const sessionManager = SessionManager.open(file, tempDir); + expect(sessionManager.getSessionId()).toBe("abc"); + expect(sessionManager.getEntries()).toHaveLength(1); + expect(sessionManager.buildSessionContext().messages).toEqual([{ role: "user", content: "hi", timestamp: 1 }]); + }); }); describe("findMostRecentSession", () => {