shumengya/sproutclaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(coding-agent): stream large session files

Browse Source 
closes #5231
This commit is contained in:
Mario Zechner
2026-05-30 21:21:42 +02:00
parent dbb9911a54
commit 3911d6f5cd
3 changed files with 130 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/coding-agent/CHANGELOG.md
View File

@@ -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

168
packages/coding-agent/src/core/session-manager.ts
View File

@@ -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<SessionInfo | null> {
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<SessionInfo | null> {
}
}
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 {

32
packages/coding-agent/test/session-manager/file-operations.test.ts
View File

@@ -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", () => {