From e957559959c9c13f094a30b3684aa9b8029a20cf Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 16 Oct 2026 14:32:16 +0000 Subject: [PATCH 1/3] fix(web): bound session catalog and history read cost Cache Pi session metadata by file identity and parse non-current history off the request thread, so an unchanged catalog, a cancelled search, and a later history page do not read or synchronously parse every JSONL body again. Fixes #834 Co-authored-by: tt-a1i --- tests/web/pi-adapter.test.ts | 90 +++-- tests/web/session-read-cost.test.ts | 319 ++++++++++++++++ web/adapter/pi-adapter.ts | 108 ++++-- web/adapter/session-catalog.ts | 540 ++++++++++++++++++++++++++++ web/adapter/session-file-worker.ts | 14 + web/adapter/session-loader.ts | 170 +++++++++ 6 files changed, 1177 insertions(+), 64 deletions(-) create mode 100644 tests/web/session-read-cost.test.ts create mode 100644 web/adapter/session-catalog.ts create mode 100644 web/adapter/session-file-worker.ts create mode 100644 web/adapter/session-loader.ts diff --git a/tests/web/pi-adapter.test.ts b/tests/web/pi-adapter.test.ts index acf838e24..97ff9966f 100644 --- a/tests/web/pi-adapter.test.ts +++ b/tests/web/pi-adapter.test.ts @@ -1984,7 +1984,7 @@ test("archived Session listing is queryable, cursor-bounded, and stale-safe", as } }); -test("archive scan continuations reach old matches after an empty bounded window in global sort order", async (t) => { +test("archive scan continuations reach old matches after an empty bounded window in global sort order", async () => { const root = await mkdtemp(join(tmpdir(), "openpi-web-archive-scan-next-")); const project = join(root, "old-project"); try { @@ -2018,10 +2018,34 @@ test("archive scan continuations reach old matches after an empty bounded window firstMessage: "old needle B", }, ]; - const listing = t.mock.method(SessionManager, "listAll", async () => [ - ...active, - ...older, - ]); + const sessionFile = ( + id: string, + cwd: string, + createdMs: number, + modifiedMs: number, + text: string, + ) => { + const timestamp = new Date(createdMs).toISOString(); + return `${JSON.stringify({ type: "session", version: 3, id, timestamp, cwd })}\n${JSON.stringify({ + type: "message", + id: "m", + parentId: null, + timestamp, + message: { role: "user", content: text, timestamp: modifiedMs }, + })}\n`; + }; + const writes: Promise[] = []; + for (const session of [...active, ...older]) { + writes.push(writeFile( + session.path, + sessionFile(session.id, session.cwd, session.created.getTime(), session.modified.getTime(), session.firstMessage), + )); + if (writes.length === 64) { + await Promise.all(writes); + writes.length = 0; + } + } + await Promise.all(writes); await writeFile( join(root, "archived-sessions.json"), JSON.stringify(older.map((session) => session.path)), @@ -2071,11 +2095,10 @@ test("archive scan continuations reach old matches after an empty bounded window assert.equal(last.nextCursor, undefined); assert.equal(last.truncation.recordsUnscanned, 0); } - listing.mock.mockImplementation(async () => [ - ...active, - { ...older[0]!, created: new Date(200_000) }, - older[1]!, - ]); + await writeFile( + older[0]!.path, + sessionFile("older-a", project, 200_000, older[0]!.modified.getTime(), "old needle A"), + ); const globallyNewest = await adapter.listArchivedSessions({ sort: "created", limit: 1, @@ -2142,34 +2165,38 @@ test("archive cursors produced for long valid Session paths remain bounded and r } }); -test("archived listing rejects pre-cancelled admission and forwards in-progress cancellation to Pi", async (t) => { +test("archived listing rejects pre-cancelled admission and stops an in-progress catalog read", async (t) => { const root = await mkdtemp(join(tmpdir(), "openpi-web-archive-cancel-")); const controller = new AbortController(); const reason = new Error("archive query cancelled"); + await writeFile( + join(root, "one.jsonl"), + `${JSON.stringify({ type: "session", version: 3, id: "one", timestamp: "2026-10-15T00:00:00.000Z", cwd: root })}\n`, + ); + const originalOpen = fsPromises.open; + let jsonlOpens = 0; let admit: () => void = () => {}; const admitted = new Promise((resolve) => { admit = resolve; }); - let receivedSignal: AbortSignal | undefined; - const listing = t.mock.method( - SessionManager, - "listAll", - async ( - ...[_directory, _progress, signal]: Parameters< - typeof SessionManager.listAll - > - ) => { - receivedSignal = signal; - admit(); - signal?.throwIfAborted(); - if (!signal) throw new Error("Pi catalog requires the query signal"); - return new Promise((_resolve, reject) => { - signal.addEventListener("abort", () => reject(signal.reason), { - once: true, - }); - }); + t.mock.method( + fsPromises, + "open", + async (...args: Parameters) => { + const path = String(args[0]); + if (path.startsWith(root) && path.endsWith(".jsonl")) { + jsonlOpens++; + admit(); + return new Promise(() => {}); + } + return originalOpen(...args); }, ); + syncBuiltinESMExports(); + t.after(() => { + t.mock.restoreAll(); + syncBuiltinESMExports(); + }); try { const adapter = new PiWebAdapter( runtimeFor(root, root, SessionManager.inMemory(root)), @@ -2179,14 +2206,13 @@ test("archived listing rejects pre-cancelled admission and forwards in-progress adapter.listArchivedSessions({}, preCancelled), (error) => error === reason, ); - assert.equal(listing.mock.callCount(), 0); + assert.equal(jsonlOpens, 0); const outcome = adapter.listArchivedSessions({}, controller.signal).then( () => undefined, (error: unknown) => error, ); await admitted; - assert.equal(listing.mock.callCount(), 1); - assert.equal(receivedSignal, controller.signal); + assert.equal(jsonlOpens, 1); controller.abort(reason); assert.equal(await outcome, reason); } finally { diff --git a/tests/web/session-read-cost.test.ts b/tests/web/session-read-cost.test.ts new file mode 100644 index 000000000..10c6f83e7 --- /dev/null +++ b/tests/web/session-read-cost.test.ts @@ -0,0 +1,319 @@ +import assert from "node:assert/strict"; +import fsPromises from "node:fs/promises"; +import { mkdir, mkdtemp, rename, rm, writeFile } from "node:fs/promises"; +import { syncBuiltinESMExports } from "node:module"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { SessionManager } from "@earendil-works/pi-coding-agent"; +import { PiWebAdapter } from "../../web/adapter/pi-adapter.ts"; +import { readSessionFork } from "../../web/protocol/session-fork.ts"; +import { WEB_SESSION_FORK } from "../../web/protocol/session-fork.ts"; +import { readSessionTurnEvidence, WEB_TURN_TIMING_ENTRY } from "../../web/protocol/turn-timing.ts"; +import { WEB_TRANSCRIPT_SEARCH_MAX_TOTAL_BYTES } from "../../web/search/transcript-search.ts"; +import { projectWebModelSearch } from "../../web/runtime/model-discovery.ts"; +import type { WebRuntimeController } from "../../web/runtime/types.ts"; + +function runtimeFor(cwd: string, sessionDirectory: string, sessionManager: SessionManager): WebRuntimeController { + return { + cwd, + workspaceSelected: true, + sessionDirectory, + sessionManager, + isIdle: () => true, + getActiveTurn: () => undefined, + cancelTurn: async (options) => ({ ...options, state: "stale-turn" }), + sendPrompt: async () => ({ pendingFollowUps: 0 }), + newSession: async () => ({ cancelled: false, sessionId: sessionManager.getSessionId() }), + switchSession: async () => ({ cancelled: false }), + listModels: () => [], + searchModels: (query, limit) => projectWebModelSearch([], query, limit), + setModel: async () => { + throw new Error("Model is not available"); + }, + subscribe: () => () => {}, + dispose: async () => {}, + }; +} + +const zeroUsage = { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, +}; + +function persistSession(manager: SessionManager, content: string, timestamp: number) { + manager.appendMessage({ role: "user", content, timestamp }); + return manager.appendMessage({ + role: "assistant", + content: [], + api: "openai-responses", + provider: "fixture", + model: "fixture", + usage: zeroUsage, + stopReason: "stop", + timestamp, + }); +} + +test("catalog metadata matches Pi and unchanged files are not read again", async (t) => { + const root = await mkdtemp(join(tmpdir(), "openpi-catalog-identity-")); + const directory = join(root, "sessions"); + t.after(() => rm(root, { recursive: true, force: true })); + const named = SessionManager.create(root, directory); + persistSession(named, "named first", 10); + named.appendSessionInfo("Visible name"); + const plain = SessionManager.create(root, directory); + persistSession(plain, "plain first", 20); + const branched = SessionManager.create(root, directory); + persistSession(branched, "kept", 30); + const forkPoint = branched.getLeafId()!; + persistSession(branched, "abandoned", 31); + branched.branch(forkPoint); + persistSession(branched, "replacement", 32); + const oracle = await SessionManager.listAll(directory); + const adapter = new PiWebAdapter(runtimeFor(root, directory, SessionManager.inMemory(root))); + const originalOpen = fsPromises.open; + const opened: string[] = []; + t.mock.method(fsPromises, "open", async (...args: Parameters) => { + const path = String(args[0]); + if (path.startsWith(directory) && path.endsWith(".jsonl")) opened.push(path); + return originalOpen(...args); + }); + syncBuiltinESMExports(); + t.after(() => { + t.mock.restoreAll(); + syncBuiltinESMExports(); + }); + + const listed = await adapter.listSessions(); + const files = listed.filter((session) => session.path.endsWith(".jsonl")); + const byPath = new Map(files.map((session) => [session.path, session])); + assert.deepEqual(files.map((session) => session.id), oracle.map((session) => session.id)); + for (const expected of oracle) { + const actual = byPath.get(expected.path); + assert.ok(actual); + assert.equal(actual.messageCount, expected.messageCount); + assert.equal(actual.name, expected.name); + assert.equal(actual.firstMessage, expected.firstMessage.slice(0, 2_000)); + assert.equal(Date.parse(actual.modified), expected.modified.getTime()); + assert.equal(Date.parse(actual.created), expected.created.getTime()); + } + const firstRead = [...opened]; + assert.equal(new Set(firstRead).size, oracle.length); + const again = await adapter.listSessions(); + assert.deepEqual( + again.filter((session) => session.path.endsWith(".jsonl")).map((session) => session.id), + files.map((session) => session.id), + ); + assert.deepEqual(opened, firstRead); + + plain.appendMessage({ role: "user", content: "appended", timestamp: 40 }); + await adapter.listSessions(); + const appended = opened.slice(firstRead.length); + assert.deepEqual(appended, [plain.getSessionFile()]); + + const from = named.getSessionFile()!; + const to = join(directory, "renamed.jsonl"); + const beforeRename = opened.length; + await rename(from, to); + const renamed = await adapter.listSessions(); + assert.equal(renamed.some((session) => session.path === from), false); + assert.equal(renamed.find((session) => session.path === to)?.id, named.getSessionId()); + const renameReads = opened.slice(beforeRename).filter((path) => path !== to && path !== from); + assert.deepEqual(renameReads, []); + + const beforeDelete = opened.length; + await rm(plain.getSessionFile()!); + const remaining = await adapter.listSessions(); + assert.equal(remaining.some((session) => session.path === plain.getSessionFile()), false); + assert.deepEqual(opened.slice(beforeDelete), []); + + const replaced = branched.getSessionFile()!; + const beforeReplace = opened.length; + await writeFile(replaced, `${JSON.stringify({ type: "session", version: 3, id: "replaced-id", timestamp: "2026-10-15T00:00:00.000Z", cwd: root })}\n${JSON.stringify({ type: "message", id: "m", parentId: null, timestamp: "2026-10-15T00:00:00.000Z", message: { role: "user", content: "replaced body", timestamp: 50 } })}\n`); + const afterReplace = await adapter.listSessions(); + assert.equal(afterReplace.some((session) => session.id === branched.getSessionId()), false); + assert.equal(afterReplace.find((session) => session.path === replaced)?.firstMessage, "replaced body"); + assert.deepEqual(opened.slice(beforeReplace), [replaced]); +}); + +test("non-current history keeps Pi's branch without a synchronous open or a second read", async (t) => { + const root = await mkdtemp(join(tmpdir(), "openpi-history-read-")); + const directory = join(root, "sessions"); + t.after(() => rm(root, { recursive: true, force: true })); + const manager = SessionManager.create(root, directory); + persistSession(manager, "kept", 1); + const forkPoint = manager.getLeafId()!; + const abandoned = manager.appendMessage({ role: "user", content: "abandoned", timestamp: 2 }); + manager.branch(forkPoint); + const replacement = persistSession(manager, "replacement", 3); + const assistant = manager.getLeafId()!; + manager.appendCustomEntry(WEB_TURN_TIMING_ENTRY, { + version: 1, + sessionId: manager.getSessionId(), + commandId: "turn-1", + epoch: 1, + startedAt: 1, + finishedAt: 3, + elapsedMs: 2, + outcome: "completed", + unreadEligible: true, + resultEntryId: assistant, + }); + manager.appendCustomEntry(WEB_SESSION_FORK, { + version: 1, + sessionId: manager.getSessionId(), + source: { sessionId: "source-session", sessionPath: join(root, "source.jsonl"), entryId: "source-entry" }, + environment: "local", + cwd: root, + workspacePath: join(root, "project"), + }); + const path = manager.getSessionFile()!; + const oracle = SessionManager.open(path, directory); + const oracleBranch = oracle.getBranch(); + const oracleIds = oracleBranch.map((entry) => entry.id); + const oracleFork = readSessionFork(oracleBranch, oracle.getSessionId(), oracle.getCwd()); + const oracleEvidence = readSessionTurnEvidence(oracleBranch, oracle.getSessionId(), false); + const adapter = new PiWebAdapter(runtimeFor(root, directory, SessionManager.inMemory(root))); + const originalOpen = SessionManager.open; + const originalRead = fsPromises.readFile; + let syncOpens = 0; + let bodyReads = 0; + SessionManager.open = ((...args: Parameters) => { + syncOpens++; + return originalOpen(...args); + }) as typeof SessionManager.open; + t.mock.method(fsPromises, "readFile", async (...args: Parameters) => { + if (args[0] === path) bodyReads++; + return originalRead(...args); + }); + syncBuiltinESMExports(); + t.after(() => { + SessionManager.open = originalOpen; + t.mock.restoreAll(); + syncBuiltinESMExports(); + }); + + const listed = await adapter.listSessions(); + const summary = listed.find((session) => session.path === path); + assert.ok(summary); + assert.equal(summary.workspacePath, oracleFork?.workspacePath); + assert.equal(summary.unread, oracleEvidence.results.length > 0); + const session = await adapter.getSession(path, [summary]); + assert.ok(session); + assert.equal(syncOpens, 0); + assert.equal(bodyReads, 1); + const projected = new Set(session.entries.map((entry) => entry.id)); + for (const id of projected) assert.ok(oracleIds.includes(id)); + assert.equal(projected.has(abandoned), false); + assert.equal(projected.has(replacement), true); + const anchor = session.entries[0]?.id; + assert.ok(anchor); + const page = await adapter.getSessionHistory(manager.getSessionId(), path, oracle.getLeafId()!, anchor); + assert.equal(page.status, "ok"); + assert.equal(bodyReads, 1); + if (page.status === "ok") { + for (const entry of page.session.entries) assert.ok(oracleIds.includes(entry.id)); + } +}); + +test("concurrent catalog reads share one scan", async (t) => { + const root = await mkdtemp(join(tmpdir(), "openpi-catalog-share-")); + const directory = join(root, "sessions"); + t.after(() => rm(root, { recursive: true, force: true })); + for (const id of ["a", "b", "c"]) { + const manager = SessionManager.create(root, directory); + persistSession(manager, id, 1); + } + const adapter = new PiWebAdapter(runtimeFor(root, directory, SessionManager.inMemory(root))); + const originalOpen = fsPromises.open; + let opens = 0; + let releaseHold: () => void = () => {}; + const hold = new Promise((resolve) => { + releaseHold = resolve; + }); + let entered: () => void = () => {}; + const started = new Promise((resolve) => { + entered = resolve; + }); + t.mock.method(fsPromises, "open", async (...args: Parameters) => { + const path = String(args[0]); + if (path.startsWith(directory) && path.endsWith(".jsonl")) { + opens++; + if (opens === 1) { + entered(); + await hold; + } + } + return originalOpen(...args); + }); + syncBuiltinESMExports(); + t.after(() => { + releaseHold(); + t.mock.restoreAll(); + syncBuiltinESMExports(); + }); + const first = adapter.listSessions(); + await started; + const second = adapter.listSessions(); + releaseHold(); + await Promise.all([first, second]); + assert.equal(opens, 3); +}); + +test("search cancellation and byte budget include the catalog pre-scan", async (t) => { + const root = await mkdtemp(join(tmpdir(), "openpi-search-budget-")); + const directory = join(root, "sessions"); + await mkdir(directory, { recursive: true }); + t.after(() => rm(root, { recursive: true, force: true })); + const small = join(directory, "small"); + await mkdir(small, { recursive: true }); + for (let index = 0; index < 6; index++) { + const manager = SessionManager.create(root, small); + persistSession(manager, `needle ${index}`, index + 1); + } + const adapter = new PiWebAdapter(runtimeFor(root, small, SessionManager.inMemory(root))); + const controller = new AbortController(); + const originalOpen = fsPromises.open; + let opens = 0; + t.mock.method(fsPromises, "open", async (...args: Parameters) => { + const path = String(args[0]); + if (path.startsWith(small) && path.endsWith(".jsonl")) { + opens++; + if (opens === 1) controller.abort(); + } + return originalOpen(...args); + }); + syncBuiltinESMExports(); + await assert.rejects(adapter.searchTranscripts("needle", { signal: controller.signal }), /abort/i); + assert.ok(opens < 6); + t.mock.restoreAll(); + syncBuiltinESMExports(); + + const bulky = join(directory, "bulky"); + await mkdir(bulky, { recursive: true }); + const fileBytes = 1024 * 1024; + const fileCount = Math.floor(WEB_TRANSCRIPT_SEARCH_MAX_TOTAL_BYTES / fileBytes) + 2; + await Promise.all(Array.from({ length: fileCount }, (_, index) => + writeFile(join(bulky, `${index}.jsonl`), Buffer.alloc(fileBytes, 120)))); + let bulkyOpens = 0; + t.mock.method(fsPromises, "open", async (...args: Parameters) => { + const path = String(args[0]); + if (path.startsWith(bulky) && path.endsWith(".jsonl")) bulkyOpens++; + return originalOpen(...args); + }); + syncBuiltinESMExports(); + t.after(() => { + t.mock.restoreAll(); + syncBuiltinESMExports(); + }); + const found = await new PiWebAdapter(runtimeFor(root, bulky, SessionManager.inMemory(root))).searchTranscripts("needle"); + assert.equal(found.partial, true); + assert.ok(found.partialReasons.includes("byte-limit")); + assert.ok(bulkyOpens < fileCount); + assert.ok(bulkyOpens * fileBytes <= WEB_TRANSCRIPT_SEARCH_MAX_TOTAL_BYTES); +}); diff --git a/web/adapter/pi-adapter.ts b/web/adapter/pi-adapter.ts index d5568b13f..95246b4f4 100644 --- a/web/adapter/pi-adapter.ts +++ b/web/adapter/pi-adapter.ts @@ -60,7 +60,13 @@ import type { import { matchesSessionIdentity } from "../runtime/session-identity.ts"; import { WebRuntimeRequestError } from "../runtime/types.ts"; import { WEB_TURN_TIMING_ENTRY, readSessionTurnEvidence, readTurnTiming } from "../protocol/turn-timing.ts"; -import { searchWebTranscripts } from "../search/transcript-search.ts"; +import { + WEB_TRANSCRIPT_SEARCH_MAX_DURATION_MS, + WEB_TRANSCRIPT_SEARCH_MAX_TOTAL_BYTES, + searchWebTranscripts, +} from "../search/transcript-search.ts"; +import { SessionCatalog } from "./session-catalog.ts"; +import { SessionLoader } from "./session-loader.ts"; import { collectSessionSources } from "./session-sources.ts"; import { isSessionPrompt, setupDisplayMessage } from "../protocol/prompt-navigation.ts"; import { @@ -465,6 +471,8 @@ export class PiWebAdapter { private initialization?: Promise; private restoreInitialWorkspace = false; private writeSequence = 0; + private readonly sessionCatalog = new SessionCatalog(); + private readonly sessionLoader = new SessionLoader(); constructor(runtime: WebRuntimeController) { this.runtime = runtime; @@ -760,7 +768,7 @@ export class PiWebAdapter { const session = await this.requireSession(path); if (session.id !== id) throw new WebRuntimeRequestError("The exact Session is unavailable", "SESSION_CONFLICT", 409); const manager = this.runtime.getSessionManagerForRead?.(id, path) ?? - (this.isCurrentSession(session) ? this.runtime.sessionManager : SessionManager.open(path)); + (this.isCurrentSession(session) ? this.runtime.sessionManager : await this.readSession(path)); if (!matchesSessionIdentity(manager, { expectedSessionId: id, expectedSessionPath: path })) throw new WebRuntimeRequestError("The Session changed before export", "SESSION_CONFLICT", 409); const sections = [`# ${(session.name || session.firstMessage || "Conversation").replace(/[\r\n]+/g, " ")}\n`]; @@ -845,11 +853,7 @@ export class PiWebAdapter { if (cursor && cursor.queryHash !== queryHash) { return { status: "stale_cursor" as const }; } - const allSessions = await SessionManager.listAll( - this.runtime.sessionDirectory, - undefined, - signal, - ); + const allSessions = (await this.sessionCatalog.list(this.runtime.sessionDirectory, { signal })).sessions; signal?.throwIfAborted(); // Order the complete Pi catalog before locating the continuation window. if (sort === "created") { @@ -960,7 +964,7 @@ export class PiWebAdapter { if (!workspaces.has(canonical)) { throw new Error("Workspace is not available"); } - const sessions = await SessionManager.listAll(this.runtime.sessionDirectory); + const sessions = (await this.sessionCatalog.list(this.runtime.sessionDirectory)).sessions; for (const session of sessions) { if (resolve(this.sessionFork(session)?.workspacePath ?? session.cwd) === canonical) { draft.ungroupedSessions.add(resolve(session.path)); @@ -1069,17 +1073,26 @@ export class PiWebAdapter { let acknowledgedResults: ReadTurn[] = acknowledged ? [acknowledged] : []; if (!session.path.startsWith("current:")) { try { - const manager = this.runtime.getSessionManagerForRead?.(session.id, session.path) ?? (this.isCurrentSession(session) ? this.runtime.sessionManager : SessionManager.open(session.path, this.runtime.sessionDirectory)); - if (matchesSessionIdentity(manager, { expectedSessionId: session.id, expectedSessionPath: session.path })) { + const manager = this.runtime.getSessionManagerForRead?.(session.id, session.path) ?? (this.isCurrentSession(session) ? this.runtime.sessionManager : undefined); + const branch = manager + ? (matchesSessionIdentity(manager, { expectedSessionId: session.id, expectedSessionPath: session.path }) ? manager.getBranch() : undefined) + : this.sessionCatalog.branch(session.path, session.id); + const entries = manager ? (branch ? manager.getEntries() : undefined) : this.sessionCatalog.entries(session.path, session.id); + if (branch && entries) { // An existing exact read position can still distinguish older unread // results. Without it, history discovery is not a live publication. - native = readSessionTurnEvidence(manager.getBranch(), session.id, Boolean(acknowledged)); + native = readSessionTurnEvidence(branch, session.id, Boolean(acknowledged)); if (acknowledged) { - const readEntry = manager.getEntries().find((entry) => { + const readEntry = entries.find((entry) => { const timing = entry.type === "custom" && entry.customType === WEB_TURN_TIMING_ENTRY ? readTurnTiming(entry.data) : undefined; return timing?.sessionId === session.id && sameReadTurn(timing, acknowledged); }); - if (readEntry) acknowledgedResults = readSessionTurnEvidence(manager.getBranch(readEntry.id), session.id, true).results; + if (readEntry) { + const acknowledgedBranch = manager + ? manager.getBranch(readEntry.id) + : this.sessionCatalog.branch(session.path, session.id, readEntry.id); + if (acknowledgedBranch) acknowledgedResults = readSessionTurnEvidence(acknowledgedBranch, session.id, true).results; + } } } } catch { @@ -1124,9 +1137,7 @@ export class PiWebAdapter { async listSessionProjection(pinnedPath?: string, chatsSort: "manual" | "updated" = "updated") { await this.ensureWorkspaceStateLoaded(); await this.ensureArchivesLoaded(); - const allSessions = await SessionManager.listAll( - this.runtime.sessionDirectory, - ); + const allSessions = (await this.sessionCatalog.list(this.runtime.sessionDirectory)).sessions; // Filter and order Pi's catalog before bounding the visible page. Archived // history has its own cursor; only an exact current/selected archive stays. const currentId = this.runtime.sessionManager.getSessionId(); @@ -1266,7 +1277,13 @@ export class PiWebAdapter { options.signal?.throwIfAborted(); // Use Pi's catalog and the same workspace visibility as requireSession. // The caller supplies a query, never paths or a broader file capability. - const catalog = await SessionManager.listAll(this.runtime.sessionDirectory); + const startedAt = Date.now(); + const catalogPage = await this.sessionCatalog.list(this.runtime.sessionDirectory, { + signal: options.signal, + maxBodyBytes: WEB_TRANSCRIPT_SEARCH_MAX_TOTAL_BYTES, + deadlineAt: startedAt + WEB_TRANSCRIPT_SEARCH_MAX_DURATION_MS, + }); + const catalog = catalogPage.sessions; options.signal?.throwIfAborted(); const sessions = catalog.filter((session) => (options.includeArchived || !this.archivedSessions.has(resolve(session.path))) && @@ -1279,13 +1296,30 @@ export class PiWebAdapter { source: "web" as const, ...(session.name ? { name: session.name } : {}), })); - return searchWebTranscripts({ + const elapsed = Date.now() - startedAt; + const remainingBytes = WEB_TRANSCRIPT_SEARCH_MAX_TOTAL_BYTES - catalogPage.bodyBytes; + const remainingMs = WEB_TRANSCRIPT_SEARCH_MAX_DURATION_MS - elapsed; + // A zero limit is rejected by the search contract. An exhausted pre-scan + // budget is a partial result, not a second pass over the same files. + const budgetRemains = remainingBytes >= 1 && remainingMs >= 1; + const searchable = budgetRemains ? sessions : []; + const result = await searchWebTranscripts({ query, - sessions, + sessions: searchable, allowedSessionRoots: [this.runtime.sessionDirectory], - allowedWorkspaces: [...new Set(sessions.map((session) => session.cwd))], + allowedWorkspaces: [...new Set(searchable.map((session) => session.cwd))], signal: options.signal, + ...(budgetRemains ? { limits: { maxDurationMs: remainingMs, maxTotalBytes: remainingBytes } } : {}), }); + const exhausted = remainingBytes < 1 ? "byte-limit" as const : remainingMs < 1 ? "time-limit" as const : undefined; + for (const reason of [catalogPage.partial ? catalogPage.partialReason : undefined, exhausted]) { + if (reason && !result.partialReasons.includes(reason)) { + result.partial = true; + result.partialReasons.push(reason); + } + } + result.scannedBytes += catalogPage.bodyBytes; + return result; } async listReadOnlyTerminalSessions( @@ -1633,7 +1667,7 @@ export class PiWebAdapter { workspaceState?: WorkspaceStateSnapshot, groupForks = false, ) { - const sessions = knownSessions ?? (await SessionManager.listAll(this.runtime.sessionDirectory)); + const sessions = knownSessions ?? (await this.sessionCatalog.list(this.runtime.sessionDirectory)).sessions; // The sidebar groups worktree forks under their source project. Files and // runtime admission still resolve the real, separately registered cwd. const project = (session: SessionIdentity & { cwd: string }) => resolve(groupForks ? this.sessionFork(session)?.workspacePath ?? session.cwd : session.cwd); @@ -1651,9 +1685,15 @@ export class PiWebAdapter { if (!session.path || session.path.startsWith("current:")) return undefined; try { const manager = this.runtime.getSessionManagerForRead?.(session.id, session.path) ?? - (this.isCurrentSession(session) ? this.runtime.sessionManager : SessionManager.open(session.path, this.runtime.sessionDirectory)); - if (!matchesSessionIdentity(manager, { expectedSessionId: session.id, expectedSessionPath: session.path })) return undefined; - return readSessionFork(manager.getBranch(), session.id, manager.getCwd()); + (this.isCurrentSession(session) ? this.runtime.sessionManager : undefined); + const branch = manager && matchesSessionIdentity(manager, { expectedSessionId: session.id, expectedSessionPath: session.path }) + ? manager.getBranch() + : this.sessionCatalog.branch(session.path, session.id); + const cwd = manager && matchesSessionIdentity(manager, { expectedSessionId: session.id, expectedSessionPath: session.path }) + ? manager.getCwd() + : this.sessionCatalog.cwd(session.path, session.id); + if (!branch || cwd === undefined) return undefined; + return readSessionFork(branch, session.id, cwd); } catch { return undefined; } @@ -1663,6 +1703,10 @@ export class PiWebAdapter { return resolve(this.sessionFork({ id: this.runtime.sessionManager.getSessionId(), path: this.runtime.sessionManager.getSessionFile() ?? "" })?.workspacePath ?? this.runtime.cwd); } + private readSession(path: string) { + return this.sessionLoader.open(path, this.runtime.sessionDirectory); + } + isCurrentSession(session: Pick) { // A copied JSONL retains its ID, but does not own the active runtime. return matchesSessionIdentity(this.runtime.sessionManager, { @@ -1684,7 +1728,7 @@ export class PiWebAdapter { this.runtime.getSessionManagerForRead?.(summary.id, summary.path) ?? (this.isCurrentSession(summary) ? this.runtime.sessionManager - : SessionManager.open(path)); + : await this.readSession(path)); const branch = manager.getBranch(); const projected = alignHistoryPage( projectEntries(branch, (path) => resolve(summary.cwd, path)), @@ -1711,7 +1755,7 @@ export class PiWebAdapter { const summary = (await this.listSessions(path)).find((session) => session.path === path); if (!summary) return { status: "not_found" as const }; const manager = this.runtime.getSessionManagerForRead?.(summary.id, summary.path) ?? - (this.isCurrentSession(summary) ? this.runtime.sessionManager : SessionManager.open(path)); + (this.isCurrentSession(summary) ? this.runtime.sessionManager : await this.readSession(path)); if (manager.getSessionId() !== sessionId || summary.id !== sessionId) return { status: "changed" as const }; const branch = manager.getBranch(); @@ -1754,7 +1798,7 @@ export class PiWebAdapter { const summary = (await this.listSessions(path)).find((session) => session.path === path); if (!summary) return { status: "not_found" as const }; const manager = this.runtime.getSessionManagerForRead?.(summary.id, summary.path) ?? - (this.isCurrentSession(summary) ? this.runtime.sessionManager : SessionManager.open(path)); + (this.isCurrentSession(summary) ? this.runtime.sessionManager : await this.readSession(path)); if (summary.id !== sessionId || manager.getSessionId() !== sessionId) return { status: "changed" as const }; const branch = manager.getBranch(); const branchById = new Map(branch.map((entry) => [entry.id, entry])); @@ -1781,7 +1825,7 @@ export class PiWebAdapter { const summary = (await this.listSessions(path)).find((session) => session.path === path); if (!summary) return { status: "not_found" as const }; const manager = this.runtime.getSessionManagerForRead?.(summary.id, summary.path) ?? - (this.isCurrentSession(summary) ? this.runtime.sessionManager : SessionManager.open(path)); + (this.isCurrentSession(summary) ? this.runtime.sessionManager : await this.readSession(path)); if (summary.id !== sessionId || manager.getSessionId() !== sessionId) return { status: "changed" as const }; const branch = manager.getBranch(); const branchById = new Map(branch.map((entry) => [entry.id, entry])); @@ -1812,7 +1856,7 @@ export class PiWebAdapter { const summary = await this.requireSession(path).catch(() => undefined); if (!summary) return { status: "not_found" as const }; const manager = this.runtime.getSessionManagerForRead?.(summary.id, summary.path) ?? - (this.isCurrentSession(summary) ? this.runtime.sessionManager : SessionManager.open(path)); + (this.isCurrentSession(summary) ? this.runtime.sessionManager : await this.readSession(path)); if (summary.id !== sessionId || manager.getSessionId() !== sessionId) return { status: "changed" as const }; const branch = manager.getBranch(); const branchById = new Map(branch.map((entry) => [entry.id, entry])); @@ -1856,7 +1900,7 @@ export class PiWebAdapter { const summary = (await this.listSessions(path)).find((session) => session.path === path); if (!summary) return { ok: false, reason: "not_found" }; const manager = this.runtime.getSessionManagerForRead?.(summary.id, summary.path) ?? - (this.isCurrentSession(summary) ? this.runtime.sessionManager : SessionManager.open(path)); + (this.isCurrentSession(summary) ? this.runtime.sessionManager : await this.readSession(path)); if (summary.id !== sessionId || manager.getSessionId() !== sessionId) return { ok: false, reason: "identity_changed" }; const branch = manager.getBranch(); @@ -1887,7 +1931,7 @@ export class PiWebAdapter { const summary = (await this.listSessions(path)).find((session) => session.path === path); if (!summary || summary.id !== sessionId) return undefined; const manager = this.runtime.getSessionManagerForRead?.(summary.id, summary.path) ?? - (this.isCurrentSession(summary) ? this.runtime.sessionManager : SessionManager.open(path)); + (this.isCurrentSession(summary) ? this.runtime.sessionManager : await this.readSession(path)); return manager.getSessionId() === sessionId ? manager : undefined; } @@ -1918,7 +1962,7 @@ export class PiWebAdapter { const summary = (await this.listSessions(path)).find((session) => session.path === path); if (!summary) return { status: "not_found" as const }; const manager = this.runtime.getSessionManagerForRead?.(summary.id, summary.path) ?? - (this.isCurrentSession(summary) ? this.runtime.sessionManager : SessionManager.open(path)); + (this.isCurrentSession(summary) ? this.runtime.sessionManager : await this.readSession(path)); if (summary.id !== sessionId || manager.getSessionId() !== sessionId) return { status: "changed" as const }; const entry = manager.getBranch().find((item) => item.id === entryId); diff --git a/web/adapter/session-catalog.ts b/web/adapter/session-catalog.ts new file mode 100644 index 000000000..2f5b0f89a --- /dev/null +++ b/web/adapter/session-catalog.ts @@ -0,0 +1,540 @@ +import { open, readdir, stat, type FileHandle } from "node:fs/promises"; +import type { Stats } from "node:fs"; +import { basename, join, resolve, sep } from "node:path"; +import { StringDecoder } from "node:string_decoder"; +import type { SessionEntry } from "@earendil-works/pi-coding-agent"; +import { WEB_SESSION_FORK } from "../protocol/session-fork.ts"; +import { WEB_TURN_TIMING_ENTRY } from "../protocol/turn-timing.ts"; + +/** Archive search only inspects this prefix of the first user message. */ +const CATALOG_FIRST_MESSAGE_CHARS = 2_000; +const STAT_CONCURRENCY = 64; +const READ_CONCURRENCY = 10; +const READ_CHUNK_BYTES = 64 * 1024; + +export type CatalogSession = { + path: string; + id: string; + cwd: string; + name?: string; + parentSessionPath?: string; + created: Date; + modified: Date; + messageCount: number; + firstMessage: string; +}; + +export type CatalogPartialReason = "byte-limit" | "time-limit"; + +export type CatalogPage = { + sessions: CatalogSession[]; + partial: boolean; + partialReason?: CatalogPartialReason; + bodyBytes: number; +}; + +export type CatalogListOptions = { + signal?: AbortSignal; + maxBodyBytes?: number; + deadlineAt?: number; + now?: () => number; +}; + +type FileIdentity = { + dev: bigint; + ino: bigint; + size: bigint; + mtimeNs: bigint; + ctimeNs: bigint; +}; + +type CompactNode = { + id: string; + parentId: string | null; + type: string; + role?: string; + stopReason?: string; + customType?: string; + data?: unknown; +}; + +type CachedFile = { + identity: FileIdentity; + session: CatalogSession; + nodes: CompactNode[]; + byId: Map; + leafId: string | null; + resolvedCwd: string; +}; + +type SharedScan = { + controller: AbortController; + signals: AbortSignal[]; + detached: boolean; + promise: Promise; +}; + +type ListedFile = { + path: string; + identity: FileIdentity; + mtimeMs: number; +}; + +function inodeKey(identity: FileIdentity) { + return `${identity.dev}:${identity.ino}`; +} + +function sameIdentity(left: FileIdentity, right: FileIdentity) { + return left.dev === right.dev && left.ino === right.ino && left.size === right.size && + left.mtimeNs === right.mtimeNs && left.ctimeNs === right.ctimeNs; +} + +function identityOf(stats: Stats): FileIdentity { + return { + dev: stats.dev, + ino: stats.ino, + size: stats.size, + mtimeNs: stats.mtimeNs, + ctimeNs: stats.ctimeNs, + }; +} + +function follow(promise: Promise, signal?: AbortSignal): Promise { + if (!signal) return promise; + if (signal.aborted) return Promise.reject(signal.reason); + return new Promise((resolvePromise, reject) => { + const onAbort = () => reject(signal.reason); + signal.addEventListener("abort", onAbort, { once: true }); + promise.then( + (value) => { + signal.removeEventListener("abort", onAbort); + resolvePromise(value); + }, + (error) => { + signal.removeEventListener("abort", onAbort); + reject(error); + }, + ); + }); +} + +async function withSignal(work: Promise, signal?: AbortSignal) { + return follow(work, signal); +} + +function textOf(message: Record) { + const content = message.content; + if (typeof content === "string") return content; + if (!Array.isArray(content)) return ""; + let text = ""; + for (const block of content) { + if (!block || typeof block !== "object" || Array.isArray(block)) continue; + const part = block as Record; + if (part.type === "text" && typeof part.text === "string") { + if (text) text += " "; + text += part.text; + } + } + return text; +} + +function activityTime(entry: Record, message: Record) { + if (message.role !== "user" && message.role !== "assistant") return undefined; + if (typeof message.timestamp === "number") return message.timestamp; + const parsed = new Date(String(entry.timestamp ?? "")).getTime(); + return Number.isNaN(parsed) ? undefined : parsed; +} + +function materialize(node: CompactNode): SessionEntry { + if (node.type === "message") { + return { + type: "message", + id: node.id, + parentId: node.parentId, + timestamp: "", + message: { + role: node.role, + content: "", + ...(node.stopReason ? { stopReason: node.stopReason } : {}), + }, + } as SessionEntry; + } + if (node.type === "custom") { + return { + type: "custom", + id: node.id, + parentId: node.parentId, + timestamp: "", + customType: node.customType ?? "", + data: node.data, + } as SessionEntry; + } + return { + type: node.type, + id: node.id, + parentId: node.parentId, + timestamp: "", + } as SessionEntry; +} + +function isAbort(error: unknown, signal?: AbortSignal) { + return Boolean(signal?.aborted && (error === signal.reason || (error instanceof Error && error.name === "AbortError"))); +} + +/** + * File-identity cache for Pi's session directory listing. + * Unchanged files are not read again. Pi's branch readers still interpret the + * retained structural entries; message bodies are not kept as a second transcript. + */ +export class SessionCatalog { + private readonly files = new Map(); + private readonly byInode = new Map(); + private readonly shared = new Map(); + + async list(directory: string, options: CatalogListOptions = {}): Promise { + const canonical = resolve(directory); + options.signal?.throwIfAborted(); + const bounded = options.maxBodyBytes !== undefined || options.deadlineAt !== undefined; + if (bounded) return this.scan(canonical, options); + const existing = this.shared.get(canonical); + if (existing) { + this.watch(existing, options.signal); + return follow(existing.promise, options.signal); + } + const scan: SharedScan = { + controller: new AbortController(), + signals: [], + detached: false, + promise: undefined as never, + }; + this.watch(scan, options.signal); + this.shared.set(canonical, scan); + scan.promise = this.scan(canonical, { ...options, signal: scan.controller.signal }); + void scan.promise.then( + () => { + if (this.shared.get(canonical) === scan) this.shared.delete(canonical); + }, + () => { + if (this.shared.get(canonical) === scan) this.shared.delete(canonical); + }, + ); + return follow(scan.promise, options.signal); + } + + branch(path: string, sessionId: string, fromId?: string) { + const cached = this.cached(path, sessionId); + if (!cached) return undefined; + const start = fromId ?? cached.leafId; + const nodes: CompactNode[] = []; + const seen = new Set(); + let current = start ? cached.byId.get(start) : undefined; + while (current && !seen.has(current.id)) { + seen.add(current.id); + nodes.push(current); + current = current.parentId ? cached.byId.get(current.parentId) : undefined; + } + nodes.reverse(); + return nodes.map(materialize); + } + + entries(path: string, sessionId: string) { + const cached = this.cached(path, sessionId); + return cached ? cached.nodes.map(materialize) : undefined; + } + + cwd(path: string, sessionId: string) { + return this.cached(path, sessionId)?.resolvedCwd; + } + + private cached(path: string, sessionId: string) { + const cached = this.files.get(resolve(path)); + return cached?.session.id === sessionId ? cached : undefined; + } + + private watch(scan: SharedScan, signal?: AbortSignal) { + if (!signal) { + scan.detached = true; + return; + } + scan.signals.push(signal); + const onAbort = () => { + if (scan.detached || scan.controller.signal.aborted) return; + if (scan.signals.every((item) => item.aborted)) scan.controller.abort(signal.reason); + }; + if (signal.aborted) onAbort(); + else signal.addEventListener("abort", onAbort, { once: true }); + } + + private remember(path: string, cached: CachedFile) { + const previous = this.files.get(path); + if (previous) { + const previousKey = inodeKey(previous.identity); + if (this.byInode.get(previousKey) === path) this.byInode.delete(previousKey); + } + this.files.set(path, cached); + this.byInode.set(inodeKey(cached.identity), path); + } + + private forget(path: string) { + const cached = this.files.get(path); + if (!cached) return; + this.files.delete(path); + const key = inodeKey(cached.identity); + if (this.byInode.get(key) === path) this.byInode.delete(key); + } + + private async scan(directory: string, options: CatalogListOptions): Promise { + const signal = options.signal; + const now = options.now ?? Date.now; + const empty: CatalogPage = { sessions: [], partial: false, bodyBytes: 0 }; + let names: string[]; + try { + signal?.throwIfAborted(); + names = await withSignal(readdir(directory), signal); + } catch (error) { + if (isAbort(error, signal)) throw error; + signal?.throwIfAborted(); + if ((error as NodeJS.ErrnoException).code === "ENOENT") return empty; + throw error; + } + const paths = names + .filter((name) => name.endsWith(".jsonl")) + .sort((left, right) => right.localeCompare(left)) + .map((name) => join(directory, name)); + const listed: ListedFile[] = []; + let statIndex = 0; + const statWorker = async () => { + while (statIndex < paths.length) { + signal?.throwIfAborted(); + const path = paths[statIndex++]!; + let stats: Stats; + try { + stats = await withSignal(stat(path, { bigint: true }), signal); + } catch (error) { + if (isAbort(error, signal)) throw error; + continue; + } + if (!stats.isFile()) continue; + listed.push({ + path, + identity: identityOf(stats), + mtimeMs: Number(stats.mtimeMs), + }); + } + }; + try { + await Promise.all(Array.from({ length: Math.min(STAT_CONCURRENCY, paths.length) }, () => statWorker())); + } catch (error) { + if (isAbort(error, signal)) throw error; + throw error; + } + const seen = new Set(listed.map((file) => file.path)); + const fresh: CachedFile[] = []; + const stale: ListedFile[] = []; + for (const file of listed) { + const cached = this.files.get(file.path); + if (cached && sameIdentity(cached.identity, file.identity)) { + fresh.push(cached); + continue; + } + const key = inodeKey(file.identity); + const previousPath = this.byInode.get(key); + const previous = previousPath ? this.files.get(previousPath) : undefined; + if (previous && previousPath && sameIdentity(previous.identity, file.identity) && !seen.has(previousPath)) { + this.files.delete(previousPath); + previous.session = { ...previous.session, path: file.path }; + this.remember(file.path, previous); + fresh.push(previous); + continue; + } + stale.push(file); + } + const prefix = directory.endsWith(sep) ? directory : `${directory}${sep}`; + for (const path of [...this.files.keys()]) { + if ((path === directory || path.startsWith(prefix)) && !seen.has(path)) this.forget(path); + } + stale.sort((left, right) => right.mtimeMs - left.mtimeMs || basename(right.path).localeCompare(basename(left.path))); + const loaded: CachedFile[] = []; + let bodyBytes = 0; + let partial = false; + let partialReason: CatalogPartialReason | undefined; + let stop = false; + let readIndex = 0; + const readWorker = async () => { + while (!stop) { + signal?.throwIfAborted(); + if (options.deadlineAt !== undefined && now() >= options.deadlineAt) { + stop = true; + partial = true; + partialReason ??= "time-limit"; + return; + } + const index = readIndex++; + if (index >= stale.length) return; + const file = stale[index]!; + const size = Number(file.identity.size); + if (options.maxBodyBytes !== undefined && bodyBytes + size > options.maxBodyBytes) { + stop = true; + partial = true; + partialReason ??= "byte-limit"; + return; + } + bodyBytes += size; + const parsed = await this.parseFile(file.path, file.identity, signal); + if (parsed) loaded.push(parsed); + } + }; + try { + await Promise.all(Array.from({ length: Math.min(READ_CONCURRENCY, stale.length) }, () => readWorker())); + } catch (error) { + if (isAbort(error, signal)) throw error; + throw error; + } + const sessions = [...fresh, ...loaded] + .map((file) => ({ ...file.session })) + .sort((left, right) => right.modified.getTime() - left.modified.getTime() || basename(right.path).localeCompare(basename(left.path))); + return { + sessions, + partial, + ...(partialReason ? { partialReason } : {}), + bodyBytes, + }; + } + + private async parseFile(path: string, identity: FileIdentity, signal?: AbortSignal) { + const opening = open(path, "r"); + let handle: FileHandle; + try { + handle = await withSignal(opening, signal); + } catch (error) { + // Cancellation can win before open settles. Close that handle if it appears later. + void opening.then((opened) => opened.close()).catch(() => undefined); + if (isAbort(error, signal)) throw error; + return null; + } + const decoder = new StringDecoder("utf8"); + const buffer = Buffer.allocUnsafe(READ_CHUNK_BYTES); + let pending = ""; + let header: Record | undefined; + let name: string | undefined; + let messageCount = 0; + let firstMessage = ""; + let lastActivity: number | undefined; + const nodes: CompactNode[] = []; + const byId = new Map(); + const consume = (line: string) => { + if (!line.trim()) return; + let value: unknown; + try { + value = JSON.parse(line); + } catch { + return; + } + if (!value || typeof value !== "object" || Array.isArray(value)) return; + const entry = value as Record; + if (!header) { + if (entry.type !== "session" || typeof entry.id !== "string") { + header = { invalid: true }; + } else header = entry; + return; + } + if (header.invalid) return; + if (entry.type === "session_info") { + name = typeof entry.name === "string" ? entry.name.trim() || undefined : undefined; + } + const id = typeof entry.id === "string" ? entry.id : undefined; + const parentId = typeof entry.parentId === "string" ? entry.parentId : null; + if (entry.type === "message") { + messageCount++; + const message = entry.message && typeof entry.message === "object" && !Array.isArray(entry.message) + ? entry.message as Record + : undefined; + if (message) { + const activity = activityTime(entry, message); + if (typeof activity === "number") lastActivity = Math.max(lastActivity ?? 0, activity); + const text = textOf(message); + if (!firstMessage && message.role === "user" && text) { + firstMessage = text.slice(0, CATALOG_FIRST_MESSAGE_CHARS); + } + if (id) { + const node: CompactNode = { + id, + parentId, + type: "message", + ...(typeof message.role === "string" ? { role: message.role } : {}), + ...(typeof message.stopReason === "string" ? { stopReason: message.stopReason } : {}), + }; + nodes.push(node); + byId.set(id, node); + } + } + return; + } + if (!id) return; + const keepData = entry.type === "custom" && + (entry.customType === WEB_SESSION_FORK || entry.customType === WEB_TURN_TIMING_ENTRY); + const node: CompactNode = { + id, + parentId, + type: typeof entry.type === "string" ? entry.type : "custom", + ...(entry.type === "custom" && typeof entry.customType === "string" ? { customType: entry.customType } : {}), + ...(keepData ? { data: entry.data } : {}), + }; + nodes.push(node); + byId.set(id, node); + }; + try { + while (true) { + signal?.throwIfAborted(); + const { bytesRead } = await withSignal(handle.read(buffer, 0, buffer.length, null), signal); + if (bytesRead === 0) break; + pending += decoder.write(buffer.subarray(0, bytesRead)); + let start = 0; + let newline = pending.indexOf("\n", start); + while (newline !== -1) { + consume(pending.slice(start, newline)); + start = newline + 1; + newline = pending.indexOf("\n", start); + } + pending = pending.slice(start); + } + pending += decoder.end(); + if (pending) consume(pending); + } catch (error) { + if (isAbort(error, signal)) throw error; + return null; + } finally { + await handle.close().catch(() => undefined); + } + if (!header || header.invalid === true || typeof header.id !== "string") return null; + const headerTime = typeof header.timestamp === "string" ? new Date(header.timestamp).getTime() : Number.NaN; + const modified = typeof lastActivity === "number" && lastActivity > 0 + ? new Date(lastActivity) + : !Number.isNaN(headerTime) + ? new Date(headerTime) + : new Date(Number(identity.mtimeNs / 1_000_000n)); + const cwd = typeof header.cwd === "string" ? header.cwd : ""; + const parentSessionPath = typeof header.parentSession === "string" ? header.parentSession : undefined; + const session: CatalogSession = { + path, + id: header.id, + cwd, + ...(name ? { name } : {}), + ...(parentSessionPath ? { parentSessionPath } : {}), + created: new Date(typeof header.timestamp === "string" ? header.timestamp : Number.NaN), + modified, + messageCount, + firstMessage: firstMessage || "(no messages)", + }; + const cached: CachedFile = { + identity, + session, + nodes, + byId, + leafId: nodes.at(-1)?.id ?? null, + resolvedCwd: cwd ? resolve(cwd) : resolve(process.cwd()), + }; + this.remember(path, cached); + return cached; + } +} diff --git a/web/adapter/session-file-worker.ts b/web/adapter/session-file-worker.ts new file mode 100644 index 000000000..1e836dc46 --- /dev/null +++ b/web/adapter/session-file-worker.ts @@ -0,0 +1,14 @@ +import { parentPort } from "node:worker_threads"; +import { parseSessionEntries } from "@earendil-works/pi-coding-agent"; + +parentPort?.on("message", (message: { id: number; bytes: Uint8Array }) => { + try { + const text = Buffer.from(message.bytes).toString("utf8"); + parentPort?.postMessage({ id: message.id, entries: parseSessionEntries(text) }); + } catch (error) { + parentPort?.postMessage({ + id: message.id, + error: error instanceof Error ? error.message : String(error), + }); + } +}); diff --git a/web/adapter/session-loader.ts b/web/adapter/session-loader.ts new file mode 100644 index 000000000..6a84d49f6 --- /dev/null +++ b/web/adapter/session-loader.ts @@ -0,0 +1,170 @@ +import { readFile, stat } from "node:fs/promises"; +import type { Stats } from "node:fs"; +import { resolve } from "node:path"; +import { Worker } from "node:worker_threads"; +import { SessionManager, type FileEntry } from "@earendil-works/pi-coding-agent"; + +const MAX_OPEN_SESSIONS = 4; + +type FileIdentity = { + dev: bigint; + ino: bigint; + size: bigint; + mtimeNs: bigint; + ctimeNs: bigint; +}; + +type WorkerResponse = { + id: number; + entries?: FileEntry[]; + error?: string; +}; + +type CachedSession = { + identity: FileIdentity; + manager: SessionManager; +}; + +let worker: Worker | undefined; +let nextId = 0; +const pending = new Map void; reject: (error: Error) => void }>(); + +function identityOf(stats: Stats): FileIdentity { + return { + dev: stats.dev, + ino: stats.ino, + size: stats.size, + mtimeNs: stats.mtimeNs, + ctimeNs: stats.ctimeNs, + }; +} + +function sameIdentity(left: FileIdentity, right: FileIdentity) { + return left.dev === right.dev && left.ino === right.ino && left.size === right.size && + left.mtimeNs === right.mtimeNs && left.ctimeNs === right.ctimeNs; +} + +function releaseWorker(current: Worker) { + if (pending.size === 0) current.unref(); +} + +function fileWorker() { + if (worker) return worker; + worker = new Worker(new URL("./session-file-worker.ts", import.meta.url), { + execArgv: ["--experimental-strip-types"], + }); + // An idle parser must not keep the process, or a finished request never ends. + worker.unref(); + const current = worker; + current.on("message", (message: WorkerResponse) => { + const request = pending.get(message.id); + if (!request) return; + pending.delete(message.id); + releaseWorker(current); + if (message.error !== undefined) request.reject(new Error(message.error)); + else request.resolve(message.entries ?? []); + }); + current.on("error", (error) => { + const requests = [...pending.values()]; + pending.clear(); + if (worker === current) worker = undefined; + current.unref(); + for (const request of requests) request.reject(error); + }); + return current; +} + +function parseEntries(bytes: Buffer) { + const id = nextId++; + const current = fileWorker(); + current.ref(); + return new Promise((resolvePromise, reject) => { + const fail = (error: Error) => { + if (pending.get(id)) pending.delete(id); + releaseWorker(current); + reject(error); + }; + pending.set(id, { + resolve: (entries) => { + resolvePromise(entries); + }, + reject: fail, + }); + try { + current.postMessage({ id, bytes }); + } catch (error) { + fail(error instanceof Error ? error : new Error(String(error))); + } + }); +} + +function readableSession(path: string, sessionDir: string, entries: FileEntry[]) { + const header = entries.find((entry) => entry.type === "session"); + const headerCwd = header && typeof header.cwd === "string" ? header.cwd : ""; + const manager = SessionManager.inMemory(headerCwd || process.cwd(), undefined, entries); + const file = resolve(path); + const directory = resolve(sessionDir); + manager.getSessionFile = () => file; + manager.getSessionDir = () => directory; + return manager; +} + +/** + * Opens a non-current Session off the request thread and reuses it while the + * file identity is unchanged. Pi remains the parser and the branch authority. + */ +export class SessionLoader { + private readonly cache = new Map(); + private readonly recent: string[] = []; + + async open(path: string, sessionDir: string) { + const resolved = resolve(path); + let stats: Stats; + try { + stats = await stat(resolved, { bigint: true }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return SessionManager.open(path, sessionDir); + throw error; + } + if (!stats.isFile()) return SessionManager.open(path, sessionDir); + const identity = identityOf(stats); + const cached = this.cache.get(resolved); + if (cached && sameIdentity(cached.identity, identity)) { + this.touch(resolved); + return cached.manager; + } + const entries = await this.readStable(resolved, identity); + if (entries.length === 0 || entries[0]?.type !== "session" || typeof entries[0].id !== "string") { + throw new Error(`Session file is not a valid pi session: ${resolved}`); + } + const manager = readableSession(resolved, sessionDir, entries); + const after = await stat(resolved, { bigint: true }).catch(() => undefined); + if (after?.isFile() && sameIdentity(identity, identityOf(after))) { + this.cache.set(resolved, { identity, manager }); + this.touch(resolved); + } + return manager; + } + + private touch(path: string) { + const index = this.recent.indexOf(path); + if (index >= 0) this.recent.splice(index, 1); + this.recent.push(path); + while (this.recent.length > MAX_OPEN_SESSIONS) { + const evicted = this.recent.shift(); + if (evicted) this.cache.delete(evicted); + } + } + + private async readStable(path: string, identity: FileIdentity) { + const first = await this.readEntries(path); + const after = await stat(path, { bigint: true }).catch(() => undefined); + if (!after?.isFile() || sameIdentity(identity, identityOf(after))) return first; + return this.readEntries(path); + } + + private async readEntries(path: string) { + // The bytes are read asynchronously here; Pi parses them off this thread. + return parseEntries(await readFile(path)); + } +} From b3f36867b2b3c9b1dab1c213c754d2d65a2a41d3 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 16 Oct 2026 14:34:36 +0000 Subject: [PATCH 2/3] fix(web): type session file identity without generic Stats The installed Node types do not treat fs.Stats as generic. Keep the bigint file identity structural, and format the session read tests. Co-authored-by: tt-a1i --- tests/web/pi-adapter.test.ts | 40 +++-- tests/web/session-read-cost.test.ts | 219 ++++++++++++++++++++-------- web/adapter/session-catalog.ts | 19 +-- web/adapter/session-loader.ts | 21 ++- 4 files changed, 204 insertions(+), 95 deletions(-) diff --git a/tests/web/pi-adapter.test.ts b/tests/web/pi-adapter.test.ts index 97ff9966f..96b29000d 100644 --- a/tests/web/pi-adapter.test.ts +++ b/tests/web/pi-adapter.test.ts @@ -2026,20 +2026,30 @@ test("archive scan continuations reach old matches after an empty bounded window text: string, ) => { const timestamp = new Date(createdMs).toISOString(); - return `${JSON.stringify({ type: "session", version: 3, id, timestamp, cwd })}\n${JSON.stringify({ - type: "message", - id: "m", - parentId: null, - timestamp, - message: { role: "user", content: text, timestamp: modifiedMs }, - })}\n`; + return `${JSON.stringify({ type: "session", version: 3, id, timestamp, cwd })}\n${JSON.stringify( + { + type: "message", + id: "m", + parentId: null, + timestamp, + message: { role: "user", content: text, timestamp: modifiedMs }, + }, + )}\n`; }; const writes: Promise[] = []; for (const session of [...active, ...older]) { - writes.push(writeFile( - session.path, - sessionFile(session.id, session.cwd, session.created.getTime(), session.modified.getTime(), session.firstMessage), - )); + writes.push( + writeFile( + session.path, + sessionFile( + session.id, + session.cwd, + session.created.getTime(), + session.modified.getTime(), + session.firstMessage, + ), + ), + ); if (writes.length === 64) { await Promise.all(writes); writes.length = 0; @@ -2097,7 +2107,13 @@ test("archive scan continuations reach old matches after an empty bounded window } await writeFile( older[0]!.path, - sessionFile("older-a", project, 200_000, older[0]!.modified.getTime(), "old needle A"), + sessionFile( + "older-a", + project, + 200_000, + older[0]!.modified.getTime(), + "old needle A", + ), ); const globallyNewest = await adapter.listArchivedSessions({ sort: "created", diff --git a/tests/web/session-read-cost.test.ts b/tests/web/session-read-cost.test.ts index 10c6f83e7..bfe2ede83 100644 --- a/tests/web/session-read-cost.test.ts +++ b/tests/web/session-read-cost.test.ts @@ -9,12 +9,19 @@ import { SessionManager } from "@earendil-works/pi-coding-agent"; import { PiWebAdapter } from "../../web/adapter/pi-adapter.ts"; import { readSessionFork } from "../../web/protocol/session-fork.ts"; import { WEB_SESSION_FORK } from "../../web/protocol/session-fork.ts"; -import { readSessionTurnEvidence, WEB_TURN_TIMING_ENTRY } from "../../web/protocol/turn-timing.ts"; +import { + readSessionTurnEvidence, + WEB_TURN_TIMING_ENTRY, +} from "../../web/protocol/turn-timing.ts"; import { WEB_TRANSCRIPT_SEARCH_MAX_TOTAL_BYTES } from "../../web/search/transcript-search.ts"; import { projectWebModelSearch } from "../../web/runtime/model-discovery.ts"; import type { WebRuntimeController } from "../../web/runtime/types.ts"; -function runtimeFor(cwd: string, sessionDirectory: string, sessionManager: SessionManager): WebRuntimeController { +function runtimeFor( + cwd: string, + sessionDirectory: string, + sessionManager: SessionManager, +): WebRuntimeController { return { cwd, workspaceSelected: true, @@ -24,7 +31,10 @@ function runtimeFor(cwd: string, sessionDirectory: string, sessionManager: Sessi getActiveTurn: () => undefined, cancelTurn: async (options) => ({ ...options, state: "stale-turn" }), sendPrompt: async () => ({ pendingFollowUps: 0 }), - newSession: async () => ({ cancelled: false, sessionId: sessionManager.getSessionId() }), + newSession: async () => ({ + cancelled: false, + sessionId: sessionManager.getSessionId(), + }), switchSession: async () => ({ cancelled: false }), listModels: () => [], searchModels: (query, limit) => projectWebModelSearch([], query, limit), @@ -45,7 +55,11 @@ const zeroUsage = { cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }; -function persistSession(manager: SessionManager, content: string, timestamp: number) { +function persistSession( + manager: SessionManager, + content: string, + timestamp: number, +) { manager.appendMessage({ role: "user", content, timestamp }); return manager.appendMessage({ role: "assistant", @@ -75,14 +89,21 @@ test("catalog metadata matches Pi and unchanged files are not read again", async branched.branch(forkPoint); persistSession(branched, "replacement", 32); const oracle = await SessionManager.listAll(directory); - const adapter = new PiWebAdapter(runtimeFor(root, directory, SessionManager.inMemory(root))); + const adapter = new PiWebAdapter( + runtimeFor(root, directory, SessionManager.inMemory(root)), + ); const originalOpen = fsPromises.open; const opened: string[] = []; - t.mock.method(fsPromises, "open", async (...args: Parameters) => { - const path = String(args[0]); - if (path.startsWith(directory) && path.endsWith(".jsonl")) opened.push(path); - return originalOpen(...args); - }); + t.mock.method( + fsPromises, + "open", + async (...args: Parameters) => { + const path = String(args[0]); + if (path.startsWith(directory) && path.endsWith(".jsonl")) + opened.push(path); + return originalOpen(...args); + }, + ); syncBuiltinESMExports(); t.after(() => { t.mock.restoreAll(); @@ -92,7 +113,10 @@ test("catalog metadata matches Pi and unchanged files are not read again", async const listed = await adapter.listSessions(); const files = listed.filter((session) => session.path.endsWith(".jsonl")); const byPath = new Map(files.map((session) => [session.path, session])); - assert.deepEqual(files.map((session) => session.id), oracle.map((session) => session.id)); + assert.deepEqual( + files.map((session) => session.id), + oracle.map((session) => session.id), + ); for (const expected of oracle) { const actual = byPath.get(expected.path); assert.ok(actual); @@ -106,7 +130,9 @@ test("catalog metadata matches Pi and unchanged files are not read again", async assert.equal(new Set(firstRead).size, oracle.length); const again = await adapter.listSessions(); assert.deepEqual( - again.filter((session) => session.path.endsWith(".jsonl")).map((session) => session.id), + again + .filter((session) => session.path.endsWith(".jsonl")) + .map((session) => session.id), files.map((session) => session.id), ); assert.deepEqual(opened, firstRead); @@ -121,23 +147,43 @@ test("catalog metadata matches Pi and unchanged files are not read again", async const beforeRename = opened.length; await rename(from, to); const renamed = await adapter.listSessions(); - assert.equal(renamed.some((session) => session.path === from), false); - assert.equal(renamed.find((session) => session.path === to)?.id, named.getSessionId()); - const renameReads = opened.slice(beforeRename).filter((path) => path !== to && path !== from); + assert.equal( + renamed.some((session) => session.path === from), + false, + ); + assert.equal( + renamed.find((session) => session.path === to)?.id, + named.getSessionId(), + ); + const renameReads = opened + .slice(beforeRename) + .filter((path) => path !== to && path !== from); assert.deepEqual(renameReads, []); const beforeDelete = opened.length; await rm(plain.getSessionFile()!); const remaining = await adapter.listSessions(); - assert.equal(remaining.some((session) => session.path === plain.getSessionFile()), false); + assert.equal( + remaining.some((session) => session.path === plain.getSessionFile()), + false, + ); assert.deepEqual(opened.slice(beforeDelete), []); const replaced = branched.getSessionFile()!; const beforeReplace = opened.length; - await writeFile(replaced, `${JSON.stringify({ type: "session", version: 3, id: "replaced-id", timestamp: "2026-10-15T00:00:00.000Z", cwd: root })}\n${JSON.stringify({ type: "message", id: "m", parentId: null, timestamp: "2026-10-15T00:00:00.000Z", message: { role: "user", content: "replaced body", timestamp: 50 } })}\n`); + await writeFile( + replaced, + `${JSON.stringify({ type: "session", version: 3, id: "replaced-id", timestamp: "2026-10-15T00:00:00.000Z", cwd: root })}\n${JSON.stringify({ type: "message", id: "m", parentId: null, timestamp: "2026-10-15T00:00:00.000Z", message: { role: "user", content: "replaced body", timestamp: 50 } })}\n`, + ); const afterReplace = await adapter.listSessions(); - assert.equal(afterReplace.some((session) => session.id === branched.getSessionId()), false); - assert.equal(afterReplace.find((session) => session.path === replaced)?.firstMessage, "replaced body"); + assert.equal( + afterReplace.some((session) => session.id === branched.getSessionId()), + false, + ); + assert.equal( + afterReplace.find((session) => session.path === replaced)?.firstMessage, + "replaced body", + ); assert.deepEqual(opened.slice(beforeReplace), [replaced]); }); @@ -148,7 +194,11 @@ test("non-current history keeps Pi's branch without a synchronous open or a seco const manager = SessionManager.create(root, directory); persistSession(manager, "kept", 1); const forkPoint = manager.getLeafId()!; - const abandoned = manager.appendMessage({ role: "user", content: "abandoned", timestamp: 2 }); + const abandoned = manager.appendMessage({ + role: "user", + content: "abandoned", + timestamp: 2, + }); manager.branch(forkPoint); const replacement = persistSession(manager, "replacement", 3); const assistant = manager.getLeafId()!; @@ -167,7 +217,11 @@ test("non-current history keeps Pi's branch without a synchronous open or a seco manager.appendCustomEntry(WEB_SESSION_FORK, { version: 1, sessionId: manager.getSessionId(), - source: { sessionId: "source-session", sessionPath: join(root, "source.jsonl"), entryId: "source-entry" }, + source: { + sessionId: "source-session", + sessionPath: join(root, "source.jsonl"), + entryId: "source-entry", + }, environment: "local", cwd: root, workspacePath: join(root, "project"), @@ -176,9 +230,19 @@ test("non-current history keeps Pi's branch without a synchronous open or a seco const oracle = SessionManager.open(path, directory); const oracleBranch = oracle.getBranch(); const oracleIds = oracleBranch.map((entry) => entry.id); - const oracleFork = readSessionFork(oracleBranch, oracle.getSessionId(), oracle.getCwd()); - const oracleEvidence = readSessionTurnEvidence(oracleBranch, oracle.getSessionId(), false); - const adapter = new PiWebAdapter(runtimeFor(root, directory, SessionManager.inMemory(root))); + const oracleFork = readSessionFork( + oracleBranch, + oracle.getSessionId(), + oracle.getCwd(), + ); + const oracleEvidence = readSessionTurnEvidence( + oracleBranch, + oracle.getSessionId(), + false, + ); + const adapter = new PiWebAdapter( + runtimeFor(root, directory, SessionManager.inMemory(root)), + ); const originalOpen = SessionManager.open; const originalRead = fsPromises.readFile; let syncOpens = 0; @@ -187,10 +251,14 @@ test("non-current history keeps Pi's branch without a synchronous open or a seco syncOpens++; return originalOpen(...args); }) as typeof SessionManager.open; - t.mock.method(fsPromises, "readFile", async (...args: Parameters) => { - if (args[0] === path) bodyReads++; - return originalRead(...args); - }); + t.mock.method( + fsPromises, + "readFile", + async (...args: Parameters) => { + if (args[0] === path) bodyReads++; + return originalRead(...args); + }, + ); syncBuiltinESMExports(); t.after(() => { SessionManager.open = originalOpen; @@ -213,11 +281,17 @@ test("non-current history keeps Pi's branch without a synchronous open or a seco assert.equal(projected.has(replacement), true); const anchor = session.entries[0]?.id; assert.ok(anchor); - const page = await adapter.getSessionHistory(manager.getSessionId(), path, oracle.getLeafId()!, anchor); + const page = await adapter.getSessionHistory( + manager.getSessionId(), + path, + oracle.getLeafId()!, + anchor, + ); assert.equal(page.status, "ok"); assert.equal(bodyReads, 1); if (page.status === "ok") { - for (const entry of page.session.entries) assert.ok(oracleIds.includes(entry.id)); + for (const entry of page.session.entries) + assert.ok(oracleIds.includes(entry.id)); } }); @@ -229,7 +303,9 @@ test("concurrent catalog reads share one scan", async (t) => { const manager = SessionManager.create(root, directory); persistSession(manager, id, 1); } - const adapter = new PiWebAdapter(runtimeFor(root, directory, SessionManager.inMemory(root))); + const adapter = new PiWebAdapter( + runtimeFor(root, directory, SessionManager.inMemory(root)), + ); const originalOpen = fsPromises.open; let opens = 0; let releaseHold: () => void = () => {}; @@ -240,17 +316,21 @@ test("concurrent catalog reads share one scan", async (t) => { const started = new Promise((resolve) => { entered = resolve; }); - t.mock.method(fsPromises, "open", async (...args: Parameters) => { - const path = String(args[0]); - if (path.startsWith(directory) && path.endsWith(".jsonl")) { - opens++; - if (opens === 1) { - entered(); - await hold; + t.mock.method( + fsPromises, + "open", + async (...args: Parameters) => { + const path = String(args[0]); + if (path.startsWith(directory) && path.endsWith(".jsonl")) { + opens++; + if (opens === 1) { + entered(); + await hold; + } } - } - return originalOpen(...args); - }); + return originalOpen(...args); + }, + ); syncBuiltinESMExports(); t.after(() => { releaseHold(); @@ -276,20 +356,29 @@ test("search cancellation and byte budget include the catalog pre-scan", async ( const manager = SessionManager.create(root, small); persistSession(manager, `needle ${index}`, index + 1); } - const adapter = new PiWebAdapter(runtimeFor(root, small, SessionManager.inMemory(root))); + const adapter = new PiWebAdapter( + runtimeFor(root, small, SessionManager.inMemory(root)), + ); const controller = new AbortController(); const originalOpen = fsPromises.open; let opens = 0; - t.mock.method(fsPromises, "open", async (...args: Parameters) => { - const path = String(args[0]); - if (path.startsWith(small) && path.endsWith(".jsonl")) { - opens++; - if (opens === 1) controller.abort(); - } - return originalOpen(...args); - }); + t.mock.method( + fsPromises, + "open", + async (...args: Parameters) => { + const path = String(args[0]); + if (path.startsWith(small) && path.endsWith(".jsonl")) { + opens++; + if (opens === 1) controller.abort(); + } + return originalOpen(...args); + }, + ); syncBuiltinESMExports(); - await assert.rejects(adapter.searchTranscripts("needle", { signal: controller.signal }), /abort/i); + await assert.rejects( + adapter.searchTranscripts("needle", { signal: controller.signal }), + /abort/i, + ); assert.ok(opens < 6); t.mock.restoreAll(); syncBuiltinESMExports(); @@ -297,21 +386,31 @@ test("search cancellation and byte budget include the catalog pre-scan", async ( const bulky = join(directory, "bulky"); await mkdir(bulky, { recursive: true }); const fileBytes = 1024 * 1024; - const fileCount = Math.floor(WEB_TRANSCRIPT_SEARCH_MAX_TOTAL_BYTES / fileBytes) + 2; - await Promise.all(Array.from({ length: fileCount }, (_, index) => - writeFile(join(bulky, `${index}.jsonl`), Buffer.alloc(fileBytes, 120)))); + const fileCount = + Math.floor(WEB_TRANSCRIPT_SEARCH_MAX_TOTAL_BYTES / fileBytes) + 2; + await Promise.all( + Array.from({ length: fileCount }, (_, index) => + writeFile(join(bulky, `${index}.jsonl`), Buffer.alloc(fileBytes, 120)), + ), + ); let bulkyOpens = 0; - t.mock.method(fsPromises, "open", async (...args: Parameters) => { - const path = String(args[0]); - if (path.startsWith(bulky) && path.endsWith(".jsonl")) bulkyOpens++; - return originalOpen(...args); - }); + t.mock.method( + fsPromises, + "open", + async (...args: Parameters) => { + const path = String(args[0]); + if (path.startsWith(bulky) && path.endsWith(".jsonl")) bulkyOpens++; + return originalOpen(...args); + }, + ); syncBuiltinESMExports(); t.after(() => { t.mock.restoreAll(); syncBuiltinESMExports(); }); - const found = await new PiWebAdapter(runtimeFor(root, bulky, SessionManager.inMemory(root))).searchTranscripts("needle"); + const found = await new PiWebAdapter( + runtimeFor(root, bulky, SessionManager.inMemory(root)), + ).searchTranscripts("needle"); assert.equal(found.partial, true); assert.ok(found.partialReasons.includes("byte-limit")); assert.ok(bulkyOpens < fileCount); diff --git a/web/adapter/session-catalog.ts b/web/adapter/session-catalog.ts index 2f5b0f89a..de086d89d 100644 --- a/web/adapter/session-catalog.ts +++ b/web/adapter/session-catalog.ts @@ -1,5 +1,4 @@ import { open, readdir, stat, type FileHandle } from "node:fs/promises"; -import type { Stats } from "node:fs"; import { basename, join, resolve, sep } from "node:path"; import { StringDecoder } from "node:string_decoder"; import type { SessionEntry } from "@earendil-works/pi-coding-agent"; @@ -89,7 +88,7 @@ function sameIdentity(left: FileIdentity, right: FileIdentity) { left.mtimeNs === right.mtimeNs && left.ctimeNs === right.ctimeNs; } -function identityOf(stats: Stats): FileIdentity { +function identityOf(stats: FileIdentity): FileIdentity { return { dev: stats.dev, ino: stats.ino, @@ -307,19 +306,17 @@ export class SessionCatalog { while (statIndex < paths.length) { signal?.throwIfAborted(); const path = paths[statIndex++]!; - let stats: Stats; try { - stats = await withSignal(stat(path, { bigint: true }), signal); + const stats = await withSignal(stat(path, { bigint: true }), signal); + if (!stats.isFile()) continue; + listed.push({ + path, + identity: identityOf(stats), + mtimeMs: Number(stats.mtimeMs), + }); } catch (error) { if (isAbort(error, signal)) throw error; - continue; } - if (!stats.isFile()) continue; - listed.push({ - path, - identity: identityOf(stats), - mtimeMs: Number(stats.mtimeMs), - }); } }; try { diff --git a/web/adapter/session-loader.ts b/web/adapter/session-loader.ts index 6a84d49f6..af01d10a3 100644 --- a/web/adapter/session-loader.ts +++ b/web/adapter/session-loader.ts @@ -1,5 +1,4 @@ import { readFile, stat } from "node:fs/promises"; -import type { Stats } from "node:fs"; import { resolve } from "node:path"; import { Worker } from "node:worker_threads"; import { SessionManager, type FileEntry } from "@earendil-works/pi-coding-agent"; @@ -29,7 +28,7 @@ let worker: Worker | undefined; let nextId = 0; const pending = new Map void; reject: (error: Error) => void }>(); -function identityOf(stats: Stats): FileIdentity { +function identityOf(stats: FileIdentity): FileIdentity { return { dev: stats.dev, ino: stats.ino, @@ -69,7 +68,8 @@ function fileWorker() { pending.clear(); if (worker === current) worker = undefined; current.unref(); - for (const request of requests) request.reject(error); + const failure = error instanceof Error ? error : new Error(String(error)); + for (const request of requests) request.reject(failure); }); return current; } @@ -119,14 +119,11 @@ export class SessionLoader { async open(path: string, sessionDir: string) { const resolved = resolve(path); - let stats: Stats; - try { - stats = await stat(resolved, { bigint: true }); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT") return SessionManager.open(path, sessionDir); + const stats = await stat(resolved, { bigint: true }).catch((error: unknown) => { + if (error instanceof Error && "code" in error && error.code === "ENOENT") return undefined; throw error; - } - if (!stats.isFile()) return SessionManager.open(path, sessionDir); + }); + if (!stats || !stats.isFile()) return SessionManager.open(path, sessionDir); const identity = identityOf(stats); const cached = this.cache.get(resolved); if (cached && sameIdentity(cached.identity, identity)) { @@ -139,7 +136,7 @@ export class SessionLoader { } const manager = readableSession(resolved, sessionDir, entries); const after = await stat(resolved, { bigint: true }).catch(() => undefined); - if (after?.isFile() && sameIdentity(identity, identityOf(after))) { + if (after && after.isFile() && sameIdentity(identity, identityOf(after))) { this.cache.set(resolved, { identity, manager }); this.touch(resolved); } @@ -159,7 +156,7 @@ export class SessionLoader { private async readStable(path: string, identity: FileIdentity) { const first = await this.readEntries(path); const after = await stat(path, { bigint: true }).catch(() => undefined); - if (!after?.isFile() || sameIdentity(identity, identityOf(after))) return first; + if (!after || !after.isFile() || sameIdentity(identity, identityOf(after))) return first; return this.readEntries(path); } From e6ecc22943dd08231a6442f21352c89694cdcae1 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 16 Oct 2026 15:19:09 +0000 Subject: [PATCH 3/3] fix(web): yield while parsing long session files Parse non-current history with Pi on the parent, in 4000-line turns separated by timers, and reuse the result while the file identity matches. A worker made the first snapshot wait on startup and kept a later prompt from being sent. Co-authored-by: tt-a1i --- tests/web/session-read-cost.test.ts | 96 ++++++++++++++++++++++++++ web/adapter/session-file-worker.ts | 14 ---- web/adapter/session-loader.ts | 103 ++++++++++------------------ 3 files changed, 133 insertions(+), 80 deletions(-) delete mode 100644 web/adapter/session-file-worker.ts diff --git a/tests/web/session-read-cost.test.ts b/tests/web/session-read-cost.test.ts index bfe2ede83..78b937961 100644 --- a/tests/web/session-read-cost.test.ts +++ b/tests/web/session-read-cost.test.ts @@ -295,6 +295,102 @@ test("non-current history keeps Pi's branch without a synchronous open or a seco } }); +test("a long non-current history matches Pi's branch and yields before the request finishes", async (t) => { + const root = await mkdtemp(join(tmpdir(), "openpi-long-history-yield-")); + const directory = join(root, "sessions"); + await mkdir(directory, { recursive: true }); + t.after(() => rm(root, { recursive: true, force: true })); + const messageCount = 10_000; + const path = join(directory, "long-history.jsonl"); + const timestamp = "2026-10-15T00:00:00.000Z"; + const lines = [ + JSON.stringify({ + type: "session", + version: 3, + id: "long-history", + timestamp, + cwd: root, + }), + ]; + for (let index = 0; index < messageCount; index++) { + lines.push( + JSON.stringify({ + type: "message", + id: `e-${index}`, + parentId: index === 0 ? null : `e-${index - 1}`, + timestamp, + message: { + role: index % 2 === 0 ? "user" : "assistant", + content: [{ type: "text", text: "m" }], + timestamp: 1_792_022_400_000, + }, + }), + ); + } + await writeFile(path, `${lines.join("\n")}\n`); + const oracle = SessionManager.open(path, directory); + const oracleBranch = oracle.getBranch(); + const adapter = new PiWebAdapter( + runtimeFor(root, directory, SessionManager.inMemory(root)), + ); + const summary = (await adapter.listSessions()).find( + (session) => session.path === path, + ); + assert.ok(summary); + assert.equal(summary.messageCount, messageCount); + const originalRead = fsPromises.readFile; + const originalSetTimeout = globalThis.setTimeout; + let bodyReads = 0; + let watchParseTurns = false; + let parseTurns = 0; + // Counts timer turns scheduled while the body is being parsed. A later + // await, such as the identity stat, must not count: setImmediate would also + // starve a caller timer between chunks. + globalThis.setTimeout = function ( + this: unknown, + ...args: Parameters + ) { + if (watchParseTurns && args[1] === 0) parseTurns++; + return originalSetTimeout.apply(globalThis, args); + } as typeof setTimeout; + t.mock.method( + fsPromises, + "readFile", + async (...args: Parameters) => { + const result = await originalRead(...args); + if (args[0] === path) { + bodyReads++; + watchParseTurns = true; + } + return result; + }, + ); + syncBuiltinESMExports(); + t.after(() => { + globalThis.setTimeout = originalSetTimeout; + t.mock.restoreAll(); + syncBuiltinESMExports(); + }); + + const session = await adapter.getSession(path, [summary]); + watchParseTurns = false; + assert.ok(parseTurns > 0); + assert.equal(bodyReads, 1); + assert.ok(session); + assert.equal(session.history?.leafEntryId, oracle.getLeafId()); + assert.equal( + session.truncation.entriesOmitted + session.entries.length, + oracleBranch.length, + ); + const earliest = await adapter.getSessionMessageWindow( + summary.id, + path, + "e-0", + ); + assert.equal(earliest.status, "ok"); + assert.equal(bodyReads, 1); +}); + test("concurrent catalog reads share one scan", async (t) => { const root = await mkdtemp(join(tmpdir(), "openpi-catalog-share-")); const directory = join(root, "sessions"); diff --git a/web/adapter/session-file-worker.ts b/web/adapter/session-file-worker.ts deleted file mode 100644 index 1e836dc46..000000000 --- a/web/adapter/session-file-worker.ts +++ /dev/null @@ -1,14 +0,0 @@ -import { parentPort } from "node:worker_threads"; -import { parseSessionEntries } from "@earendil-works/pi-coding-agent"; - -parentPort?.on("message", (message: { id: number; bytes: Uint8Array }) => { - try { - const text = Buffer.from(message.bytes).toString("utf8"); - parentPort?.postMessage({ id: message.id, entries: parseSessionEntries(text) }); - } catch (error) { - parentPort?.postMessage({ - id: message.id, - error: error instanceof Error ? error.message : String(error), - }); - } -}); diff --git a/web/adapter/session-loader.ts b/web/adapter/session-loader.ts index af01d10a3..89b303226 100644 --- a/web/adapter/session-loader.ts +++ b/web/adapter/session-loader.ts @@ -1,9 +1,10 @@ import { readFile, stat } from "node:fs/promises"; import { resolve } from "node:path"; -import { Worker } from "node:worker_threads"; -import { SessionManager, type FileEntry } from "@earendil-works/pi-coding-agent"; +import { parseSessionEntries, SessionManager, type FileEntry } from "@earendil-works/pi-coding-agent"; const MAX_OPEN_SESSIONS = 4; +/** Enough lines that a yield can run, without a timer per message. */ +const PARSE_TURN_LINES = 4_000; type FileIdentity = { dev: bigint; @@ -13,21 +14,11 @@ type FileIdentity = { ctimeNs: bigint; }; -type WorkerResponse = { - id: number; - entries?: FileEntry[]; - error?: string; -}; - type CachedSession = { identity: FileIdentity; manager: SessionManager; }; -let worker: Worker | undefined; -let nextId = 0; -const pending = new Map void; reject: (error: Error) => void }>(); - function identityOf(stats: FileIdentity): FileIdentity { return { dev: stats.dev, @@ -43,59 +34,39 @@ function sameIdentity(left: FileIdentity, right: FileIdentity) { left.mtimeNs === right.mtimeNs && left.ctimeNs === right.ctimeNs; } -function releaseWorker(current: Worker) { - if (pending.size === 0) current.unref(); -} - -function fileWorker() { - if (worker) return worker; - worker = new Worker(new URL("./session-file-worker.ts", import.meta.url), { - execArgv: ["--experimental-strip-types"], - }); - // An idle parser must not keep the process, or a finished request never ends. - worker.unref(); - const current = worker; - current.on("message", (message: WorkerResponse) => { - const request = pending.get(message.id); - if (!request) return; - pending.delete(message.id); - releaseWorker(current); - if (message.error !== undefined) request.reject(new Error(message.error)); - else request.resolve(message.entries ?? []); - }); - current.on("error", (error) => { - const requests = [...pending.values()]; - pending.clear(); - if (worker === current) worker = undefined; - current.unref(); - const failure = error instanceof Error ? error : new Error(String(error)); - for (const request of requests) request.reject(failure); - }); - return current; +function yieldTurn() { + // setTimeout, not setImmediate: a chain of check-phase callbacks starves caller timers. + return new Promise((resolveTurn) => setTimeout(resolveTurn, 0)); } -function parseEntries(bytes: Buffer) { - const id = nextId++; - const current = fileWorker(); - current.ref(); - return new Promise((resolvePromise, reject) => { - const fail = (error: Error) => { - if (pending.get(id)) pending.delete(id); - releaseWorker(current); - reject(error); - }; - pending.set(id, { - resolve: (entries) => { - resolvePromise(entries); - }, - reject: fail, - }); - try { - current.postMessage({ id, bytes }); - } catch (error) { - fail(error instanceof Error ? error : new Error(String(error))); +async function parseInTurns(content: string) { + const entries: FileEntry[] = []; + let start = 0; + let cursor = 0; + let linesInChunk = 0; + let chunks = 0; + const flush = async (end: number) => { + const slice = chunks === 0 && end >= content.length ? content : content.slice(start, end); + // A single short file stays one Pi parse. A longer file yields before every chunk. + if (chunks > 0 || end < content.length) await yieldTurn(); + if (slice.length > 0) entries.push(...parseSessionEntries(slice) as FileEntry[]); + chunks++; + start = end; + linesInChunk = 0; + }; + while (cursor < content.length) { + const newline = content.indexOf("\n", cursor); + if (newline === -1) { + cursor = content.length; + linesInChunk++; + break; } - }); + cursor = newline + 1; + linesInChunk++; + if (linesInChunk === PARSE_TURN_LINES) await flush(cursor); + } + if (start < cursor) await flush(cursor); + return entries; } function readableSession(path: string, sessionDir: string, entries: FileEntry[]) { @@ -110,8 +81,9 @@ function readableSession(path: string, sessionDir: string, entries: FileEntry[]) } /** - * Opens a non-current Session off the request thread and reuses it while the - * file identity is unchanged. Pi remains the parser and the branch authority. + * Opens a non-current Session without a synchronous full-file read. Long files + * yield before each Pi parse turn, including the first. The result is reused + * while the file identity is unchanged. Pi remains the parser and the branch authority. */ export class SessionLoader { private readonly cache = new Map(); @@ -161,7 +133,6 @@ export class SessionLoader { } private async readEntries(path: string) { - // The bytes are read asynchronously here; Pi parses them off this thread. - return parseEntries(await readFile(path)); + return parseInTurns(await readFile(path, "utf8")); } }