diff --git a/extensions/acpx/src/pi-session-catalog.test.ts b/extensions/acpx/src/pi-session-catalog.test.ts index bf77ad5e9e43..b81e6d1f6f8a 100644 --- a/extensions/acpx/src/pi-session-catalog.test.ts +++ b/extensions/acpx/src/pi-session-catalog.test.ts @@ -79,7 +79,16 @@ const createPiStore = ( ); const installFakePi = () => installFakePiFixture(temporaryDirectories, originalPath); +function usePiCandidateCacheClock(): () => void { + let now = Date.now(); + vi.spyOn(Date, "now").mockImplementation(() => now); + return () => { + now += 32_001; + }; +} + afterEach(async () => { + vi.restoreAllMocks(); acpRuntimeMocks.resolveAcpSessionAvailability.mockReset().mockReturnValue({ available: true }); nodeHostMocks.runNodePtyCommand.mockClear(); process.env.PATH = originalPath; @@ -247,6 +256,38 @@ describe("Pi session catalog", () => { expect(firstSummary?.canContinue).toBe(true); }); + it("memoizes file candidates across cadence and re-walks after expiry", async () => { + const sessionDirectory = await createPiStore(); + const baseEnv = { + ...process.env, + PI_CODING_AGENT_SESSION_DIR: sessionDirectory, + PI_CODING_AGENT_DIR: path.dirname(path.dirname(sessionDirectory)), + }; + let now = 1_000; + const nowSpy = vi.spyOn(Date, "now").mockImplementation(() => now); + const readdirSpy = vi.spyOn(fs, "readdir"); + const statSpy = vi.spyOn(fs, "stat"); + try { + await listPiSummaryPage(baseEnv, { offset: 0, limit: 20 }); + const readdirCount = readdirSpy.mock.calls.length; + const statCount = statSpy.mock.calls.length; + + now += 31_999; + await listPiSummaryPage(baseEnv, { offset: 0, limit: 20 }); + expect(readdirSpy).toHaveBeenCalledTimes(readdirCount); + expect(statSpy).toHaveBeenCalledTimes(statCount); + + now += 2; + await listPiSummaryPage(baseEnv, { offset: 0, limit: 20 }); + expect(readdirSpy.mock.calls.length).toBeGreaterThan(readdirCount); + expect(statSpy.mock.calls.length).toBeGreaterThan(statCount); + } finally { + nowSpy.mockRestore(); + readdirSpy.mockRestore(); + statSpy.mockRestore(); + } + }); + it("summarizes and pages a large session within transport limits", async () => { await createPiStore("x".repeat(600 * 1024)); const listed = await listLocalPiSessionPage({ limit: 20 }); @@ -364,12 +405,14 @@ describe("Pi session catalog", () => { }, ]; await fs.writeFile(file, `${entries.map((entry) => JSON.stringify(entry)).join("\n")}\n`); + const expireCandidates = usePiCandidateCacheClock(); expect((await listLocalPiSessionPage({ limit: 20 })).sessions[0]?.name).toBe("Assigned title"); await fs.appendFile( file, `${JSON.stringify({ type: "session_info", id: "clear", parentId: "a", timestamp: "2026-07-13T10:00:04Z", name: "" })}\n`, ); + expireCandidates(); expect((await listLocalPiSessionPage({ limit: 20 })).sessions[0]?.name).toBe("fallback title"); }); @@ -386,6 +429,7 @@ describe("Pi session catalog", () => { cwd: "/workspace", }), ); + const expireCandidates = usePiCandidateCacheClock(); expect((await listLocalPiSessionPage({ limit: 20 })).sessions[0]?.threadId).toBe("pi-session"); await fs.appendFile( @@ -398,6 +442,7 @@ describe("Pi session catalog", () => { name: "No final newline", })}`, ); + expireCandidates(); expect((await listLocalPiSessionPage({ limit: 20 })).sessions[0]?.name).toBe( "No final newline", ); @@ -406,6 +451,7 @@ describe("Pi session catalog", () => { it("rebuilds metadata after a same-file replacement grows", async () => { const directory = await createPiStore("old session"); const file = path.join(directory, "session.jsonl"); + const expireCandidates = usePiCandidateCacheClock(); await listLocalPiSessionPage({ limit: 20 }); const entries = [ @@ -432,6 +478,7 @@ describe("Pi session catalog", () => { }, ]; await fs.writeFile(file, `${entries.map((entry) => JSON.stringify(entry)).join("\n")}\n`); + expireCandidates(); await expect(listLocalPiSessionPage({ limit: 20 })).resolves.toMatchObject({ sessions: [ @@ -547,6 +594,7 @@ describe("Pi session catalog", () => { process.env.PI_CODING_AGENT_DIR = agentDirectory; process.env.HOME = homeDirectory; process.env.USERPROFILE = homeDirectory; + const expireCandidates = usePiCandidateCacheClock(); await expect(listLocalPiSessionPage({ limit: 20 })).resolves.toMatchObject({ sessions: [ @@ -567,6 +615,7 @@ describe("Pi session catalog", () => { name: "", })}\n`, ); + expireCandidates(); expect((await listLocalPiSessionPage({ limit: 20 })).sessions[0]?.name).toBeUndefined(); }); diff --git a/extensions/acpx/src/pi-session-store.ts b/extensions/acpx/src/pi-session-store.ts index 53474f461afc..4f27016d523a 100644 --- a/extensions/acpx/src/pi-session-store.ts +++ b/extensions/acpx/src/pi-session-store.ts @@ -12,6 +12,8 @@ const MAX_SESSION_BYTES = 32 * 1024 * 1024; const MAX_SUMMARY_LINE_BYTES = 1024 * 1024; const APPEND_PROOF_EDGE_BYTES = 64 * 1024; const IO_CONCURRENCY = 8; +const PI_FILE_CANDIDATE_CACHE_TTL_MS = 32_000; +const PI_FILE_CANDIDATE_CACHE_MAX_ENTRIES = 8; const SESSION_ID_PATTERN = /^(?!-)[A-Za-z0-9._:-]{1,256}$/u; type PiSessionSummary = SessionCatalogSession & { file: string; version: number }; @@ -40,10 +42,19 @@ type CachedSummary = PiFileCandidate & { appendProof: { head: Buffer; tail: Buffer }; }; +type PiFileCandidateCacheEntry = { + expiresAt: number; + candidates: Promise; +}; + // Pi owns session-file mutation. The bounded cache resumes append-only metadata // scans, avoiding a full reread every time an active transcript grows. const summaryCache = new Map(); const threadFileCache = new Map(); +// Candidate snapshots are valid for one session-root/ACP-root identity for 32s, just beyond stable +// polling. Root changes or expiry re-walk; rejected scans are removed so transient I/O recovers. +// Keeping this bounded avoids rescanning every file each poll without retaining obsolete stores. +const piFileCandidateCache = new Map(); function threadCacheKey(storeRoot: string, threadId: string): string { return `${storeRoot}\0${threadId}`; @@ -151,7 +162,7 @@ async function mapConcurrent( return results; } -async function piFileCandidates(env: NodeJS.ProcessEnv): Promise { +async function scanPiFileCandidates(env: NodeJS.ProcessEnv): Promise { const { root, files } = await discoverPiSessionFiles(env); const configuredAcpRoot = piAcpSessionStoreRoot(env); const acpRoot = configuredAcpRoot ? await realpathOrResolve(configuredAcpRoot) : undefined; @@ -177,6 +188,38 @@ async function piFileCandidates(env: NodeJS.ProcessEnv): Promise right.mtimeMs - left.mtimeMs); } +async function piFileCandidates(env: NodeJS.ProcessEnv): Promise { + const store = piSessionStore(env); + const key = `${store.root}\0${store.flat}\0${piAcpSessionStoreRoot(env) ?? ""}`; + const cached = piFileCandidateCache.get(key); + if (cached && cached.expiresAt > Date.now()) { + piFileCandidateCache.delete(key); + piFileCandidateCache.set(key, cached); + return await cached.candidates; + } + if (cached) { + piFileCandidateCache.delete(key); + } + const candidates = scanPiFileCandidates(env); + const entry = { expiresAt: Date.now() + PI_FILE_CANDIDATE_CACHE_TTL_MS, candidates }; + piFileCandidateCache.set(key, entry); + while (piFileCandidateCache.size > PI_FILE_CANDIDATE_CACHE_MAX_ENTRIES) { + const oldest = piFileCandidateCache.keys().next(); + if (oldest.done) { + break; + } + piFileCandidateCache.delete(oldest.value); + } + try { + return await candidates; + } catch (error) { + if (piFileCandidateCache.get(key) === entry) { + piFileCandidateCache.delete(key); + } + throw error; + } +} + function pathIsWithin(root: string, candidate: string): boolean { const relative = path.relative(root, candidate); return ( diff --git a/extensions/anthropic/session-catalog.test.ts b/extensions/anthropic/session-catalog.test.ts index 3884678d0246..a21f93dfe05b 100644 --- a/extensions/anthropic/session-catalog.test.ts +++ b/extensions/anthropic/session-catalog.test.ts @@ -1579,7 +1579,7 @@ describe("Claude session catalog", () => { } }); - it("reuses the assembled scan within the TTL without per-file filesystem work", async () => { + it("serves an unchanged assembled scan without reparsing transcript files", async () => { const home = await createHome(); const sessionIds = ["cached-session-a", "cached-session-b"]; await writeProject({ @@ -1609,42 +1609,260 @@ describe("Claude session catalog", () => { value.endsWith("sessions-index.json") || path.basename(value).startsWith("local_")); expect(realpathSpy.mock.calls.filter(([filePath]) => isCatalogFile(filePath))).toEqual([]); - expect(statSpy.mock.calls.filter(([filePath]) => isCatalogFile(filePath))).toEqual([]); + expect( + statSpy.mock.calls.some( + ([filePath]) => typeof filePath === "string" && filePath.endsWith(".jsonl"), + ), + ).toBe(true); expect(openSpy).not.toHaveBeenCalled(); expect(readFileSpy.mock.calls.filter(([filePath]) => isCatalogFile(filePath))).toEqual([]); }); - it("bounds append-only snapshot staleness to 15 seconds", async () => { + it("does not shorten cache validity for Desktop rows with no captured transcript", async () => { + const home = await createHome(); + await writeProject({ + home, + entries: [], + transcripts: { existing: [sdkCliMessage("existing", "Existing")] }, + }); + await writeDesktopMetadata(home, "missing", { + cliSessionId: "missing-desktop-transcript", + title: "Missing", + }); + let now = 1_000; + vi.spyOn(Date, "now").mockImplementation(() => now); + const openSpy = vi.spyOn(fs, "open"); + const first = await listLocalClaudeSessionPage({}, home); + openSpy.mockClear(); + + now += 15_001; + await expect(listLocalClaudeSessionPage({}, home)).resolves.toEqual(first); + expect(openSpy).not.toHaveBeenCalled(); + }); + + it("retries an unchanged tree after project-root canonicalization recovers", async () => { + const home = await createHome(); + const projectRoot = path.join(home, ".claude", "projects"); + await writeProject({ + home, + entries: [], + transcripts: { recovered: [sdkCliMessage("recovered", "Recovered")] }, + }); + const realpath = fs.realpath.bind(fs); + let failRoot = true; + vi.spyOn(fs, "realpath").mockImplementation(async (...args) => { + if (failRoot && args[0] === projectRoot) { + failRoot = false; + throw new Error("transient realpath failure"); + } + return await realpath(...args); + }); + + await expect(listLocalClaudeSessionPage({}, home)).resolves.toEqual({ sessions: [] }); + await expect(listLocalClaudeSessionPage({}, home)).resolves.toMatchObject({ + sessions: [expect.objectContaining({ threadId: "recovered" })], + }); + }); + + it("retries transient index safe-file failures during discovery", async () => { + const home = await createHome(); + const sessionId = "safe-file-retry"; + const transcriptPath = path.join( + home, + ".claude", + "projects", + "-workspace", + `${sessionId}.jsonl`, + ); + await writeProject({ + home, + entries: [{ sessionId, fullPath: transcriptPath }], + transcripts: { [sessionId]: [sdkCliMessage(sessionId, "Recovered")] }, + }); + const realpath = fs.realpath.bind(fs); + let transcriptAttempts = 0; + vi.spyOn(fs, "realpath").mockImplementation(async (...args) => { + if (args[0] === transcriptPath && transcriptAttempts++ === 0) { + throw new Error("transient transcript realpath failure"); + } + return await realpath(...args); + }); + + await expect(listLocalClaudeSessionPage({}, home)).resolves.toMatchObject({ + sessions: [expect.objectContaining({ threadId: sessionId })], + }); + expect(transcriptAttempts).toBe(2); + }); + + it("expires a partial discovery scan on the short transient-I/O retry bound", async () => { + const home = await createHome(); + const sessionId = "partial-scan-retry"; + const transcriptPath = path.join( + home, + ".claude", + "projects", + "-workspace", + `${sessionId}.jsonl`, + ); + await writeProject({ + home, + entries: [], + transcripts: { [sessionId]: [sdkCliMessage(sessionId, "Recovered")] }, + }); + const open = fs.open.bind(fs); + let transcriptAttempts = 0; + let now = 1_000; + vi.spyOn(Date, "now").mockImplementation(() => now); + vi.spyOn(fs, "open").mockImplementation(async (...args) => { + if (args[0] === transcriptPath && transcriptAttempts++ === 0) { + throw new Error("transient transcript open failure"); + } + return await open(...args); + }); + + await expect(listLocalClaudeSessionPage({}, home)).resolves.toEqual({ sessions: [] }); + now += 15_001; + await expect(listLocalClaudeSessionPage({}, home)).resolves.toMatchObject({ + sessions: [expect.objectContaining({ threadId: sessionId })], + }); + expect(transcriptAttempts).toBe(2); + }); + + it("does not shorten cache validity for a permanently missing indexed transcript", async () => { + const home = await createHome(); + const missingPath = path.join( + home, + ".claude", + "projects", + "-workspace", + "missing-indexed.jsonl", + ); + await writeProject({ + home, + entries: [{ sessionId: "missing-indexed", fullPath: missingPath }], + transcripts: {}, + }); + let now = 1_000; + vi.spyOn(Date, "now").mockImplementation(() => now); + const realpathSpy = vi.spyOn(fs, "realpath"); + + await expect(listLocalClaudeSessionPage({}, home)).resolves.toEqual({ sessions: [] }); + expect(realpathSpy.mock.calls.filter(([filePath]) => filePath === missingPath)).toHaveLength(1); + now += 15_001; + await expect(listLocalClaudeSessionPage({}, home)).resolves.toEqual({ sessions: [] }); + expect(realpathSpy.mock.calls.filter(([filePath]) => filePath === missingPath)).toHaveLength(1); + }); + + it("invalidates the assembled scan when an existing transcript is appended", async () => { const home = await createHome(); const projectDir = path.join(home, ".claude", "projects", "-workspace"); const sessionId = "append-staleness"; const transcriptPath = path.join(projectDir, `${sessionId}.jsonl`); + const futureTranscriptPath = path.join(projectDir, "future-sibling.jsonl"); await writeProject({ home, entries: [], - transcripts: { [sessionId]: [sdkCliMessage(sessionId, "Initial")] }, + transcripts: { + [sessionId]: [sdkCliMessage(sessionId, "Initial")], + "future-sibling": [sdkCliMessage("future-sibling", "Future")], + }, }); const baseNow = Date.now(); const fixedDirectoryTime = new Date(baseNow - 10_000); + await fs.utimes(futureTranscriptPath, new Date(baseNow + 10_000), new Date(baseNow + 10_000)); await fs.utimes(projectDir, fixedDirectoryTime, fixedDirectoryTime); - const nowSpy = vi.spyOn(Date, "now").mockReturnValue(baseNow); const initial = await listLocalClaudeSessionPage({}, home); - const initialUpdatedAt = initial.sessions[0]?.updatedAt; + const initialUpdatedAt = initial.sessions.find( + (session) => session.threadId === sessionId, + )?.updatedAt; await fs.appendFile(transcriptPath, `${JSON.stringify({ type: "progress" })}\n`); const appendedAt = new Date(baseNow + 2_000); await fs.utimes(transcriptPath, appendedAt, appendedAt); - // Content writes do not portably change the parent directory mtime. Pin it so this test owns - // only the documented TTL path on every filesystem used by CI. + // Content writes do not portably change the parent directory mtime. Pin it so only the child + // mtime component of the tree stamp can invalidate this snapshot on every CI filesystem. await fs.utimes(projectDir, fixedDirectoryTime, fixedDirectoryTime); - nowSpy.mockReturnValue(baseNow + 14_999); - const staleWithinBound = await listLocalClaudeSessionPage({}, home); - expect(staleWithinBound.sessions[0]?.updatedAt).toBe(initialUpdatedAt); + const refreshed = await listLocalClaudeSessionPage({}, home); + expect(initialUpdatedAt).not.toBe(appendedAt.getTime()); + expect(refreshed.sessions.find((session) => session.threadId === sessionId)?.updatedAt).toBe( + appendedAt.getTime(), + ); + }); - nowSpy.mockReturnValue(baseNow + 15_000); - const refreshedAtBound = await listLocalClaudeSessionPage({}, home); - expect(refreshedAtBound.sessions[0]?.updatedAt).toBe(appendedAt.getTime()); + it("invalidates the assembled scan after same-size same-mtime atomic replacement", async () => { + const home = await createHome(); + const projectDir = path.join(home, ".claude", "projects", "-workspace"); + const sessionId = "atomic-replacement"; + const transcriptPath = path.join(projectDir, `${sessionId}.jsonl`); + const fixedTime = new Date("2026-07-20T12:00:00.000Z"); + await writeProject({ + home, + entries: [], + transcripts: { [sessionId]: [sdkCliMessage(sessionId, "Alpha")] }, + }); + await fs.utimes(transcriptPath, fixedTime, fixedTime); + await fs.utimes(projectDir, fixedTime, fixedTime); + expect((await listLocalClaudeSessionPage({}, home)).sessions[0]?.name).toBe("Alpha"); + + const replacementPath = path.join(projectDir, "replacement.tmp"); + await fs.writeFile(replacementPath, `${JSON.stringify(sdkCliMessage(sessionId, "Bravo"))}\n`); + await fs.utimes(replacementPath, fixedTime, fixedTime); + await fs.rename(replacementPath, transcriptPath); + await fs.utimes(projectDir, fixedTime, fixedTime); + + expect((await listLocalClaudeSessionPage({}, home)).sessions[0]?.name).toBe("Bravo"); + }); + + it("keeps the metadata byte frontier in serial directory order under parallel stats", async () => { + const home = await createHome(); + const projectDir = path.join(home, ".claude", "projects", "-workspace"); + await fs.mkdir(projectDir, { recursive: true }); + await fs.writeFile(path.join(projectDir, "sessions-index.json"), '{"version":1,"entries":[]}'); + const fileBytes = 1024 * 1024; + const chunkBytes = 16 * 1024; + const leadingFiller = Buffer.from(`${"x".repeat(chunkBytes - 1)}\n`.repeat(63)); + for (let index = 0; index < 66; index += 1) { + const sessionId = `budget-${String(index).padStart(2, "0")}`; + const messageLine = Buffer.from(`${JSON.stringify(sdkCliMessage(sessionId, sessionId))}\n`); + const finalFillerBytes = fileBytes - leadingFiller.length - messageLine.length; + const finalFiller = Buffer.from(`${"x".repeat(finalFillerBytes - 1)}\n`); + await fs.writeFile( + path.join(projectDir, `${sessionId}.jsonl`), + Buffer.concat([leadingFiller, finalFiller, messageLine]), + ); + } + const serialNames = (await fs.readdir(projectDir)) + .filter((name) => name.endsWith(".jsonl")) + .map((name) => name.slice(0, -".jsonl".length)); + const expected = serialNames.slice(0, 64).toSorted(); + const realpath = fs.realpath.bind(fs); + let activeRealpaths = 0; + let maxConcurrentRealpaths = 0; + vi.spyOn(fs, "realpath").mockImplementation(async (...args) => { + const target = args[0]; + if (typeof target !== "string" || !target.endsWith(".jsonl")) { + return await realpath(...args); + } + activeRealpaths += 1; + maxConcurrentRealpaths = Math.max(maxConcurrentRealpaths, activeRealpaths); + await new Promise((resolve) => { + setImmediate(resolve); + }); + try { + return await realpath(...args); + } finally { + activeRealpaths -= 1; + } + }); + + const cold = await listLocalClaudeSessionPage({ limit: 100 }, home); + expect(maxConcurrentRealpaths).toBeGreaterThan(1); + expect(cold.sessions.map((session) => session.threadId).toSorted()).toEqual(expected); + + await fs.utimes(projectDir, new Date(), new Date(Date.now() + 2_000)); + const warm = await listLocalClaudeSessionPage({ limit: 100 }, home); + expect(warm.sessions.map((session) => session.threadId).toSorted()).toEqual(expected); }); it("invalidates the assembled scan on a project directory mtime change", async () => { @@ -1854,10 +2072,11 @@ describe("Claude session catalog", () => { } return await readFile(...args); }); + let now = 1_000; + vi.spyOn(Date, "now").mockImplementation(() => now); expect((await listLocalClaudeSessionPage({}, home)).sessions).toEqual([]); - const refreshTime = new Date(Date.now() + 2_000); - await fs.utimes(projectDir, refreshTime, refreshTime); + now += 15_001; await expect(listLocalClaudeSessionPage({}, home)).resolves.toMatchObject({ sessions: [{ threadId: sessionId, name: "Recovered index" }], diff --git a/extensions/anthropic/session-catalog.ts b/extensions/anthropic/session-catalog.ts index ed0ac6622341..ceddf60995c7 100644 --- a/extensions/anthropic/session-catalog.ts +++ b/extensions/anthropic/session-catalog.ts @@ -65,7 +65,10 @@ const MAX_CATALOG_DISCOVERY_FILES = 10_000; const MAX_CATALOG_DISCOVERY_CACHE_ENTRIES = 20_000; const MAX_CATALOG_JSON_CACHE_ENTRIES = 4_000; const MAX_CLAUDE_SESSION_SCAN_CACHE_ENTRIES = 8; -const CLAUDE_SESSION_SCAN_TTL_MS = 15_000; +const CLAUDE_SESSION_SCAN_HARD_TTL_MS = 5 * 60_000; +const CLAUDE_PARTIAL_SCAN_TTL_MS = 15_000; +const CLAUDE_DESKTOP_SCAN_TTL_MS = 60_000; +const CLAUDE_CATALOG_IO_CONCURRENCY = 32; const CLAUDE_METADATA_PREFIX_BYTES = 1024 * 1024; const CLAUDE_METADATA_READ_CHUNK_BYTES = 16 * 1024; const MAX_CATALOG_METADATA_SCAN_BYTES = 64 * 1024 * 1024; @@ -149,14 +152,61 @@ type CatalogJsonCacheEntry = { type ClaudeSessionScanCacheEntry = { treeStamp: string; - expiresAt: number; + hardExpiresAt: number; + desktopStoreAvailable: boolean; + desktopExpiresAt: number; records: Promise; }; +type SafeSessionFile = { filePath: string; stat: Stats } | undefined; + +type ClaudeProjectDirectorySnapshot = { + directory: string; + childNames: string[]; +}; + +type ClaudeChildFileSignature = readonly [name: string, mtimeMs: number, size: number, ino: number]; + +type ClaudeProjectsTreeSnapshot = { + root: string; + resolvedRoot?: string; + projectDirectories: ClaudeProjectDirectorySnapshot[]; + treeStamp: string; +}; + +type ClaudeSessionScanContext = ClaudeProjectsTreeSnapshot & { + complete: boolean; + safeFiles: Map>; +}; + +// Transcript discoveries stay valid only for the same root/id/inode/mtime/size and are LRU-bounded; +// a false hit would corrupt pagination, so warm scans re-charge the original deterministic byte cost. const catalogDiscoveryCache = new Map(); +// Parsed index/Desktop JSON stays valid for one path+mtime+size and is LRU-bounded; read failures are +// never cached, so transient metadata I/O cannot hide a later successful read. const catalogJsonCache = new Map(); +// Whole scans are root-scoped and bounded; tree/Desktop/hard expiries below own invalidation, avoiding +// an unbounded home map while preserving the exact resolved records promise for concurrent callers. const claudeSessionScanCache = new Map(); +async function mapConcurrent( + values: T[], + limit: number, + mapper: (value: T) => Promise, +): Promise { + const results: R[] = []; + results.length = values.length; + let nextIndex = 0; + const workers = Array.from({ length: Math.min(limit, values.length) }, async () => { + while (nextIndex < values.length) { + const index = nextIndex++; + results[index] = await mapper(values[index]!); + } + }); + await Promise.all(workers); + return results; +} + function setBoundedCache(cache: Map, key: K, value: V, maxEntries: number): void { cache.delete(key); cache.set(key, value); @@ -295,7 +345,7 @@ async function safeSessionFile( resolvedRoot: string, candidate: string, sessionId: string, -): Promise<{ filePath: string; stat: Stats } | undefined> { +): Promise { if (!isWithin(root, candidate) || path.basename(candidate) !== `${sessionId}.jsonl`) { return undefined; } @@ -306,13 +356,49 @@ async function safeSessionFile( } const stat = await fs.stat(resolvedCandidate); return stat.isFile() ? { filePath: resolvedCandidate, stat } : undefined; - } catch { - return undefined; + } catch (error) { + const code = error && typeof error === "object" && "code" in error ? error.code : undefined; + if (code === "ENOENT" || code === "ENOTDIR") { + return undefined; + } + throw new Error("Claude session file validation failed", { cause: error }); } } -async function readJsonFile(filePath: string): Promise { - const stat = await fs.stat(filePath).catch(() => undefined); +function safeSessionFileForScan( + context: ClaudeSessionScanContext, + candidate: string, + sessionId: string, +): Promise { + if (!context.resolvedRoot) { + return Promise.resolve(undefined); + } + const key = `${sessionId}\0${path.resolve(candidate)}`; + let pending = context.safeFiles.get(key); + if (!pending) { + // Canonical path + stat are valid only for this assembled scan. Sharing the promise prevents + // index fallback and discovery from serially resolving the same file twice. + const request = safeSessionFile(context.root, context.resolvedRoot, candidate, sessionId); + pending = request.catch(() => { + context.complete = false; + if (context.safeFiles.get(key) === pending) { + context.safeFiles.delete(key); + } + return undefined; + }); + context.safeFiles.set(key, pending); + } + return pending; +} + +async function readJsonFile( + filePath: string, + options: { onIoFailure?: () => void } = {}, +): Promise { + const stat = await fs.stat(filePath).catch(() => { + options.onIoFailure?.(); + return undefined; + }); if (!stat?.isFile()) { catalogJsonCache.delete(filePath); return undefined; @@ -322,8 +408,15 @@ async function readJsonFile(filePath: string): Promise { setBoundedCache(catalogJsonCache, filePath, cached, MAX_CATALOG_JSON_CACHE_ENTRIES); return cached.value; } + let content: string; try { - const value = JSON.parse(await fs.readFile(filePath, "utf8")) as unknown; + content = await fs.readFile(filePath, "utf8"); + } catch { + options.onIoFailure?.(); + return undefined; + } + try { + const value = JSON.parse(content) as unknown; setBoundedCache( catalogJsonCache, filePath, @@ -350,32 +443,80 @@ function projectsDir(homeDir: string): string { return path.join(homeDir, ".claude", "projects"); } -async function readProjectsTreeStamp(root: string): Promise { +async function readProjectsTreeSnapshot(root: string): Promise { let entries: Dirent[]; try { entries = await fs.readdir(root, { withFileTypes: true }); } catch { - return "unavailable"; + return { root, projectDirectories: [], treeStamp: "unavailable" }; } - const directoryNames = entries - .filter((entry) => entry.isDirectory()) - .map((entry) => entry.name) - .toSorted(); - const directorySnapshots = await Promise.all( - directoryNames.map(async (name) => { - const directory = path.join(root, name); - const [stat, childNames] = await Promise.all([ + const directoryEntries = entries.filter((entry) => entry.isDirectory()); + const [resolvedRoot, directories] = await Promise.all([ + fs.realpath(root).catch(() => undefined), + mapConcurrent(directoryEntries, CLAUDE_CATALOG_IO_CONCURRENCY, async (entry) => { + const directory = path.join(root, entry.name); + const [stat, children] = await Promise.all([ fs.stat(directory).catch(() => undefined), - fs.readdir(directory).catch(() => undefined), + fs.readdir(directory, { withFileTypes: true }).catch(() => undefined), ]); - return [ - name, - stat?.isDirectory() === true ? stat.mtimeMs : null, - childNames?.toSorted() ?? null, - ] as const; + return { entry, directory, stat, children }; }), + ]); + const childTargets = directories.flatMap(({ directory, children }, directoryIndex) => + (children ?? []).map((child) => ({ directoryIndex, directory, child })), ); - return JSON.stringify(directorySnapshots); + const childSignatures = await mapConcurrent( + childTargets, + CLAUDE_CATALOG_IO_CONCURRENCY, + async ({ directoryIndex, directory, child }) => { + const childStat = await fs.stat(path.join(directory, child.name)).catch(() => undefined); + const signature = childStat?.isFile() + ? ([child.name, childStat.mtimeMs, childStat.size, childStat.ino] as const) + : undefined; + return { directoryIndex, signature }; + }, + ); + const signaturesByDirectory = Array.from( + { length: directories.length }, + (): ClaudeChildFileSignature[] => [], + ); + for (const { directoryIndex, signature } of childSignatures) { + if (signature) { + signaturesByDirectory[directoryIndex]?.push(signature); + } + } + const directorySnapshots = directories.map(({ entry, directory, stat, children }, index) => { + const fileSignatures = signaturesByDirectory[index] ?? []; + const maxChildMtime = fileSignatures.reduce( + (maximum, [, mtime]) => Math.max(maximum ?? mtime, mtime), + null, + ); + return { + directory, + childNames: children?.map((child) => child.name) ?? [], + stamp: [ + entry.name, + stat?.isDirectory() === true ? stat.mtimeMs : null, + children?.map((child) => child.name) ?? null, + maxChildMtime ?? null, + fileSignatures, + ] as const, + }; + }); + return { + root, + ...(resolvedRoot ? { resolvedRoot } : {}), + projectDirectories: directorySnapshots.map(({ directory, childNames }) => ({ + directory, + childNames, + })), + treeStamp: JSON.stringify([resolvedRoot ?? null, directorySnapshots.map(({ stamp }) => stamp)]), + }; +} + +async function desktopSessionStoreAvailable(homeDir: string): Promise { + const stat = await fs.stat(desktopSessionsDir(homeDir)).catch(() => undefined); + return stat?.isDirectory() === true; } function desktopSessionsDir(homeDir: string): string { @@ -430,19 +571,35 @@ async function readDesktopMetadata(homeDir: string): Promise<{ return { active, archived }; } -async function readIndexRecords(homeDir: string): Promise<{ +async function readIndexRecords(context: ClaudeSessionScanContext): Promise<{ records: Map; sidechainIds: Set; }> { - const root = projectsDir(homeDir); const records = new Map(); const sidechainIds = new Set(); - const resolvedRoot = await fs.realpath(root).catch(() => undefined); - if (!resolvedRoot) { + if (!context.resolvedRoot) { return { records, sidechainIds }; } - for (const projectDir of await childDirectories(root)) { - const raw = await readJsonFile(path.join(projectDir, "sessions-index.json")); + const indexes = await mapConcurrent( + context.projectDirectories, + CLAUDE_CATALOG_IO_CONCURRENCY, + async ({ directory, childNames }) => ({ + directory, + raw: childNames.includes("sessions-index.json") + ? await readJsonFile(path.join(directory, "sessions-index.json"), { + onIoFailure: () => { + context.complete = false; + }, + }) + : undefined, + }), + ); + const candidates: Array<{ + directory: string; + entry: SessionIndexEntry; + sessionId: string; + }> = []; + for (const { directory, raw } of indexes) { if (!isRecord(raw) || !Array.isArray(raw.entries)) { continue; } @@ -455,54 +612,69 @@ async function readIndexRecords(homeDir: string): Promise<{ if (!sessionId) { continue; } + candidates.push({ directory, entry, sessionId }); + } + } + const safeFiles = await mapConcurrent( + candidates, + CLAUDE_CATALOG_IO_CONCURRENCY, + async ({ directory, entry, sessionId }) => { if (entry.isSidechain === true) { - sidechainIds.add(sessionId); - records.delete(sessionId); - continue; + return undefined; } const indexedPath = optionalString(entry.fullPath, MAX_STRING_LENGTH); - const safeFile = await safeSessionFile( - root, - resolvedRoot, - indexedPath ?? path.join(projectDir, `${sessionId}.jsonl`), + return await safeSessionFileForScan( + context, + indexedPath ?? path.join(directory, `${sessionId}.jsonl`), sessionId, ); - if (!safeFile) { - continue; - } - const createdAt = timestampMs(entry.created); - const updatedAt = timestampMs(entry.modified) ?? timestampMs(entry.fileMtime); - const summary = optionalString(entry.summary, 500); - const firstPrompt = optionalString(entry.firstPrompt, 500); - records.set(sessionId, { - threadId: sessionId, - name: summary ?? firstPrompt ?? null, - cwd: optionalString(entry.projectPath), - status: "stored", - ...(createdAt !== undefined ? { createdAt } : {}), - ...(updatedAt !== undefined ? { updatedAt, recencyAt: updatedAt } : {}), - source: "claude-cli", - modelProvider: "anthropic", - ...(optionalString(entry.gitBranch, 500) - ? { gitBranch: optionalString(entry.gitBranch, 500) } - : {}), - archived: false, - filePath: safeFile.filePath, - }); + }, + ); + for (const [index, candidate] of candidates.entries()) { + const { entry, sessionId } = candidate; + if (entry.isSidechain === true) { + sidechainIds.add(sessionId); + records.delete(sessionId); + continue; } + const safeFile = safeFiles[index]; + if (!safeFile) { + continue; + } + const createdAt = timestampMs(entry.created); + const updatedAt = timestampMs(entry.modified) ?? timestampMs(entry.fileMtime); + const summary = optionalString(entry.summary, 500); + const firstPrompt = optionalString(entry.firstPrompt, 500); + records.set(sessionId, { + threadId: sessionId, + name: summary ?? firstPrompt ?? null, + cwd: optionalString(entry.projectPath), + status: "stored", + ...(createdAt !== undefined ? { createdAt } : {}), + ...(updatedAt !== undefined ? { updatedAt, recencyAt: updatedAt } : {}), + source: "claude-cli", + modelProvider: "anthropic", + ...(optionalString(entry.gitBranch, 500) + ? { gitBranch: optionalString(entry.gitBranch, 500) } + : {}), + archived: false, + filePath: safeFile.filePath, + }); } return { records, sidechainIds }; } -async function locateSessionFile(homeDir: string, sessionId: string): Promise { - const root = projectsDir(homeDir); - const resolvedRoot = await fs.realpath(root).catch(() => undefined); - if (!resolvedRoot) { - return undefined; - } - for (const projectDir of await childDirectories(root)) { - const candidate = path.join(projectDir, `${sessionId}.jsonl`); - const safeFile = await safeSessionFile(root, resolvedRoot, candidate, sessionId); +async function locateSessionFile( + context: ClaudeSessionScanContext, + sessionId: string, +): Promise { + const fileName = `${sessionId}.jsonl`; + for (const { directory, childNames } of context.projectDirectories) { + if (!childNames.includes(fileName)) { + continue; + } + const candidate = path.join(directory, fileName); + const safeFile = await safeSessionFileForScan(context, candidate, sessionId); if (safeFile) { return safeFile.filePath; } @@ -511,13 +683,12 @@ async function locateSessionFile(homeDir: string, sessionId: string): Promise, sidechainIds: Set, ): Promise { - const root = projectsDir(homeDir); - const resolvedRoot = await fs.realpath(root).catch(() => undefined); - if (!resolvedRoot) { + const { root } = context; + if (!context.resolvedRoot) { // The root (or a parent) is gone. Entries are tagged with the logical root, so evict by that // rather than a lexical containment test the canonical cache keys would never satisfy. for (const [cachedPath, entry] of catalogDiscoveryCache) { @@ -531,198 +702,207 @@ async function discoverCliRecords( let scannedBytes = 0; let truncated = false; const seenFilePaths = new Set(); - scan: for (const projectDir of await childDirectories(root)) { - let names: string[]; - try { - names = await fs.readdir(projectDir); - } catch { - continue; - } - for (const name of names) { + const candidates: Array<{ directory: string; name: string; sessionId: string }> = []; + collect: for (const { directory, childNames } of context.projectDirectories) { + for (const name of childNames) { if (!name.endsWith(".jsonl")) { continue; } if (discoveredFiles >= MAX_CATALOG_DISCOVERY_FILES) { truncated = true; - break scan; + break collect; } discoveredFiles += 1; const sessionId = name.slice(0, -".jsonl".length); - if (!sessionId || records.has(sessionId) || sidechainIds.has(sessionId)) { - continue; + if (sessionId) { + candidates.push({ directory, name, sessionId }); } - const safeFile = await safeSessionFile( - root, - resolvedRoot, - path.join(projectDir, name), - sessionId, - ); - if (!safeFile) { - continue; + } + } + const safeFiles = await mapConcurrent( + candidates, + CLAUDE_CATALOG_IO_CONCURRENCY, + async ({ directory, name, sessionId }) => + records.has(sessionId) || sidechainIds.has(sessionId) + ? undefined + : await safeSessionFileForScan(context, path.join(directory, name), sessionId), + ); + for (const [candidateIndex, candidate] of candidates.entries()) { + const { sessionId } = candidate; + // Resolve metadata concurrently, but make every semantic decision in the original directory + // order. In particular, duplicates and the global byte frontier must match a serial cold scan. + if (records.has(sessionId) || sidechainIds.has(sessionId)) { + continue; + } + const safeFile = safeFiles[candidateIndex]; + if (!safeFile) { + continue; + } + const { filePath, stat: fileStat } = safeFile; + seenFilePaths.add(filePath); + const cached = catalogDiscoveryCache.get(filePath); + // Claude transcripts only append while active, then stay static, so mtime+size+ino identify + // the parsed content (ino also rejects an atomic replacement that reused the same mtime/size), + // and sessionId ensures the record is served only under the filename-derived id it was built + // for. These files are owner-owned and append-only; a mid-scan read-permission revocation is + // not a state the Claude CLI produces, so a hit intentionally skips the open() re-check. + if ( + cached && + cached.root === root && + cached.mtimeMs === fileStat.mtimeMs && + cached.size === fileStat.size && + cached.ino === fileStat.ino && + cached.sessionId === sessionId && + // Only replay the cached record if a cold scan would also reach its metadata under the + // current remaining byte budget. Once earlier files grow, replaying a record whose original + // scan cost now crosses the frontier would surface a record a cold scan stops before; fall + // through to a bounded rescan instead so warm and cold discovery (and pagination) match. + scannedBytes + cached.scannedBytes <= MAX_CATALOG_METADATA_SCAN_BYTES + ) { + if (cached.sidechain) { + sidechainIds.add(sessionId); } - const { filePath, stat: fileStat } = safeFile; - seenFilePaths.add(filePath); - const cached = catalogDiscoveryCache.get(filePath); - // Claude transcripts only append while active, then stay static, so mtime+size+ino identify - // the parsed content (ino also rejects an atomic replacement that reused the same mtime/size), - // and sessionId ensures the record is served only under the filename-derived id it was built - // for. These files are owner-owned and append-only; a mid-scan read-permission revocation is - // not a state the Claude CLI produces, so a hit intentionally skips the open() re-check. - if ( - cached && - cached.root === root && - cached.mtimeMs === fileStat.mtimeMs && - cached.size === fileStat.size && - cached.ino === fileStat.ino && - cached.sessionId === sessionId && - // Only replay the cached record if a cold scan would also reach its metadata under the - // current remaining byte budget. Once earlier files grow, replaying a record whose original - // scan cost now crosses the frontier would surface a record a cold scan stops before; fall - // through to a bounded rescan instead so warm and cold discovery (and pagination) match. - scannedBytes + cached.scannedBytes <= MAX_CATALOG_METADATA_SCAN_BYTES - ) { - if (cached.sidechain) { - sidechainIds.add(sessionId); - } - if (cached.record) { - records.set(sessionId, cached.record); - } - // Cache hits read no transcript bytes, but they still charge the file's original scan cost - // so the byte-budget cutoff matches a cold scan; otherwise repeated calls would free budget - // and progressively discover more files. - scannedBytes += cached.scannedBytes; - if (scannedBytes >= MAX_CATALOG_METADATA_SCAN_BYTES) { - truncated = true; - break scan; - } - continue; - } - const handle = await fs.open(filePath, "r").catch(() => undefined); - if (!handle) { - continue; - } - let cacheable = false; - let fileScannedBytes = 0; - try { - const stat = await handle.stat(); - let aiTitle: string | undefined; - let pending = Buffer.alloc(0); - let fileOffset = 0; - let stopFile = false; - const inspectLine = (line: Buffer): boolean => { - let raw: unknown; - try { - raw = JSON.parse(line.toString("utf8")) as unknown; - } catch { - return false; - } - if (!isRecord(raw) || raw.sessionId !== sessionId) { - return false; - } - if (raw.type === "ai-title") { - aiTitle = optionalString(raw.aiTitle, 500) ?? aiTitle; - return false; - } - if (typeof raw.entrypoint === "string" && !isCliEntrypoint(raw.entrypoint)) { - return true; - } - if (isCliEntrypoint(raw.entrypoint) && raw.isSidechain === true) { - sidechainIds.add(sessionId); - return true; - } - if ( - !isCliEntrypoint(raw.entrypoint) || - raw.type !== "user" || - !isRecord(raw.message) || - raw.message.role !== "user" - ) { - return false; - } - const fragments: string[] = []; - collectTranscriptText(raw.message.content, fragments); - const firstPrompt = optionalString(fragments[0], 500); - const createdAt = timestampMs(raw.timestamp); - records.set(sessionId, { - threadId: sessionId, - name: aiTitle ?? firstPrompt ?? null, - cwd: optionalString(raw.cwd), - status: "stored", - ...(createdAt !== undefined ? { createdAt } : {}), - updatedAt: stat.mtimeMs, - recencyAt: stat.mtimeMs, - source: "claude-cli", - modelProvider: "anthropic", - ...(optionalString(raw.version, 256) - ? { cliVersion: optionalString(raw.version, 256) } - : {}), - ...(optionalString(raw.gitBranch, 500) - ? { gitBranch: optionalString(raw.gitBranch, 500) } - : {}), - archived: false, - filePath, - }); - return true; - }; - while ( - !stopFile && - fileOffset < stat.size && - fileOffset < CLAUDE_METADATA_PREFIX_BYTES && - scannedBytes < MAX_CATALOG_METADATA_SCAN_BYTES - ) { - const size = Math.min( - CLAUDE_METADATA_READ_CHUNK_BYTES, - stat.size - fileOffset, - CLAUDE_METADATA_PREFIX_BYTES - fileOffset, - MAX_CATALOG_METADATA_SCAN_BYTES - scannedBytes, - ); - const chunk = Buffer.allocUnsafe(size); - const { bytesRead } = await handle.read(chunk, 0, size, fileOffset); - if (bytesRead === 0) { - break; - } - fileOffset += bytesRead; - scannedBytes += bytesRead; - pending = pending.length - ? Buffer.concat([pending, chunk.subarray(0, bytesRead)]) - : chunk.subarray(0, bytesRead); - let newline: number; - while (!stopFile && (newline = pending.indexOf(0x0a)) >= 0) { - stopFile = inspectLine(pending.subarray(0, newline)); - pending = pending.subarray(newline + 1); - } - } - if (!stopFile && fileOffset >= stat.size && pending.length > 0) { - inspectLine(pending); - } - // A read whose chunk was capped by the remaining global budget stops on a smaller boundary - // than a cold scan would, so its fileOffset undercounts the true unconstrained scan cost. - // Don't cache such an entry: replaying its low cost later (with more budget free) would let - // the warm scan cross the frontier and surface sessions a cold scan omits. - const budgetConstrained = scannedBytes >= MAX_CATALOG_METADATA_SCAN_BYTES; - cacheable = - !budgetConstrained && - (stopFile || fileOffset >= stat.size || fileOffset >= CLAUDE_METADATA_PREFIX_BYTES); - fileScannedBytes = fileOffset; - } finally { - await handle.close(); - } - // Negative and sidechain-only results are cached too; unchanged files should not be reparsed. - if (cacheable) { - cacheCatalogDiscovery(filePath, { - root, - mtimeMs: fileStat.mtimeMs, - size: fileStat.size, - ino: fileStat.ino, - sessionId, - scannedBytes: fileScannedBytes, - record: records.get(sessionId) ?? null, - sidechain: sidechainIds.has(sessionId), - }); + if (cached.record) { + records.set(sessionId, cached.record); } + // Cache hits read no transcript bytes, but they still charge the file's original scan cost + // so the byte-budget cutoff matches a cold scan; otherwise repeated calls would free budget + // and progressively discover more files. + scannedBytes += cached.scannedBytes; if (scannedBytes >= MAX_CATALOG_METADATA_SCAN_BYTES) { truncated = true; - break scan; + break; } + continue; + } + const handle = await fs.open(filePath, "r").catch(() => { + context.complete = false; + return undefined; + }); + if (!handle) { + continue; + } + let cacheable = false; + let fileScannedBytes = 0; + try { + const stat = await handle.stat(); + let aiTitle: string | undefined; + let pending = Buffer.alloc(0); + let fileOffset = 0; + let stopFile = false; + const inspectLine = (line: Buffer): boolean => { + let raw: unknown; + try { + raw = JSON.parse(line.toString("utf8")) as unknown; + } catch { + return false; + } + if (!isRecord(raw) || raw.sessionId !== sessionId) { + return false; + } + if (raw.type === "ai-title") { + aiTitle = optionalString(raw.aiTitle, 500) ?? aiTitle; + return false; + } + if (typeof raw.entrypoint === "string" && !isCliEntrypoint(raw.entrypoint)) { + return true; + } + if (isCliEntrypoint(raw.entrypoint) && raw.isSidechain === true) { + sidechainIds.add(sessionId); + return true; + } + if ( + !isCliEntrypoint(raw.entrypoint) || + raw.type !== "user" || + !isRecord(raw.message) || + raw.message.role !== "user" + ) { + return false; + } + const fragments: string[] = []; + collectTranscriptText(raw.message.content, fragments); + const firstPrompt = optionalString(fragments[0], 500); + const createdAt = timestampMs(raw.timestamp); + records.set(sessionId, { + threadId: sessionId, + name: aiTitle ?? firstPrompt ?? null, + cwd: optionalString(raw.cwd), + status: "stored", + ...(createdAt !== undefined ? { createdAt } : {}), + updatedAt: stat.mtimeMs, + recencyAt: stat.mtimeMs, + source: "claude-cli", + modelProvider: "anthropic", + ...(optionalString(raw.version, 256) + ? { cliVersion: optionalString(raw.version, 256) } + : {}), + ...(optionalString(raw.gitBranch, 500) + ? { gitBranch: optionalString(raw.gitBranch, 500) } + : {}), + archived: false, + filePath, + }); + return true; + }; + while ( + !stopFile && + fileOffset < stat.size && + fileOffset < CLAUDE_METADATA_PREFIX_BYTES && + scannedBytes < MAX_CATALOG_METADATA_SCAN_BYTES + ) { + const size = Math.min( + CLAUDE_METADATA_READ_CHUNK_BYTES, + stat.size - fileOffset, + CLAUDE_METADATA_PREFIX_BYTES - fileOffset, + MAX_CATALOG_METADATA_SCAN_BYTES - scannedBytes, + ); + const chunk = Buffer.allocUnsafe(size); + const { bytesRead } = await handle.read(chunk, 0, size, fileOffset); + if (bytesRead === 0) { + break; + } + fileOffset += bytesRead; + scannedBytes += bytesRead; + pending = pending.length + ? Buffer.concat([pending, chunk.subarray(0, bytesRead)]) + : chunk.subarray(0, bytesRead); + let newline: number; + while (!stopFile && (newline = pending.indexOf(0x0a)) >= 0) { + stopFile = inspectLine(pending.subarray(0, newline)); + pending = pending.subarray(newline + 1); + } + } + if (!stopFile && fileOffset >= stat.size && pending.length > 0) { + inspectLine(pending); + } + // A read whose chunk was capped by the remaining global budget stops on a smaller boundary + // than a cold scan would, so its fileOffset undercounts the true unconstrained scan cost. + // Don't cache such an entry: replaying its low cost later (with more budget free) would let + // the warm scan cross the frontier and surface sessions a cold scan omits. + const budgetConstrained = scannedBytes >= MAX_CATALOG_METADATA_SCAN_BYTES; + cacheable = + !budgetConstrained && + (stopFile || fileOffset >= stat.size || fileOffset >= CLAUDE_METADATA_PREFIX_BYTES); + fileScannedBytes = fileOffset; + } finally { + await handle.close(); + } + // Negative and sidechain-only results are cached too; unchanged files should not be reparsed. + if (cacheable) { + cacheCatalogDiscovery(filePath, { + root, + mtimeMs: fileStat.mtimeMs, + size: fileStat.size, + ino: fileStat.ino, + sessionId, + scannedBytes: fileScannedBytes, + record: records.get(sessionId) ?? null, + sidechain: sidechainIds.has(sessionId), + }); + } + if (scannedBytes >= MAX_CATALOG_METADATA_SCAN_BYTES) { + truncated = true; + break; } } if (!truncated) { @@ -735,13 +915,17 @@ async function discoverCliRecords( } } -async function scanClaudeSessions(homeDir: string): Promise { +async function scanClaudeSessions( + homeDir: string, + snapshot: ClaudeProjectsTreeSnapshot, +): Promise<{ records: CatalogRecord[]; complete: boolean }> { + const context: ClaudeSessionScanContext = { ...snapshot, complete: true, safeFiles: new Map() }; const [indexed, desktop] = await Promise.all([ - readIndexRecords(homeDir), + readIndexRecords(context), readDesktopMetadata(homeDir), ]); const records = indexed.records; - await discoverCliRecords(homeDir, records, indexed.sidechainIds); + await discoverCliRecords(context, records, indexed.sidechainIds); for (const sessionId of desktop.archived) { records.delete(sessionId); } @@ -750,7 +934,7 @@ async function scanClaudeSessions(homeDir: string): Promise { continue; } const existing = records.get(sessionId); - const filePath = existing?.filePath ?? (await locateSessionFile(homeDir, sessionId)); + const filePath = existing?.filePath ?? (await locateSessionFile(context, sessionId)); if (!filePath) { continue; } @@ -775,11 +959,14 @@ async function scanClaudeSessions(homeDir: string): Promise { filePath, }); } - return [...records.values()].toSorted((left, right) => { - const recency = - (right.recencyAt ?? right.updatedAt ?? 0) - (left.recencyAt ?? left.updatedAt ?? 0); - return recency || left.threadId.localeCompare(right.threadId); - }); + return { + records: [...records.values()].toSorted((left, right) => { + const recency = + (right.recencyAt ?? right.updatedAt ?? 0) - (left.recencyAt ?? left.updatedAt ?? 0); + return recency || left.threadId.localeCompare(right.threadId); + }), + complete: context.complete, + }; } async function listClaudeSessions( @@ -787,30 +974,48 @@ async function listClaudeSessions( options: { forceRefresh?: boolean } = {}, ): Promise { const root = projectsDir(homeDir); - const treeStamp = await readProjectsTreeStamp(root); + const [treeSnapshot, desktopStoreAvailable] = await Promise.all([ + readProjectsTreeSnapshot(root), + desktopSessionStoreAvailable(homeDir), + ]); + const now = Date.now(); const cached = claudeSessionScanCache.get(root); - // This tree stamp is the catalog cache's cheap freshness owner: exact child membership makes - // same-tick adds, deletes, and renames visible even when a filesystem reuses the directory mtime. - // File appends and Desktop-only changes may keep this snapshot stale for at most 15 seconds, - // below the UI's 30-second cadence. Clients observe them on the first poll after cache expiry. + // Child membership + file mtime/size signatures invalidate CLI rows on the next poll; five minutes + // backstops metadata anomalies. Desktop has a 60s bound when its macOS store exists; Linux skips it. + // Specific-thread force refresh bypasses both, or a stale page could hide a just-created session. if ( options.forceRefresh !== true && cached && - cached.treeStamp === treeStamp && - cached.expiresAt > Date.now() + cached.treeStamp === treeSnapshot.treeStamp && + cached.hardExpiresAt > now && + cached.desktopStoreAvailable === desktopStoreAvailable && + (!desktopStoreAvailable || cached.desktopExpiresAt > now) ) { setBoundedCache(claudeSessionScanCache, root, cached, MAX_CLAUDE_SESSION_SCAN_CACHE_ENTRIES); return await cached.records; } - const records = scanClaudeSessions(homeDir); + const scan = scanClaudeSessions(homeDir, treeSnapshot); + let scanComplete = true; + const records = scan.then((result) => { + scanComplete = result.complete; + return result.records; + }); const entry = { - treeStamp, - expiresAt: Date.now() + CLAUDE_SESSION_SCAN_TTL_MS, + treeStamp: treeSnapshot.treeStamp, + hardExpiresAt: now + CLAUDE_SESSION_SCAN_HARD_TTL_MS, + desktopStoreAvailable, + desktopExpiresAt: now + CLAUDE_DESKTOP_SCAN_TTL_MS, records, }; setBoundedCache(claudeSessionScanCache, root, entry, MAX_CLAUDE_SESSION_SCAN_CACHE_ENTRIES); try { - return await records; + const result = await records; + if (!scanComplete && claudeSessionScanCache.get(root) === entry) { + // Partial results still serve this caller, but retry within 15s so transient per-file I/O + // cannot hide recovered sessions behind the five-minute unchanged-tree backstop. + entry.hardExpiresAt = Date.now() + CLAUDE_PARTIAL_SCAN_TTL_MS; + } + return result; } catch (error) { if (claudeSessionScanCache.get(root) === entry) { claudeSessionScanCache.delete(root); @@ -957,8 +1162,8 @@ export async function readLocalClaudeTranscriptPage( (record) => record.threadId === params.threadId, )?.filePath; if (!filePath) { - // A just-created session can predate the short catalog TTL. Specific reads must retry against - // disk so opening a new thread never fails only because the sidebar snapshot is still warm. + // A just-created session can race the stamp snapshot. Specific reads must retry against disk so + // opening a new thread never fails only because the assembled catalog is still warm. filePath = (await listClaudeSessions(homeDir, { forceRefresh: true })).find( (record) => record.threadId === params.threadId, )?.filePath; diff --git a/extensions/codex/src/session-catalog-terminal.ts b/extensions/codex/src/session-catalog-terminal.ts index fe4b8ae21c23..8ed280537c71 100644 --- a/extensions/codex/src/session-catalog-terminal.ts +++ b/extensions/codex/src/session-catalog-terminal.ts @@ -58,8 +58,8 @@ export async function requireCatalogEligibleThread( control: CodexSessionCatalogControl, threadId: string, ): Promise { - // Mutating actions use a fresh pinned control and authoritative thread/read. For passive callers, - // keep the three-second positive hit; only a miss needs to bypass the list memo. + // Mutating actions use a fresh pinned control and authoritative thread/read. Passive positive hits + // may use the cadence-safe page memo; only a miss must bypass it before rejecting a new thread. const cached = await findCatalogEligibleThread(control, threadId, false); if (cached) { return cached; diff --git a/extensions/codex/src/session-catalog.test.ts b/extensions/codex/src/session-catalog.test.ts index 5c384803d089..32354a14cdc6 100644 --- a/extensions/codex/src/session-catalog.test.ts +++ b/extensions/codex/src/session-catalog.test.ts @@ -555,7 +555,7 @@ describe("Codex supervision catalog", () => { expect(cloneSpy).toHaveBeenCalledTimes(4); }); - it("briefly memoizes thread lists and invalidates on TTL or config identity", async () => { + it("memoizes thread lists across the stable poll cadence and invalidates by config", async () => { let now = 1_000; let runtimeConfig = {} as OpenClawConfig; commandRpcMocks.codexControlRequest.mockResolvedValue({ @@ -571,7 +571,11 @@ describe("Codex supervision catalog", () => { await control.listPage({ limit: 25 }); expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledOnce(); - now += 3_001; + now += 31_999; + await control.listPage({ limit: 25 }); + expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledOnce(); + + now += 2; await control.listPage({ limit: 25 }); expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledTimes(2); @@ -580,6 +584,103 @@ describe("Codex supervision catalog", () => { expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledTimes(3); }); + it("serves the last real page after a refresh failure and retries the next poll", async () => { + let now = 1_000; + commandRpcMocks.codexControlRequest.mockResolvedValue({ + data: [idleThread({ id: "thread-stale", source: "cli" })], + }); + const control = createCodexSessionCatalogControl({ + getPluginConfig: () => ({ supervision: { enabled: true } }), + getRuntimeConfig: () => config, + now: () => now, + }); + const first = await control.listPage({ limit: 25 }); + + now += 32_001; + commandRpcMocks.codexControlRequest.mockRejectedValueOnce(new Error("app-server unavailable")); + await expect(control.listPage({ limit: 25 })).resolves.toEqual(first); + expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledTimes(2); + + commandRpcMocks.codexControlRequest.mockResolvedValue({ + data: [idleThread({ id: "thread-recovered", source: "cli" })], + }); + await expect(control.listPage({ limit: 25 })).resolves.toMatchObject({ + sessions: [expect.objectContaining({ threadId: "thread-recovered" })], + }); + expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledTimes(3); + }); + + it("evicts a cold failed page so the next caller retries immediately", async () => { + commandRpcMocks.codexControlRequest.mockRejectedValueOnce(new Error("cold failure")); + const control = createCodexSessionCatalogControl({ + getPluginConfig: () => ({ supervision: { enabled: true } }), + getRuntimeConfig: () => config, + }); + + await expect(control.listPage({ limit: 25 })).rejects.toThrow("cold failure"); + commandRpcMocks.codexControlRequest.mockResolvedValue({ + data: [idleThread({ id: "thread-recovered", source: "cli" })], + }); + await expect(control.listPage({ limit: 25 })).resolves.toMatchObject({ + sessions: [expect.objectContaining({ threadId: "thread-recovered" })], + }); + expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledTimes(2); + }); + + it("propagates a forced refresh failure while preserving the stale retry state", async () => { + let now = 1_000; + commandRpcMocks.codexControlRequest.mockResolvedValue({ + data: [idleThread({ id: "thread-stale", source: "cli" })], + }); + const control = createCodexSessionCatalogControl({ + getPluginConfig: () => ({ supervision: { enabled: true } }), + getRuntimeConfig: () => config, + now: () => now, + }); + await control.listPage({ limit: 25 }); + + commandRpcMocks.codexControlRequest.mockRejectedValueOnce(new Error("forced refresh failed")); + await expect(control.listPage({ limit: 25, forceRefresh: true })).rejects.toThrow( + "forced refresh failed", + ); + + commandRpcMocks.codexControlRequest.mockResolvedValue({ + data: [idleThread({ id: "thread-recovered", source: "cli" })], + }); + now += 1; + await expect(control.listPage({ limit: 25 })).resolves.toMatchObject({ + sessions: [expect.objectContaining({ threadId: "thread-recovered" })], + }); + expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledTimes(3); + }); + + it("serves stale data to a passive waiter overlapping a failed forced refresh", async () => { + commandRpcMocks.codexControlRequest.mockResolvedValue({ + data: [idleThread({ id: "thread-stale", source: "cli" })], + }); + const control = createCodexSessionCatalogControl({ + getPluginConfig: () => ({ supervision: { enabled: true } }), + getRuntimeConfig: () => config, + }); + const stale = await control.listPage({ limit: 25 }); + let rejectRefresh!: (error: Error) => void; + commandRpcMocks.codexControlRequest.mockImplementationOnce( + async () => + await new Promise((_resolve, reject) => { + rejectRefresh = reject; + }), + ); + + const forced = control.listPage({ limit: 25, forceRefresh: true }); + await vi.waitFor(() => expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledTimes(2)); + const passive = control.listPage({ limit: 25 }); + const forcedResult = expect(forced).rejects.toThrow("forced refresh failed"); + rejectRefresh(new Error("forced refresh failed")); + + await forcedResult; + await expect(passive).resolves.toEqual(stale); + }); + it("force-refreshes after a cached specific-thread miss", async () => { let includeThread = false; commandRpcMocks.codexControlRequest.mockImplementation(async () => ({ diff --git a/extensions/codex/src/session-catalog.ts b/extensions/codex/src/session-catalog.ts index d53c5913fdb8..51d17b7e8162 100644 --- a/extensions/codex/src/session-catalog.ts +++ b/extensions/codex/src/session-catalog.ts @@ -124,7 +124,7 @@ const boundCatalogSessionId = (value: unknown) => boundedCatalogString(value, MAX_SESSION_ID_LENGTH); const CODEX_SUPERVISION_SESSION_KEY_PREFIX = "harness:codex:supervision:"; -const CODEX_SESSION_CATALOG_LIST_TTL_MS = 3_000; +const CODEX_SESSION_CATALOG_LIST_TTL_MS = 32_000; const CODEX_SESSION_CATALOG_LIST_CACHE_MAX_ENTRIES = 32; type CodexCatalogRequestOptions = { @@ -135,6 +135,7 @@ type CodexCatalogRequestOptions = { type CodexCatalogPageCacheEntry = { expiresAt: number; page: Promise; + stalePage?: Promise; }; function codexCatalogPageCacheKey(params: CodexSessionCatalogPageParams): string { @@ -429,17 +430,31 @@ export function createCodexSessionCatalogControl(params: { const key = codexCatalogPageCacheKey(pageParams); const cached = cache.get(key); if (pageParams.forceRefresh !== true && cached && cached.expiresAt > now()) { - // The app-server may scan rollout metadata for thread/list. Share a page for three seconds; - // config identity and forceRefresh invalidate it so specific actions cannot miss new rows. + // thread/list traverses Codex's thread store and rollout metadata. A config-scoped 32s page + // survives the 30s poll cadence; TTL and forceRefresh bound staleness for new native rows. + // Without this memo every sidebar poll repeats the app-server roundtrip and metadata scan. cache.delete(key); cache.set(key, cached); - return await cached.page; + try { + return await cached.page; + } catch (error) { + if (cached.stalePage) { + return await cached.stalePage; + } + throw error; + } } if (cached) { cache.delete(key); } + const serveStaleOnError = pageParams.forceRefresh !== true; const page = control.listPage(pageParams); - const entry = { expiresAt: now() + CODEX_SESSION_CATALOG_LIST_TTL_MS, page }; + const stalePage = cached?.stalePage ?? cached?.page; + const entry: CodexCatalogPageCacheEntry = { + expiresAt: now() + CODEX_SESSION_CATALOG_LIST_TTL_MS, + page, + ...(stalePage ? { stalePage } : {}), + }; cache.set(key, entry); while (cache.size > CODEX_SESSION_CATALOG_LIST_CACHE_MAX_ENTRIES) { const oldest = cache.keys().next(); @@ -449,8 +464,27 @@ export function createCodexSessionCatalogControl(params: { cache.delete(oldest.value); } try { - return await page; + const result = await page; + delete entry.stalePage; + return result; } catch (error) { + if (stalePage) { + let stale: CodexSessionCatalogPage | undefined; + try { + stale = await stalePage; + } catch { + // The prior page was not real data, so propagate the current app-server failure. + } + if (stale) { + if (cache.get(key) === entry) { + cache.delete(key); + cache.set(key, { expiresAt: now(), page: Promise.resolve(stale) }); + } + if (serveStaleOnError) { + return stale; + } + } + } if (cache.get(key) === entry) { cache.delete(key); } diff --git a/extensions/opencode/session-catalog-plugin.ts b/extensions/opencode/session-catalog-plugin.ts index 8b9f70846b51..146702d17fce 100644 --- a/extensions/opencode/session-catalog-plugin.ts +++ b/extensions/opencode/session-catalog-plugin.ts @@ -171,7 +171,9 @@ function isOpenCodeSessionCatalogEnabled(pluginConfig: unknown): boolean { ); } -function createOpenCodeSessionNodeHostCommands(): OpenClawPluginNodeHostCommand[] { +function createOpenCodeSessionNodeHostCommands( + api: OpenClawPluginApi, +): OpenClawPluginNodeHostCommand[] { const available = ({ config, env }: { config: unknown; env: NodeJS.ProcessEnv }) => fullConfigCatalogEnabled(config) && executableOnPath("opencode", env); return [ @@ -181,7 +183,11 @@ function createOpenCodeSessionNodeHostCommands(): OpenClawPluginNodeHostCommand[ dangerous: false, isAvailable: available, handle: async (paramsJSON) => - JSON.stringify(await listLocalOpenCodeSessionPage(parseNodeParams(paramsJSON))), + JSON.stringify( + await listLocalOpenCodeSessionPage(parseNodeParams(paramsJSON), { + configIdentity: currentOpenCodeCatalogConfig(api), + }), + ), }, { command: OPENCODE_SESSION_READ_COMMAND, @@ -334,8 +340,9 @@ async function listOpenCodeHosts( query: Parameters[0], ): Promise { const runtime = api.runtime; + const config = currentOpenCodeCatalogConfig(api); const canContinue = resolveAcpSessionAvailability({ - config: currentOpenCodeCatalogConfig(api), + config, backendId: ACPX_BACKEND_ID, agentId: OPENCODE_ACP_AGENT_ID, }).available; @@ -355,11 +362,14 @@ async function listOpenCodeHosts( label: "Local OpenCode", kind: "gateway", connected: true, - ...(await listLocalOpenCodeSessionPage({ - limit: query.limitPerHost, - ...(query.search ? { searchTerm: query.search } : {}), - cursor: query.cursors?.[LOCAL_HOST_ID], - }).then((page) => setCatalogCapabilities(page, { canContinue, canOpenTerminal: true }))), + ...(await listLocalOpenCodeSessionPage( + { + limit: query.limitPerHost, + ...(query.search ? { searchTerm: query.search } : {}), + cursor: query.cursors?.[LOCAL_HOST_ID], + }, + { configIdentity: config }, + ).then((page) => setCatalogCapabilities(page, { canContinue, canOpenTerminal: true }))), }); } catch { hosts.push({ @@ -484,15 +494,18 @@ async function continueOpenCodeSession( sourceKey, findExisting: () => listAdoptedOpenCodeSessions(api).get(sourceKey), create: async () => { - const page = await listLocalOpenCodeSessionPage({ - searchTerm: threadId, - limit: MAX_PAGE_LIMIT, - }).catch(() => undefined); + const config = currentOpenCodeCatalogConfig(api); + const page = await listLocalOpenCodeSessionPage( + { + searchTerm: threadId, + limit: MAX_PAGE_LIMIT, + }, + { configIdentity: config, forceRefresh: true }, + ).catch(() => undefined); const record = page?.sessions.find((session) => session.threadId === threadId); if (!record) { throw new OpenCodeCatalogParamsError("OpenCode session is unavailable"); } - const config = currentOpenCodeCatalogConfig(api); const currentAvailability = resolveAcpSessionAvailability({ config, backendId: ACPX_BACKEND_ID, @@ -564,7 +577,7 @@ export function registerOpenCodeSessionCatalog(api: OpenClawPluginApi): void { unwrapNodePayload, }), }); - for (const command of createOpenCodeSessionNodeHostCommands()) { + for (const command of createOpenCodeSessionNodeHostCommands(api)) { api.registerNodeHostCommand(command); } for (const policy of createOpenCodeSessionNodeInvokePolicies()) { diff --git a/extensions/opencode/session-catalog.test.ts b/extensions/opencode/session-catalog.test.ts index 6f8aa73a2a09..031c66826e97 100644 --- a/extensions/opencode/session-catalog.test.ts +++ b/extensions/opencode/session-catalog.test.ts @@ -424,6 +424,37 @@ describe("OpenCode session catalog", () => { }, ); + it.runIf(process.platform !== "win32")( + "memoizes the CLI database query across cadence and invalidates by config identity", + async () => { + await installFakeOpenCode(); + let now = 1_000; + const nowSpy = vi.spyOn(Date, "now").mockImplementation(() => now); + const configIdentity = {}; + try { + await listLocalOpenCodeSessionPage({ limit: 20 }, { configIdentity }); + await listLocalOpenCodeSessionPage({ limit: 20 }, { configIdentity }); + expect(childProcessMocks.spawn).toHaveBeenCalledOnce(); + + now += 31_999; + await listLocalOpenCodeSessionPage({ limit: 20 }, { configIdentity }); + expect(childProcessMocks.spawn).toHaveBeenCalledOnce(); + + await listLocalOpenCodeSessionPage({ limit: 20 }, { configIdentity, forceRefresh: true }); + expect(childProcessMocks.spawn).toHaveBeenCalledTimes(2); + + await listLocalOpenCodeSessionPage({ limit: 20 }, { configIdentity: {} }); + expect(childProcessMocks.spawn).toHaveBeenCalledTimes(3); + + now += 32_001; + await listLocalOpenCodeSessionPage({ limit: 20 }, { configIdentity }); + expect(childProcessMocks.spawn).toHaveBeenCalledTimes(4); + } finally { + nowSpy.mockRestore(); + } + }, + ); + it.runIf(process.platform !== "win32")( "hides and rejects Continue when ACP cannot resume OpenCode", async () => { diff --git a/extensions/opencode/session-catalog.ts b/extensions/opencode/session-catalog.ts index 7b3ec1fd30fa..d39ab7566a81 100644 --- a/extensions/opencode/session-catalog.ts +++ b/extensions/opencode/session-catalog.ts @@ -22,6 +22,8 @@ const MAX_CLI_OUTPUT_BYTES = 32 * 1024 * 1024; const MAX_TRANSCRIPT_ITEM_BYTES = 512 * 1024; const MAX_TRANSCRIPT_PAGE_BYTES = 20 * 1024 * 1024; const CLI_TIMEOUT_MS = 30_000; +const OPENCODE_QUERY_CACHE_TTL_MS = 32_000; +const OPENCODE_QUERY_CACHE_MAX_ENTRIES = 32; const SESSION_ID_PATTERN = /^(?!-)[A-Za-z0-9._:-]{1,256}$/u; const SAFE_ENV_KEYS = [ "APPDATA", @@ -46,6 +48,34 @@ const SAFE_ENV_KEYS = [ "XDG_STATE_HOME", ] as const; +type OpenCodeQueryCacheEntry = { + expiresAt: number; + result: Promise; + resolved?: true; +}; + +type OpenCodeQueryCacheOptions = { + configIdentity?: object; + forceRefresh?: boolean; +}; + +const openCodeConfigIdentities = new WeakMap(); +// Query results are valid for one immutable OpenClaw config identity, CLI environment, and SQL text. +// Config/env changes or 32s expiry invalidate them; failures are removed so recovery retries at once. +// The bounded map prevents pagination variants from growing while avoiding a subprocess every poll. +const openCodeQueryCache = new Map(); +let nextOpenCodeConfigIdentity = 1; + +function openCodeQueryCacheKey(query: string, configIdentity: object): string { + let identity = openCodeConfigIdentities.get(configIdentity); + if (identity === undefined) { + identity = nextOpenCodeConfigIdentity++; + openCodeConfigIdentities.set(configIdentity, identity); + } + const environment = SAFE_ENV_KEYS.map((key) => `${key}=${process.env[key] ?? ""}`).join("\0"); + return `${String(identity)}\0${environment}\0${query}`; +} + export type OpenCodeSessionPage = { sessions: SessionCatalogSession[]; nextCursor?: string; @@ -307,6 +337,49 @@ export async function queryOpenCodeDatabase(query: string): Promise { return output.trim() ? (JSON.parse(output) as unknown) : []; } +async function queryCachedOpenCodeSessions( + query: string, + options: OpenCodeQueryCacheOptions, +): Promise { + const key = openCodeQueryCacheKey(query, options.configIdentity ?? process.env); + const cached = openCodeQueryCache.get(key); + if (options.forceRefresh !== true && cached && cached.expiresAt > Date.now()) { + openCodeQueryCache.delete(key); + openCodeQueryCache.set(key, cached); + return await cached.result; + } + if (cached) { + openCodeQueryCache.delete(key); + } + const result = queryOpenCodeDatabase(query); + const entry: OpenCodeQueryCacheEntry = { + expiresAt: Date.now() + OPENCODE_QUERY_CACHE_TTL_MS, + result, + }; + openCodeQueryCache.set(key, entry); + while (openCodeQueryCache.size > OPENCODE_QUERY_CACHE_MAX_ENTRIES) { + const oldest = openCodeQueryCache.keys().next(); + if (oldest.done) { + break; + } + openCodeQueryCache.delete(oldest.value); + } + try { + const value = await result; + entry.resolved = true; + return value; + } catch (error) { + if (openCodeQueryCache.get(key) === entry) { + if (cached?.resolved) { + openCodeQueryCache.set(key, cached); + } else { + openCodeQueryCache.delete(key); + } + } + throw error; + } +} + export async function exportOpenCodeSession(threadId: string): Promise { const output = await runOpenCode(["--pure", "export", threadId]); return JSON.parse(output) as unknown; @@ -341,7 +414,10 @@ function parseOpenCodeSession(value: unknown): SessionCatalogSession | undefined }; } -export async function listLocalOpenCodeSessionPage(value?: unknown): Promise { +export async function listLocalOpenCodeSessionPage( + value?: unknown, + options: OpenCodeQueryCacheOptions = {}, +): Promise { const params = parseListParams(value); const offset = decodeCursor(params.cursor); const requestedCount = params.searchTerm @@ -353,7 +429,7 @@ export async function listLocalOpenCodeSessionPage(value?: unknown): Promise MAX_CLI_LIST_SESSIONS) { throw new Error("OpenCode returned an invalid session list"); } diff --git a/src/gateway/server-methods/session-catalog.test.ts b/src/gateway/server-methods/session-catalog.test.ts index 742746234d46..8fabdec33cde 100644 --- a/src/gateway/server-methods/session-catalog.test.ts +++ b/src/gateway/server-methods/session-catalog.test.ts @@ -232,6 +232,27 @@ describe("session catalog Gateway methods", () => { } }); + it("shares settled identical lists across out-of-phase clients until the window expires", async () => { + let now = 1_000; + const nowSpy = vi.spyOn(Date, "now").mockImplementation(() => now); + const list = vi.fn(async () => []); + hoisted.activeRegistry.sessionCatalogs = [{ provider: provider("codex", { list }) }]; + const config = {}; + try { + await call("sessions.catalog.list", {}, config); + + now += 2_500; + await call("sessions.catalog.list", {}, config); + expect(list).toHaveBeenCalledOnce(); + + now += 501; + await call("sessions.catalog.list", {}, config); + expect(list).toHaveBeenCalledTimes(2); + } finally { + nowSpy.mockRestore(); + } + }); + it("projects authoritative creator ownership onto streamed and final catalog rows", async () => { const broadcastToConnIds = vi.fn(); const host = { @@ -326,7 +347,7 @@ describe("session catalog Gateway methods", () => { }); }); - it("does not clone the shared list projection for each catalog request", async () => { + it("does not clone the shared list projection for a settled shared catalog result", async () => { const storedEntries = [ { sessionKey: "agent:main:shared", @@ -347,12 +368,13 @@ describe("session catalog Gateway methods", () => { }, ]; const cloneSpy = vi.spyOn(globalThis, "structuredClone"); + const config = {}; try { - await call("sessions.catalog.list", {}); - await call("sessions.catalog.list", {}); + await call("sessions.catalog.list", {}, config); + await call("sessions.catalog.list", {}, config); expect(cloneSpy).not.toHaveBeenCalled(); - expect(hoisted.listSessionEntriesReadOnly).toHaveBeenCalledTimes(2); + expect(hoisted.listSessionEntriesReadOnly).toHaveBeenCalledOnce(); expect(hoisted.listSessionEntriesReadOnly).toHaveBeenLastCalledWith({ agentId: "main", clone: false, @@ -689,6 +711,8 @@ describe("session catalog Gateway methods", () => { }); it("retries an exception-derived create target failure without a config reload", async () => { + let now = 1_000; + const nowSpy = vi.spyOn(Date, "now").mockImplementation(() => now); const resolveCreateSession = vi .fn() .mockImplementationOnce(() => { @@ -703,25 +727,30 @@ describe("session catalog Gateway methods", () => { ]; const config = {}; - const unavailable = await call("sessions.catalog.list", {}, config); - expect(unavailable).toHaveBeenCalledWith(true, { - catalogs: [ - expect.objectContaining({ - capabilities: { continueSession: false, archive: false }, - }), - ], - }); - const recovered = await call("sessions.catalog.list", {}, config); - expect(recovered).toHaveBeenCalledWith(true, { - catalogs: [ - expect.objectContaining({ - capabilities: expect.objectContaining({ - createSession: { model: "anthropic/claude-opus-4-8" }, + try { + const unavailable = await call("sessions.catalog.list", {}, config); + expect(unavailable).toHaveBeenCalledWith(true, { + catalogs: [ + expect.objectContaining({ + capabilities: { continueSession: false, archive: false }, }), - }), - ], - }); - expect(resolveCreateSession).toHaveBeenCalledTimes(2); + ], + }); + now += 3_001; + const recovered = await call("sessions.catalog.list", {}, config); + expect(recovered).toHaveBeenCalledWith(true, { + catalogs: [ + expect.objectContaining({ + capabilities: expect.objectContaining({ + createSession: { model: "anthropic/claude-opus-4-8" }, + }), + }), + ], + }); + expect(resolveCreateSession).toHaveBeenCalledTimes(2); + } finally { + nowSpy.mockRestore(); + } }); it("keeps creation available when catalog history listing fails", async () => { diff --git a/src/gateway/server-methods/session-catalog.ts b/src/gateway/server-methods/session-catalog.ts index bc14734d7b08..34c2c269600a 100644 --- a/src/gateway/server-methods/session-catalog.ts +++ b/src/gateway/server-methods/session-catalog.ts @@ -14,6 +14,7 @@ import { validateSessionsCatalogReadParams, } from "../../../packages/gateway-protocol/src/index.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import type { PluginRegistry } from "../../plugins/registry-types.js"; import { getActivePluginSessionExtensionRegistry } from "../../plugins/runtime.js"; import { gatewaySubagentState } from "../../plugins/runtime/gateway-bindings.js"; import type { @@ -31,6 +32,8 @@ import type { GatewayRequestHandlers, RespondFn } from "./types.js"; import { assertValidParams } from "./validation.js"; const SESSION_CATALOG_SEARCH_MAX_UTF16_UNITS = 500; +const SESSION_CATALOG_SHARE_WINDOW_MS = 3_000; +const SESSION_CATALOG_LIST_CACHE_MAX_ENTRIES = 128; function createSessionCatalogRequestNodeSnapshot(): NonNullable< SessionCatalogListProviderParams["listNodes"] @@ -64,8 +67,41 @@ function catalogError(error: unknown): { code: string; message: string } { }; } +type CatalogRegistrationSnapshot = { + registry: PluginRegistry | null; + source: PluginRegistry["sessionCatalogs"] | undefined; + registrations: PluginRegistry["sessionCatalogs"]; + providers: SessionCatalogProvider[]; +}; + +let cachedCatalogRegistrations: CatalogRegistrationSnapshot | undefined; + +function catalogRegistrationSnapshot(): CatalogRegistrationSnapshot { + const registry = getActivePluginSessionExtensionRegistry(); + const source = registry?.sessionCatalogs; + if ( + cachedCatalogRegistrations?.registry === registry && + cachedCatalogRegistrations.source === source + ) { + return cachedCatalogRegistrations; + } + const sortedRegistrations = (source ?? []).toSorted((left, right) => + left.provider.id.localeCompare(right.provider.id), + ); + // Plugin registration arrays are process-stable until the active registry seam changes. Hoisting + // this sort avoids rebuilding identical order every poll; registry/list identity invalidates it. + // A stale snapshot would route requests to retired plugin instances, so callers share this owner. + cachedCatalogRegistrations = { + registry, + source, + registrations: sortedRegistrations, + providers: sortedRegistrations.map((entry) => entry.provider), + }; + return cachedCatalogRegistrations; +} + function providers(): SessionCatalogProvider[] { - return registrations().map((entry) => entry.provider); + return catalogRegistrationSnapshot().providers; } export function resolveSessionCatalogProvider( @@ -75,9 +111,7 @@ export function resolveSessionCatalogProvider( } function registrations() { - return (getActivePluginSessionExtensionRegistry()?.sessionCatalogs ?? []).toSorted( - (left, right) => left.provider.id.localeCompare(right.provider.id), - ); + return catalogRegistrationSnapshot().registrations; } type SessionCatalogCreateTargetResolution = @@ -93,10 +127,19 @@ const providerCreateTargetsByConfig = new WeakMap< WeakMap> >(); -const catalogListsByConfig = new WeakMap< - OpenClawConfig, - Map> ->(); +type CatalogListResult = { catalogs: SessionCatalog[] }; + +type CatalogListCacheEntry = { + expiresAt?: number; + result: Promise; +}; + +type CatalogListCacheState = { + registrations: CatalogRegistrationSnapshot; + entries: Map; +}; + +const catalogListsByConfig = new WeakMap(); function providerCreateTargetCache( config: OpenClawConfig, @@ -185,15 +228,16 @@ function sessionCatalogListKey(params: { ]); } -function catalogListInflightMap( +function catalogListCache( config: OpenClawConfig, -): Map> { - let inFlight = catalogListsByConfig.get(config); - if (!inFlight) { - inFlight = new Map(); - catalogListsByConfig.set(config, inFlight); + registrationSnapshot: CatalogRegistrationSnapshot, +): Map { + let state = catalogListsByConfig.get(config); + if (!state || state.registrations !== registrationSnapshot) { + state = { registrations: registrationSnapshot, entries: new Map() }; + catalogListsByConfig.set(config, state); } - return inFlight; + return state.entries; } function providerOrRespond( @@ -267,15 +311,23 @@ export const sessionCatalogHandlers: GatewayRequestHandlers = { ); return; } + const catalogRegistrations = catalogRegistrationSnapshot(); let selected: SessionCatalogProvider[]; if (request.catalogId) { - const provider = providerOrRespond(request.catalogId, respond); + const provider = catalogRegistrations.providers.find( + (candidate) => candidate.id === request.catalogId, + ); if (!provider) { + respond( + false, + undefined, + errorShape(ErrorCodes.INVALID_REQUEST, `unknown session catalog: ${request.catalogId}`), + ); return; } selected = [provider]; } else { - selected = providers(); + selected = catalogRegistrations.providers; } const config = context.getRuntimeConfig(); const resolvedAgent = resolveAgentIdOrRespondError({ @@ -295,14 +347,19 @@ export const sessionCatalogHandlers: GatewayRequestHandlers = { request, search, }); - const inFlight = catalogListInflightMap(config); - const pending = inFlight.get(listKey); - if (pending) { + const cache = catalogListCache(config, catalogRegistrations); + const cached = cache.get(listKey); + if (cached && (cached.expiresAt === undefined || cached.expiresAt > Date.now())) { // progressId is connection-owned and excluded from the work key. Followers skip progressive // frames and receive only the authoritative final result emitted for every caller below. - respond(true, await pending); + cache.delete(listKey); + cache.set(listKey, cached); + respond(true, await cached.result); return; } + if (cached) { + cache.delete(listKey); + } const operation = (async () => { const requestEntries = createSessionCatalogRequestEntrySnapshot({ cfg: config, @@ -357,14 +414,29 @@ export const sessionCatalogHandlers: GatewayRequestHandlers = { ); return { catalogs: catalogList }; })(); - // Sharing ends when this exact promise settles; later polls always execute against fresh state. - inFlight.set(listKey, operation); - try { - respond(true, await operation); - } finally { - if (inFlight.get(listKey) === operation) { - inFlight.delete(listKey); + const entry: CatalogListCacheEntry = { result: operation }; + // Exact request/config/registration results remain shareable for 3s after settling. This catches + // out-of-phase clients but expires before the UI's 5s fast follow, so changed rows surface there. + // Expired and rejected work is removed; retaining it would mask provider recovery or new sessions. + cache.set(listKey, entry); + while (cache.size > SESSION_CATALOG_LIST_CACHE_MAX_ENTRIES) { + const oldest = cache.keys().next(); + if (oldest.done) { + break; } + cache.delete(oldest.value); + } + try { + const result = await operation; + if (cache.get(listKey) === entry) { + entry.expiresAt = Date.now() + SESSION_CATALOG_SHARE_WINDOW_MS; + } + respond(true, result); + } catch (error) { + if (cache.get(listKey) === entry) { + cache.delete(listKey); + } + throw error; } }, diff --git a/src/infra/executable-path.test.ts b/src/infra/executable-path.test.ts index 72af5886709b..2c07f86dcf63 100644 --- a/src/infra/executable-path.test.ts +++ b/src/infra/executable-path.test.ts @@ -93,6 +93,39 @@ describe("executable path helpers", () => { }); }); + it("slides PATH hit and miss expiry for steady pollers", async () => { + await withTempDir({ prefix: "openclaw-exec-path-" }, async (base) => { + const binDir = path.join(base, "bin"); + await fs.mkdir(binDir); + const executable = path.join(binDir, "runner"); + await fs.writeFile(executable, "#!/bin/sh\nexit 0\n", { mode: 0o755 }); + let now = 1_000; + const nowSpy = vi.spyOn(Date, "now").mockImplementation(() => now); + const statSpy = vi.spyOn(nodeFs, "statSync"); + try { + expect(resolveExecutableFromPathEnv("runner", binDir)).toBe(executable); + expect(resolveExecutableFromPathEnv("missing", binDir)).toBeUndefined(); + const initialProbeCount = statSpy.mock.calls.length; + + now += 59_000; + expect(resolveExecutableFromPathEnv("runner", binDir)).toBe(executable); + expect(resolveExecutableFromPathEnv("missing", binDir)).toBeUndefined(); + now += 59_000; + expect(resolveExecutableFromPathEnv("runner", binDir)).toBe(executable); + expect(resolveExecutableFromPathEnv("missing", binDir)).toBeUndefined(); + expect(statSpy).toHaveBeenCalledTimes(initialProbeCount); + + now += 60_001; + expect(resolveExecutableFromPathEnv("runner", binDir)).toBe(executable); + expect(resolveExecutableFromPathEnv("missing", binDir)).toBeUndefined(); + expect(statSpy.mock.calls.length).toBeGreaterThan(initialProbeCount); + } finally { + nowSpy.mockRestore(); + statSpy.mockRestore(); + } + }); + }); + it("does not reuse relative PATH probes after cwd changes", async () => { await withTempDir({ prefix: "openclaw-exec-path-" }, async (base) => { const firstCwd = path.join(base, "first"); diff --git a/src/infra/executable-path.ts b/src/infra/executable-path.ts index 244786008670..d69409a2527e 100644 --- a/src/infra/executable-path.ts +++ b/src/infra/executable-path.ts @@ -169,9 +169,12 @@ export function resolveExecutableFromPathEnv( ): string | undefined { const cacheKey = executablePathCacheKey(executable, pathEnv, env, options?.includeExtensionless); const cached = executablePathCache.get(cacheKey); - if (cached && cached.expiresAt > Date.now()) { - // PATH probes synchronously stat every candidate. Reuse hits and misses until config reload or - // this short TTL expires; otherwise catalog polling repeatedly blocks the Gateway event loop. + const now = Date.now(); + if (cached && cached.expiresAt > now) { + // Hits and misses remain valid while the same PATH/PATHEXT/cwd key is used; config reload clears + // the map. Installs/removals under an unchanged key intentionally need reload or a 60s idle gap, + // because steady pollers must never fall back into synchronous PATH stat loops. + cached.expiresAt = now + EXECUTABLE_PATH_CACHE_TTL_MS; executablePathCache.delete(cacheKey); executablePathCache.set(cacheKey, cached); return cached.resolved ?? undefined;