diff --git a/lib/session-list-scanner.bench.mjs b/lib/session-list-scanner.bench.mjs new file mode 100644 index 0000000000..79cc000e09 --- /dev/null +++ b/lib/session-list-scanner.bench.mjs @@ -0,0 +1,81 @@ +// Run: node --expose-gc lib/session-list-scanner.bench.mjs +// Synthetic files only. "No index" does not mean an empty OS filesystem cache. +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import { appendFileSync, mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { createJiti } from "jiti"; + +const jiti = createJiti(import.meta.url); +const { SessionManager } = await jiti.import("@earendil-works/pi-coding-agent"); +const { listSessionsIncremental, resetSessionScanIndexForTests } = await jiti.import("./session-list-scanner.ts"); +if (process.argv.includes("--persisted-index")) { + assert.ok(process.env.PI_CODING_AGENT_DIR, "the parent benchmark must supply a temporary agent directory"); + const start = performance.now(); + const sessions = await listSessionsIncremental(); + console.log(JSON.stringify({ count: sessions.length, elapsed: performance.now() - start })); + process.exit(0); +} +const root = mkdtempSync(join(tmpdir(), "pi-web-scanner-bench-")); +const previousAgentDir = process.env.PI_CODING_AGENT_DIR; +process.env.PI_CODING_AGENT_DIR = root; +const count = 2000; +const timestamp = "2026-01-01T00:00:00.000Z"; +const line = (entry) => JSON.stringify(entry) + "\n"; +const message = line({ type: "message", timestamp, message: { role: "user", content: "x".repeat(1024) } }); +const samples = new Map(); +let bytes = 0; + +async function measure(label, scan) { + globalThis.gc?.(); + const start = performance.now(); + const sessions = await scan(); + const elapsed = performance.now() - start; + assert.equal(sessions.length, count); + const values = samples.get(label) ?? []; + values.push(elapsed); + samples.set(label, values); +} + +try { + const dir = join(root, "sessions", "project"); + mkdirSync(dir, { recursive: true }); + for (let i = 0; i < count; i++) { + const content = line({ type: "session", version: 3, id: String(i), cwd: root, timestamp }) + message.repeat(64); + writeFileSync(join(dir, `${i}.jsonl`), content); + bytes += Buffer.byteLength(content); + } + for (let trial = 0; trial < 3; trial++) { + rmSync(join(root, "pi-web-session-index.json"), { force: true }); + resetSessionScanIndexForTests(); + await measure("SDK full scan", () => SessionManager.listAll()); + await measure("Scanner, no index", listSessionsIncremental); + await measure("Scanner, memory index", listSessionsIncremental); + const child = spawnSync(process.execPath, [fileURLToPath(import.meta.url), "--persisted-index"], { + env: process.env, encoding: "utf8", timeout: 30_000, + }); + assert.equal(child.status, 0, child.error?.message ?? child.stderr); + const persisted = JSON.parse(child.stdout); + assert.equal(persisted.count, count); + const label = "Scanner, persisted index (new process)"; + const values = samples.get(label) ?? []; + values.push(persisted.elapsed); + samples.set(label, values); + appendFileSync(join(dir, "0.jsonl"), message); + await measure("Scanner, one appended file", listSessionsIncremental); + } + console.log(`${count} synthetic sessions, ${(bytes / 1024 / 1024).toFixed(1)} MiB, 64 messages/file; three trials.`); + console.log("Scan latency only; excludes process startup, SDK imports, API project resolution, and browser rendering."); + console.table([...samples].map(([mode, values]) => ({ + mode, + medianMs: Number([...values].sort((a, b) => a - b)[1].toFixed(1)), + samplesMs: values.map((value) => value.toFixed(1)).join(", "), + }))); +} finally { + resetSessionScanIndexForTests(); + if (previousAgentDir === undefined) delete process.env.PI_CODING_AGENT_DIR; + else process.env.PI_CODING_AGENT_DIR = previousAgentDir; + rmSync(root, { recursive: true, force: true }); +} diff --git a/lib/session-list-scanner.test.mjs b/lib/session-list-scanner.test.mjs new file mode 100644 index 0000000000..7c6e845b28 --- /dev/null +++ b/lib/session-list-scanner.test.mjs @@ -0,0 +1,169 @@ +import assert from "node:assert/strict"; +import fs from "node:fs"; +import { syncBuiltinESMExports } from "node:module"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { createJiti } from "jiti"; + +const jiti = createJiti(import.meta.url); +const { SessionManager } = await jiti.import("@earendil-works/pi-coding-agent"); +const { listSessionsIncremental, resetSessionScanIndexForTests } = await jiti.import("./session-list-scanner.ts"); +const { listAllSessions, invalidateSessionListCache } = await jiti.import("./session-reader.ts"); +const timestamp = "2026-01-01T00:00:00.000Z"; +const line = (entry) => JSON.stringify(entry) + "\n"; +const message = (id, role, content, time = timestamp) => ({ + type: "message", id, parentId: null, timestamp: time, message: { role, content }, +}); + +function fixture(t) { + const previousAgentDir = process.env.PI_CODING_AGENT_DIR; + const root = fs.mkdtempSync(join(tmpdir(), "pi-web-scanner-")); + const dir = join(root, "sessions", "project"); + fs.mkdirSync(dir, { recursive: true }); + process.env.PI_CODING_AGENT_DIR = root; + resetSessionScanIndexForTests(); + invalidateSessionListCache(); + t.after(() => { + resetSessionScanIndexForTests(); + invalidateSessionListCache(); + globalThis.__piSessionPathCache = undefined; + globalThis.__piPathToSessionIdCache = undefined; + if (previousAgentDir === undefined) delete process.env.PI_CODING_AGENT_DIR; + else process.env.PI_CODING_AGENT_DIR = previousAgentDir; + fs.rmSync(root, { recursive: true, force: true }); + }); + function write(id, entries = [], extraHeader = {}) { + const path = join(dir, `${id}.jsonl`); + fs.writeFileSync(path, line({ type: "session", version: 3, id, cwd: root, timestamp, ...extraHeader }) + entries.map(line).join("")); + return path; + } + return { root, dir, write, indexPath: join(root, "pi-web-session-index.json") }; +} + +async function sdkMetadata() { + return (await SessionManager.listAll()).map((session) => { + const info = { ...session }; + delete info.allMessagesText; + return info; + }); +} + +test("matches SDK metadata and tie ordering across cache hits, mutations, and restarts", async (t) => { + const { write, indexPath } = fixture(t); + const a = write("a", [ + message("a1", "user", [{ type: "image", data: "ignored", mimeType: "image/png" }, { type: "text", text: "first" }, { type: "text", text: "request" }]), + message("a2", "assistant", [{ type: "text", text: "answer" }]), + message("a3", "toolResult", [{ type: "text", text: "output" }], "2026-01-03T00:00:00.000Z"), + { type: "custom", data: "x".repeat(256 * 1024) }, + ]); + const b = write("b", [message("b1", "user", "forked request")], { parentSession: a }); + const c = write("c"); + const opened = []; + const createReadStream = fs.createReadStream; + t.mock.method(fs, "createReadStream", (path, ...args) => { + opened.push(path); + return createReadStream(path, ...args); + }); + syncBuiltinESMExports(); + t.after(() => { + t.mock.restoreAll(); + syncBuiltinESMExports(); + }); + async function check(label, expectedReads) { + const expected = await sdkMetadata(); + opened.length = 0; + assert.deepEqual(await listSessionsIncremental(), expected, label); + assert.deepEqual([...opened].sort(), [...expectedReads].sort(), `${label}: scanned files`); + } + + await check("cold", [a, b, c]); + await check("warm", []); + if (process.platform !== "win32") assert.equal(fs.statSync(indexPath).mode & 0o777, 0o600); + resetSessionScanIndexForTests(); + await check("persisted cache", []); + + fs.appendFileSync(a, line({ type: "session_info", name: " Renamed " })); + await check("rename preserves equal-time order", [a]); + fs.appendFileSync(a, line({ type: "session_info", name: " " })); + await check("explicit name clear", [a]); + fs.appendFileSync(b, line(message("b2", "assistant", "new activity", "2026-01-02T00:00:00.000Z"))); + await check("append", [b]); + write("a", [message("replacement", "user", "rewritten")]); + await check("whole-file rewrite", [a]); + const d = write("d", [message("d1", "user", "new session")]); + fs.rmSync(c); + await check("create and delete", [d]); + + resetSessionScanIndexForTests(); + fs.appendFileSync(a, line({ type: "session_info", name: "changed while stopped" })); + fs.rmSync(b); + await check("restart revalidates changed and deleted files", [a]); + const persisted = JSON.parse(fs.readFileSync(indexPath, "utf8")); + assert.deepEqual(Object.keys(persisted.entries).sort(), [a, d].sort()); +}); + +test("discards malformed persisted entries and rebuilds the list from valid session files", async (t) => { + const { write, indexPath } = fixture(t); + const a = write("a", [message("a1", "user", "healthy")]); + write("b", [message("b1", "user", "also healthy")]); + const expected = await listSessionsIncremental(); + const pristine = fs.readFileSync(indexPath, "utf8"); + const corruptions = [ + ["missing fingerprint", (entry) => { delete entry.fp; }], + ["null fingerprint", (entry) => { entry.fp = null; }], + ["string size", (entry) => { entry.fp.size = String(entry.fp.size); }], + ["negative size", (entry) => { entry.fp.size = -1; }], + ["invalid mtime", (entry) => { entry.fp.mtimeMs = null; }], + ["missing metadata", (entry) => { delete entry.info; }], + ["mismatched path", (entry) => { entry.info.path = "/elsewhere/session.jsonl"; }], + ["invalid id", (entry) => { entry.info.id = null; }], + ["invalid cwd", (entry) => { entry.info.cwd = []; }], + ["invalid first message", (entry) => { entry.info.firstMessage = 1; }], + ["invalid name", (entry) => { entry.info.name = {}; }], + ["invalid parent", (entry) => { entry.info.parentSessionPath = []; }], + ["invalid count", (entry) => { entry.info.messageCount = -1; }], + ["invalid created", (entry) => { entry.info.created = "not a date"; }], + ["invalid modified", (entry) => { entry.info.modified = "not a date"; }], + ["null date", (entry) => { entry.info.created = null; }], + ]; + for (const [label, corrupt] of corruptions) { + const index = JSON.parse(pristine); + corrupt(index.entries[a]); + fs.writeFileSync(indexPath, JSON.stringify(index)); + resetSessionScanIndexForTests(); + assert.deepEqual(await listSessionsIncremental(), expected, label); + const sessions = await listAllSessions({ force: true }); + assert.deepEqual(sessions.map((s) => s.id), expected.map((s) => s.id), label); + resetSessionScanIndexForTests(); + assert.deepEqual(await listSessionsIncremental(), expected, `${label}: repaired cache survives restart`); + } + for (const source of ["{", "null", '{"version":2,"entries":{}}', '{"version":1,"entries":[]}', '{"version":1,"entries":{"broken":null}}']) { + fs.writeFileSync(indexPath, source); + resetSessionScanIndexForTests(); + assert.deepEqual(await listSessionsIncremental(), expected, source); + } +}); + +test("retains SDK handling of partial lines, activity timestamps, and empty sessions", async (t) => { + const { dir, write } = fixture(t); + const a = write("a", [ + message("a1", "assistant", "not the first user message"), + message("a2", "user", []), + { ...message("a3", "user", "user message"), message: { role: "user", content: "user message", timestamp: Date.parse("2026-01-04T00:00:00.000Z") } }, + ]); + fs.appendFileSync(a, '{"unfinished":'); + const b = write("b"); + fs.writeFileSync(b, "\nnot json\n" + fs.readFileSync(b, "utf8")); + fs.writeFileSync(join(dir, "invalid.jsonl"), line(message("invalid", "user", "no header"))); + assert.deepEqual(await listSessionsIncremental(), await sdkMetadata()); +}); + +test("discovers sessions through project directory symlinks", { skip: process.platform === "win32" }, async (t) => { + const { root, dir, write } = fixture(t); + write("a", [message("a1", "user", "linked")]); + const external = join(root, "external-project"); + fs.renameSync(dir, external); + fs.symlinkSync(external, dir, "dir"); + assert.deepEqual(await listSessionsIncremental(), await sdkMetadata()); +}); diff --git a/lib/session-list-scanner.ts b/lib/session-list-scanner.ts new file mode 100644 index 0000000000..563949c6c3 --- /dev/null +++ b/lib/session-list-scanner.ts @@ -0,0 +1,368 @@ +// Cache session-list metadata without building the SDK's unused allMessagesText. +// New and changed files still require a full scan; unchanged files only need stat. +// ponytail: size/mtime fingerprints miss same-size edits with restored mtime; +// use content hashes if detecting those edits becomes necessary. +import { createReadStream, existsSync, readFileSync } from "node:fs"; +import { readdir, stat } from "node:fs/promises"; +import type { Dirent } from "node:fs"; +import { join } from "node:path"; +import { createInterface } from "node:readline"; +import { getAgentDir } from "@earendil-works/pi-coding-agent"; +import { writePrivateFileAtomicSync } from "./atomic-file"; + +export interface ScannedSessionInfo { + path: string; + id: string; + cwd: string; + name?: string; + created: Date; + modified: Date; + messageCount: number; + firstMessage: string; + parentSessionPath?: string; +} + +interface Fingerprint { + size: number; + mtimeMs: number; +} + +interface IndexEntry { + fp: Fingerprint; + info: ScannedSessionInfo; +} + +type RawEntry = Record; + +function isRecord(value: unknown): value is RawEntry { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +const INDEX_FORMAT_VERSION = 1; + +declare global { + var __piWebScanIndex: Map | undefined; + var __piWebScanIndexLoaded: boolean | undefined; + var __piWebScanIndexSaveQueued: boolean | undefined; +} + +function parseLine(line: string): RawEntry | null { + if (!line.trim()) return null; + try { + const entry = JSON.parse(line) as RawEntry; + return entry && typeof entry === "object" ? entry : null; + } catch { + return null; + } +} + +function extractTextContent(message: RawEntry): string { + const content = message.content; + if (typeof content === "string") return content; + if (!Array.isArray(content)) return ""; + return content + .filter( + (block): block is { type: string; text: string } => + !!block && + typeof block === "object" && + (block as RawEntry).type === "text" && + typeof (block as RawEntry).text === "string", + ) + .map((block) => block.text) + .join(" "); +} + +function activityTimeOf(entry: RawEntry): number | undefined { + const message = entry.message as RawEntry | undefined; + if ( + !message || + typeof message.role !== "string" || + !("content" in message) || + (message.role !== "user" && message.role !== "assistant") + ) { + return undefined; + } + if (typeof message.timestamp === "number") return message.timestamp; + const t = new Date(entry.timestamp as string).getTime(); + return Number.isNaN(t) ? undefined : t; +} + +// Uses the SDK's buildSessionInfo() semantics for displayed metadata. +export async function scanSessionFileInfo( + filePath: string, +): Promise { + try { + const stats = await stat(filePath); + let header: RawEntry | null = null; + let name: string | undefined; + let messageCount = 0; + let firstMessage = ""; + let lastActivityTime: number | undefined; + + const rl = createInterface({ + input: createReadStream(filePath, { encoding: "utf8" }), + crlfDelay: Infinity, + }); + + for await (const line of rl) { + const entry = parseLine(line); + if (!entry) continue; + + if (!header) { + if (entry.type !== "session") return null; + header = entry; + continue; + } + + if (entry.type === "session_info") { + name = + typeof entry.name === "string" && entry.name.trim() + ? entry.name.trim() + : undefined; + } + if (entry.type !== "message") continue; + messageCount++; + + const activityTime = activityTimeOf(entry); + if (typeof activityTime === "number") { + lastActivityTime = Math.max(lastActivityTime ?? 0, activityTime); + } + + const message = entry.message as RawEntry | undefined; + if ( + !message || + typeof message.role !== "string" || + !("content" in message) + ) + continue; + if (message.role !== "user" && message.role !== "assistant") continue; + + const textContent = extractTextContent(message); + if (!textContent) continue; + if (!firstMessage && message.role === "user") firstMessage = textContent; + } + + if (!header) return null; + + const cwd = typeof header.cwd === "string" ? header.cwd : ""; + const parentSessionPath = + typeof header.parentSession === "string" + ? header.parentSession + : undefined; + 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.id as string, + cwd, + name, + parentSessionPath, + created: new Date(header.timestamp as string), + modified, + messageCount, + firstMessage: firstMessage || "(no messages)", + }; + } catch { + return null; + } +} + +async function enumerateSessionFiles(sessionsDir: string): Promise { + let dirs: Dirent[]; + try { + const entries = await readdir(sessionsDir, { withFileTypes: true }); + dirs = entries.filter( + (entry) => entry.isDirectory() || entry.isSymbolicLink(), + ); + } catch { + return []; + } + + const files: string[] = []; + for (const dir of dirs) { + const dirPath = join(sessionsDir, dir.name); + try { + for (const f of await readdir(dirPath)) { + if (f.endsWith(".jsonl")) files.push(join(dirPath, f)); + } + } catch { + // unreadable project dir: same skip-as-absent semantics as the SDK + } + } + return files; +} + +const MAX_CONCURRENT_SCANS = 10; + +async function runPool(items: T[], worker: (item: T) => Promise) { + let next = 0; + const inFlight = new Set>(); + while (next < items.length || inFlight.size > 0) { + while (next < items.length && inFlight.size < MAX_CONCURRENT_SCANS) { + const item = items[next++]; + const task = worker(item).finally(() => inFlight.delete(task)); + inFlight.add(task); + } + if (inFlight.size > 0) await Promise.race(inFlight); + } +} + +function getIndex(): Map { + if (!globalThis.__piWebScanIndex) globalThis.__piWebScanIndex = new Map(); + return globalThis.__piWebScanIndex; +} + +function indexFilePath(): string { + return join(getAgentDir(), "pi-web-session-index.json"); +} + +function loadPersistedIndex(): void { + if (globalThis.__piWebScanIndexLoaded) return; + globalThis.__piWebScanIndexLoaded = true; + const path = indexFilePath(); + if (!existsSync(path)) return; + try { + const parsed: unknown = JSON.parse(readFileSync(path, "utf8")); + if (!isRecord(parsed) || parsed.version !== INDEX_FORMAT_VERSION || !isRecord(parsed.entries)) return; + const index = getIndex(); + for (const [pathKey, entry] of Object.entries(parsed.entries)) { + if (!isRecord(entry) || !isRecord(entry.fp) || !isRecord(entry.info)) continue; + const { fp, info } = entry; + if ( + typeof fp.size !== "number" || !Number.isSafeInteger(fp.size) || fp.size < 0 || + typeof fp.mtimeMs !== "number" || !Number.isFinite(fp.mtimeMs) || + info.path !== pathKey || + typeof info.id !== "string" || + typeof info.cwd !== "string" || + typeof info.firstMessage !== "string" || + (info.name !== undefined && typeof info.name !== "string") || + (info.parentSessionPath !== undefined && typeof info.parentSessionPath !== "string") || + typeof info.messageCount !== "number" || !Number.isSafeInteger(info.messageCount) || info.messageCount < 0 || + typeof info.created !== "string" || typeof info.modified !== "string" + ) continue; + const created = new Date(info.created); + const modified = new Date(info.modified); + if (!Number.isFinite(created.getTime()) || !Number.isFinite(modified.getTime())) continue; + index.set(pathKey, { + fp: { size: fp.size, mtimeMs: fp.mtimeMs }, + info: { + path: pathKey, + id: info.id, + cwd: info.cwd, + name: info.name, + parentSessionPath: info.parentSessionPath, + firstMessage: info.firstMessage, + messageCount: info.messageCount, + created, + modified, + }, + }); + } + } catch { + // corrupt index => cold rebuild, never an error surfaced to callers + } +} + +function queueIndexPersist(): void { + if (globalThis.__piWebScanIndexSaveQueued) return; + globalThis.__piWebScanIndexSaveQueued = true; + queueMicrotask(() => { + globalThis.__piWebScanIndexSaveQueued = undefined; + try { + const entries: Record = {}; + for (const [pathKey, entry] of getIndex()) entries[pathKey] = entry; + writePrivateFileAtomicSync( + indexFilePath(), + JSON.stringify({ version: INDEX_FORMAT_VERSION, entries }), + ); + } catch { + // persistence is best-effort; the in-memory index remains authoritative + } + }); +} + +/** + * Incremental equivalent of SessionManager.listAll(): rescans only files whose + * (size, mtimeMs) changed since the last pass. Output ordering matches the SDK + * catalogue (modified descending). + */ +export async function listSessionsIncremental(): Promise { + loadPersistedIndex(); + + const sessionsDir = join(getAgentDir(), "sessions"); + const files = await enumerateSessionFiles(sessionsDir); + + const index = getIndex(); + const present = new Set(files); + const stale: string[] = []; + for (const known of index.keys()) { + if (!present.has(known)) stale.push(known); + } + for (const pathKey of stale) index.delete(pathKey); + + const fingerprints = await Promise.all( + files.map(async (filePath) => { + try { + const s = await stat(filePath); + return { + filePath, + fp: { size: s.size, mtimeMs: s.mtimeMs } as Fingerprint, + }; + } catch { + return { filePath, fp: null as Fingerprint | null }; + } + }), + ); + + const changed: Array<{ filePath: string; fp: Fingerprint; resultIndex: number }> = []; + const results: (ScannedSessionInfo | null)[] = new Array(files.length).fill(null); + for (const [resultIndex, { filePath, fp }] of fingerprints.entries()) { + if (!fp) { + index.delete(filePath); + continue; + } + const cached = index.get(filePath); + if ( + cached && + cached.fp.size === fp.size && + cached.fp.mtimeMs === fp.mtimeMs + ) { + results[resultIndex] = cached.info; + continue; + } + changed.push({ filePath, fp, resultIndex }); + } + + await runPool(changed, async ({ filePath, fp, resultIndex }) => { + const info = await scanSessionFileInfo(filePath); + if (info) { + index.set(filePath, { fp, info }); + results[resultIndex] = info; + } else { + index.delete(filePath); + } + }); + + if (changed.length > 0 || stale.length > 0) queueIndexPersist(); + + // Preserve catalogue order for timestamp ties, independently of cache hits + // and the order in which concurrent file reads complete. + return results.filter((info) => info !== null) + .sort((a, b) => b.modified.getTime() - a.modified.getTime()); +} + +/** Test seam: drop all in-memory index state. */ +export function resetSessionScanIndexForTests(): void { + globalThis.__piWebScanIndex = undefined; + globalThis.__piWebScanIndexLoaded = undefined; + globalThis.__piWebScanIndexSaveQueued = undefined; +} diff --git a/lib/session-reader.test.mjs b/lib/session-reader.test.mjs index 347649394f..9a353b6e70 100644 --- a/lib/session-reader.test.mjs +++ b/lib/session-reader.test.mjs @@ -19,6 +19,10 @@ const { resolveSessionPath, } = await jiti.import("./session-reader.ts"); const { SessionManager } = await jiti.import("@earendil-works/pi-coding-agent"); +const scanner = await jiti.import("./session-list-scanner.ts"); +// Jiti caches imported bindings, so keep one dispatch function for all mocks. +let listSessions = scanner.listSessionsIncremental; +scanner.listSessionsIncremental = (...args) => listSessions(...args); function resetSessionListState() { globalThis.__piSessionListCache = undefined; @@ -433,10 +437,10 @@ test("session listing reads subagent relations and terminal status without reope }), ].join("\n")); - const originalListAll = SessionManager.listAll; + const originalListSessions = listSessions; const originalOpen = SessionManager.open; let fullOpens = 0; - SessionManager.listAll = async () => [{ + listSessions = async () => [{ path: filePath, id: childId, cwd: dir, @@ -453,7 +457,7 @@ test("session listing reads subagent relations and terminal status without reope }; resetSessionListState(); t.after(() => { - SessionManager.listAll = originalListAll; + listSessions = originalListSessions; SessionManager.open = originalOpen; invalidateSessionPathCache(childId); resetSessionListState(); @@ -504,15 +508,15 @@ test("resolves a matching session header without a catalogue scan", async (t) => cwd: "/tmp/project", })}\n${JSON.stringify(userEntry("u1", null, "not scanned"))}\n`); - const originalListAll = SessionManager.listAll; + const originalListSessions = listSessions; let scans = 0; - SessionManager.listAll = async () => { + listSessions = async () => { scans += 1; return []; }; setTestAgentDir(t, agentDir); t.after(() => { - SessionManager.listAll = originalListAll; + listSessions = originalListSessions; }); assert.equal(await resolveSessionPath(sessionId), filePath); @@ -537,15 +541,15 @@ test("does not resolve a parent path outside the default session storage", async cwd: "/tmp/project", })}\n`); - const originalListAll = SessionManager.listAll; + const originalListSessions = listSessions; let scans = 0; - SessionManager.listAll = async () => { + listSessions = async () => { scans += 1; return []; }; setTestAgentDir(t, agentDir); t.after(() => { - SessionManager.listAll = originalListAll; + listSessions = originalListSessions; rmSync(externalDir, { recursive: true, force: true }); }); @@ -573,9 +577,9 @@ test( mkdirSync(join(agentDir, "sessions"), { recursive: true }); symlinkSync(externalDir, join(agentDir, "sessions", "linked-project"), "dir"); - const originalListAll = SessionManager.listAll; + const originalListSessions = listSessions; let scans = 0; - SessionManager.listAll = async () => { + listSessions = async () => { scans += 1; return [ { @@ -592,7 +596,7 @@ test( }; setTestAgentDir(t, agentDir); t.after(() => { - SessionManager.listAll = originalListAll; + listSessions = originalListSessions; rmSync(externalDir, { recursive: true, force: true }); }); @@ -625,15 +629,15 @@ test("falls back to the catalogue when a targeted candidate header is invalid", cwd: "/tmp/project", })}\n`); - const originalListAll = SessionManager.listAll; + const originalListSessions = listSessions; let scans = 0; - SessionManager.listAll = async () => { + listSessions = async () => { scans += 1; return []; }; setTestAgentDir(t, agentDir); t.after(() => { - SessionManager.listAll = originalListAll; + listSessions = originalListSessions; }); assert.equal(await resolveSessionPath(sessionId), null); @@ -655,7 +659,7 @@ test("falls back to the catalogue when a targeted lookup finds duplicate IDs", a })}\n`); } - const originalListAll = SessionManager.listAll; + const originalListSessions = listSessions; const fallbackDir = join(agentDir, "sessions", "catalogue"); const fallbackPath = join(fallbackDir, "fallback.jsonl"); mkdirSync(fallbackDir, { recursive: true }); @@ -667,7 +671,7 @@ test("falls back to the catalogue when a targeted lookup finds duplicate IDs", a cwd: "/tmp/project", })}\n`); let scans = 0; - SessionManager.listAll = async () => { + listSessions = async () => { scans += 1; return [{ path: fallbackPath, @@ -682,7 +686,7 @@ test("falls back to the catalogue when a targeted lookup finds duplicate IDs", a }; setTestAgentDir(t, agentDir); t.after(() => { - SessionManager.listAll = originalListAll; + listSessions = originalListSessions; }); assert.equal(await resolveSessionPath(sessionId), fallbackPath); @@ -690,15 +694,15 @@ test("falls back to the catalogue when a targeted lookup finds duplicate IDs", a }); test("forced session listing bypasses the fresh server cache", async (t) => { - const originalListAll = SessionManager.listAll; + const originalListSessions = listSessions; let scans = 0; - SessionManager.listAll = async () => { + listSessions = async () => { scans += 1; return []; }; resetSessionListState(); t.after(() => { - SessionManager.listAll = originalListAll; + listSessions = originalListSessions; resetSessionListState(); }); @@ -711,7 +715,7 @@ test("forced session listing bypasses the fresh server cache", async (t) => { }); test("a scan invalidated in flight retries before returning to its caller", async (t) => { - const originalListAll = SessionManager.listAll; + const originalListSessions = listSessions; let scans = 0; let releaseFirstScan; let markFirstScanStarted; @@ -721,7 +725,7 @@ test("a scan invalidated in flight retries before returning to its caller", asyn const firstScanGate = new Promise((resolve) => { releaseFirstScan = resolve; }); - SessionManager.listAll = async () => { + listSessions = async () => { scans += 1; if (scans === 1) { markFirstScanStarted(); @@ -731,7 +735,7 @@ test("a scan invalidated in flight retries before returning to its caller", asyn }; resetSessionListState(); t.after(() => { - SessionManager.listAll = originalListAll; + listSessions = originalListSessions; resetSessionListState(); }); diff --git a/lib/session-reader.ts b/lib/session-reader.ts index 109c6502d5..62bede34d2 100644 --- a/lib/session-reader.ts +++ b/lib/session-reader.ts @@ -6,13 +6,13 @@ import { closeSync, type Dirent, fstatSync, openSync, readSync } from "fs"; import { readdir } from "fs/promises"; import { isAbsolute, join, normalize as normalizePath, relative, resolve as resolvePath, sep } from "path"; import type { AgentMessage, ImageContent, SessionEntry, SessionHeader, SessionInfo, SessionContext } from "./types"; -import type { SessionInfo as PiSessionInfo } from "@earendil-works/pi-coding-agent"; import { normalizeToolCalls } from "./normalize"; import { projectIdentityKey } from "./project-identity"; import { sessionPathKey } from "./session-path"; import { MAX_TOOL_RESULT_IMAGE_BYTES, TOOL_RESULT_IMAGE_MIMES } from "./tool-result-images"; import { resolveProject, type ProjectInfo } from "./worktree"; import { readSubagentRun, SUBAGENT_META_TYPE } from "./subagents"; +import { listSessionsIncremental } from "./session-list-scanner"; export { getAgentDir }; @@ -140,11 +140,11 @@ export function mergeSessionLists( } async function loadAllSessions(): Promise { - const piSessions: PiSessionInfo[] = await SessionManager.listAll(); + const scanned = await listSessionsIncremental(); const pathToId = new Map(); - for (const s of piSessions) pathToId.set(sessionPathKey(s.path), s.id); + for (const s of scanned) pathToId.set(sessionPathKey(s.path), s.id); - const sessions = piSessions.map((s) => { + const sessions = scanned.map((s) => { cacheSessionPath(s.id, s.path); const originSessionId = s.parentSessionPath ? pathToId.get(sessionPathKey(s.parentSessionPath)) : undefined; let subagent = null; @@ -158,8 +158,8 @@ async function loadAllSessions(): Promise { id: s.id, cwd: s.cwd, name: s.name, - created: s.created instanceof Date ? s.created.toISOString() : String(s.created), - modified: s.modified instanceof Date ? s.modified.toISOString() : String(s.modified), + created: s.created.toISOString(), + modified: s.modified.toISOString(), messageCount: s.messageCount, firstMessage: s.firstMessage || "(no messages)", parentSessionId: originSessionId,