From 9b61a6779d7607ad55d42e3e143ed4fc839118c5 Mon Sep 17 00:00:00 2001 From: Josh Lane Date: Mon, 28 Sep 2026 08:38:45 -0700 Subject: [PATCH 1/4] fix(claude): stream and cache session import descriptors --- .../agent/providers/claude/agent.test.ts | 141 ++++++++++++++++++ .../server/agent/providers/claude/agent.ts | 88 ++++++++--- 2 files changed, 212 insertions(+), 17 deletions(-) diff --git a/packages/server/src/server/agent/providers/claude/agent.test.ts b/packages/server/src/server/agent/providers/claude/agent.test.ts index dbd10aea4d..6ca0938bce 100644 --- a/packages/server/src/server/agent/providers/claude/agent.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.test.ts @@ -1,6 +1,8 @@ +import { execFile } from "node:child_process"; import * as fs from "node:fs/promises"; import * as os from "node:os"; import * as path from "node:path"; +import { promisify } from "node:util"; import { afterEach, describe, expect, test, vi } from "vitest"; import type { PermissionResult, SDKMessage, SDKUserMessage } from "@anthropic-ai/claude-agent-sdk"; @@ -1666,6 +1668,145 @@ describe("normalizeClaudeAskUserQuestionUpdatedInput", () => { }); describe("ClaudeAgentClient.listImportableSessions", () => { + test("refreshes descriptors after append, rewrite, replacement, and deletion", async () => { + const configDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-claude-import-cache-")); + try { + const projectDir = path.join(configDir, "projects", "changing-transcript"); + await fs.mkdir(projectDir, { recursive: true }); + const sessionFile = path.join(projectDir, "changing.jsonl"); + const initial = `${JSON.stringify({ + type: "user", + sessionId: "changing-session", + cwd: "/changing-project", + message: { content: "First prompt" }, + })}\n`; + function title(customTitle: string): string { + return `${JSON.stringify({ type: "custom-title", customTitle })}\n`; + } + await fs.writeFile(sessionFile, initial + title("First title")); + const timestamp = new Date("2026-06-01T12:00:00.000Z"); + await fs.utimes(sessionFile, timestamp, timestamp); + + const client = new ClaudeAgentClient({ + logger: createTestLogger(), + runtimeSettings: { env: { CLAUDE_CONFIG_DIR: configDir } }, + }); + const first = await client.listImportableSessions({ limit: 1 }); + expect(first.map((session) => session.title)).toEqual(["First title"]); + // Callers must not be able to mutate a cached descriptor, including its Date. + first[0]!.title = "Changed by caller"; + first[0]!.lastActivityAt.setTime(0); + const repeated = await Promise.all([ + client.listImportableSessions({ limit: 1 }), + client.listImportableSessions({ limit: 1 }), + ]); + expect(repeated.map((sessions) => [sessions[0]!.title, sessions[0]!.lastActivityAt])).toEqual( + [ + ["First title", timestamp], + ["First title", timestamp], + ], + ); + + await fs.appendFile(sessionFile, title("After append")); + expect( + (await client.listImportableSessions({ limit: 1 })).map((session) => session.title), + ).toEqual(["After append"]); + + // Same-length rewrites with a restored mtime still invalidate the descriptor. + await fs.writeFile(sessionFile, initial + title("Other title")); + await fs.utimes(sessionFile, timestamp, timestamp); + expect( + (await client.listImportableSessions({ limit: 1 })).map((session) => session.title), + ).toEqual(["Other title"]); + + const replacement = path.join(projectDir, "replacement.tmp"); + await fs.writeFile(replacement, initial + title("Final title")); + await fs.utimes(replacement, timestamp, timestamp); + await fs.rename(replacement, sessionFile); + expect( + (await client.listImportableSessions({ limit: 1 })).map((session) => session.title), + ).toEqual(["Final title"]); + + await fs.unlink(sessionFile); + await expect(client.listImportableSessions({ limit: 1 })).resolves.toEqual([]); + } finally { + await fs.rm(configDir, { recursive: true, force: true }); + } + }); + + test("discovers a transcript larger than the available heap", async () => { + const configDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-claude-import-heap-")); + try { + const projectDir = path.join(configDir, "projects", "large-transcript"); + await fs.mkdir(projectDir, { recursive: true }); + const sessionFile = path.join(projectDir, "large.jsonl"); + const transcript = await fs.open(sessionFile, "w"); + try { + await transcript.writeFile( + `${JSON.stringify({ + type: "user", + sessionId: "large-session", + cwd: "/large-project", + message: { content: "First prompt" }, + })}\r\n`, + ); + const block = `${JSON.stringify({ type: "assistant", text: "x".repeat(1024) })}\n`.repeat( + 1024, + ); + for (let index = 0; index < 128; index += 1) { + await transcript.writeFile(block); + } + // A malformed record must not hide the final unterminated record. + await transcript.writeFile( + `incomplete json\n${JSON.stringify({ + type: "user", + message: { content: "Last prompt 🐢" }, + })}`, + ); + } finally { + await transcript.close(); + } + + const { stdout } = await promisify(execFile)( + process.execPath, + [ + "--max-old-space-size=96", + "--import", + import.meta.resolve("tsx"), + "--input-type=module", + "--eval", + ` + import { deepStrictEqual } from "node:assert"; + import { ClaudeAgentClient } from ${JSON.stringify(new URL("./agent.ts", import.meta.url).href)}; + import pino from ${JSON.stringify(import.meta.resolve("pino"))}; + const client = new ClaudeAgentClient({ + logger: pino({ level: "silent" }), + runtimeSettings: { env: { CLAUDE_CONFIG_DIR: ${JSON.stringify(configDir)} } }, + }); + const listings = await Promise.all(Array.from({ length: 4 }, () => client.listImportableSessions({ limit: 1 }))); + const sessions = listings[0]; + for (const listing of listings) deepStrictEqual(listing, sessions); + console.log(JSON.stringify(sessions)); + `, + ], + { timeout: 20_000 }, + ); + + expect(JSON.parse(stdout)).toEqual([ + { + providerHandleId: "large-session", + cwd: "/large-project", + title: "First prompt", + firstPromptPreview: "First prompt", + lastPromptPreview: "Last prompt 🐢", + lastActivityAt: (await fs.stat(sessionFile)).mtime.toISOString(), + }, + ]); + } finally { + await fs.rm(configDir, { recursive: true, force: true }); + } + }); + test("uses the latest native custom title and leaves fixture mtimes unchanged", async () => { const tmpConfigDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-claude-import-")); const previousConfigDir = process.env.CLAUDE_CONFIG_DIR; diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index e5153e2102..5a385232a5 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -4,6 +4,9 @@ import { randomUUID } from "node:crypto"; import fs from "node:fs"; import { promises } from "node:fs"; import path from "node:path"; +import { createInterface } from "node:readline"; +import { LRUCache } from "lru-cache"; +import pLimit from "p-limit"; import { type AgentDefinition, type CanUseTool, @@ -1506,6 +1509,18 @@ export class ClaudeAgentClient implements AgentClient { private readonly resolveBinary: () => Promise; private readonly resolveVersion: (signal?: AbortSignal) => Promise; private readonly rewindSdk: ClaudeRewindSdk; + // Share the limit across overlapping picker requests, not just within each listing. + private readonly importDescriptorLimit = pLimit(4); + private readonly importDescriptorReads = new Map< + string, + Promise + >(); + private readonly importDescriptorCache = new LRUCache({ + max: 500, + maxSize: 8 * 1024 * 1024, + sizeCalculation: (descriptor, key) => + 128 + 2 * (JSON.stringify(descriptor).length + key.length), + }); constructor(options: ClaudeAgentClientOptions) { this.defaults = options.defaults; @@ -1636,13 +1651,40 @@ export class ClaudeAgentClient implements AgentClient { rootIsProjectDir: Boolean(options?.cwd), }); const parsed = await Promise.all( - candidates.map((candidate) => parseClaudeSessionDescriptor(candidate.path, candidate.mtime)), + candidates.map((candidate) => this.readImportDescriptor(candidate)), ); return parsed .filter((session): session is ImportableProviderSession => session !== null) .slice(0, limit); } + private async readImportDescriptor( + candidate: ClaudeSessionCandidate, + ): Promise { + const cached = this.importDescriptorCache.get(candidate.cacheKey); + if (cached) return structuredClone(cached); + + let pending = this.importDescriptorReads.get(candidate.cacheKey); + if (!pending) { + pending = this.importDescriptorLimit(async () => { + const descriptor = await parseClaudeSessionDescriptor(candidate.path, candidate.mtime); + if (descriptor) { + const current = await fsPromises.stat(candidate.path).catch(() => null); + if (current && claudeSessionCacheKey(candidate.path, current) === candidate.cacheKey) { + // Copy the small descriptor so preview slices cannot retain a large transcript line. + this.importDescriptorCache.set(candidate.cacheKey, structuredClone(descriptor)); + } + } + return descriptor; + }).finally(() => { + this.importDescriptorReads.delete(candidate.cacheKey); + }); + this.importDescriptorReads.set(candidate.cacheKey, pending); + } + const descriptor = await pending; + return descriptor ? structuredClone(descriptor) : null; + } + async importSession(input: ImportProviderSessionInput, context: ImportProviderSessionContext) { return importSessionFromPersistence({ provider: "claude", @@ -6166,6 +6208,11 @@ function createAsyncMessageInput(): AsyncMessageInput { interface ClaudeSessionCandidate { path: string; mtime: Date; + cacheKey: string; +} + +function claudeSessionCacheKey(filePath: string, stats: fs.Stats): string { + return `${filePath}\0${stats.dev}:${stats.ino}:${stats.size}:${stats.mtimeMs}:${stats.ctimeMs}`; } async function pathExists(target: string): Promise { @@ -6211,7 +6258,12 @@ async function collectRecentClaudeSessions( fileEntries.map(async (fullPath) => { try { const fileStats = await fsPromises.stat(fullPath); - return { path: fullPath, mtime: fileStats.mtime }; + if (!fileStats.isFile()) return null; + return { + path: fullPath, + mtime: fileStats.mtime, + cacheKey: claudeSessionCacheKey(fullPath, fileStats), + }; } catch { return null; } @@ -6292,13 +6344,6 @@ async function parseClaudeSessionDescriptor( filePath: string, mtime: Date, ): Promise { - let content: string; - try { - content = await fsPromises.readFile(filePath, "utf8"); - } catch { - return null; - } - const acc: ClaudeSessionDescriptorAccumulator = { sessionId: null, cwd: null, @@ -6309,15 +6354,24 @@ async function parseClaudeSessionDescriptor( lastPromptPreview: null, }; - for (const line of content.split(/\r?\n/)) { - if (!line || !shouldParseClaudeSessionDescriptorLine(line, acc)) continue; - let entry: unknown; - try { - entry = JSON.parse(line); - } catch { - continue; + const stream = fs.createReadStream(filePath, { encoding: "utf8" }); + const lines = createInterface({ input: stream, crlfDelay: Infinity }); + try { + for await (const line of lines) { + if (!line || !shouldParseClaudeSessionDescriptorLine(line, acc)) continue; + let entry: unknown; + try { + entry = JSON.parse(line); + } catch { + continue; + } + applyClaudeSessionEntryToAccumulator(entry, acc); } - applyClaudeSessionEntryToAccumulator(entry, acc); + } catch { + return null; + } finally { + lines.close(); + stream.destroy(); } const { sessionId, cwd } = acc; From c4a8efddb1f63db1c4d68be3dff12ce1690183db Mon Sep 17 00:00:00 2001 From: Josh Lane Date: Tue, 6 Oct 2026 09:08:19 -0400 Subject: [PATCH 2/4] fix(claude): bound and schedule import reads Reject JSONL records over 4 MiB explicitly to preserve metadata. Limit shared reads to four globally and one per listing for fairness. Deduplicate reads by cacheKey with independent subscriber cancellation. Set a 45-second provider deadline before the 60-second client timeout. Clone returned descriptors to isolate overlapping callers. Validation: 97 focused tests passed; independent review approved. Detected-By: user --- .../agent/providers/claude/agent.test.ts | 285 ++++++++++++++++++ .../server/agent/providers/claude/agent.ts | 130 +++++--- .../claude/import-descriptor-scheduler.ts | 80 +++++ 3 files changed, 447 insertions(+), 48 deletions(-) create mode 100644 packages/server/src/server/agent/providers/claude/import-descriptor-scheduler.ts diff --git a/packages/server/src/server/agent/providers/claude/agent.test.ts b/packages/server/src/server/agent/providers/claude/agent.test.ts index 6ca0938bce..29cc75380c 100644 --- a/packages/server/src/server/agent/providers/claude/agent.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.test.ts @@ -17,6 +17,7 @@ import { resolveClaudeCodeVersion, toClaudeSdkMcpConfig, } from "./agent.js"; +import { createImportDescriptorScheduler } from "./import-descriptor-scheduler.js"; import { claudeProjectDirSync } from "./project-dir.js"; import { streamSession } from "../test-utils/session-stream-adapter.js"; import type { @@ -1707,6 +1708,13 @@ describe("ClaudeAgentClient.listImportableSessions", () => { ], ); + repeated[0]![0]!.title = "Changed by overlapping caller"; + repeated[0]![0]!.lastActivityAt.setTime(0); + expect([repeated[1]![0]!.title, repeated[1]![0]!.lastActivityAt]).toEqual([ + "First title", + timestamp, + ]); + await fs.appendFile(sessionFile, title("After append")); expect( (await client.listImportableSessions({ limit: 1 })).map((session) => session.title), @@ -1807,6 +1815,143 @@ describe("ClaudeAgentClient.listImportableSessions", () => { } }); + test("rejects a single record larger than the heap without allocating the full line", async () => { + const configDir = await fs.mkdtemp( + path.join(os.tmpdir(), "paseo-claude-import-single-record-"), + ); + try { + const projectDir = path.join(configDir, "projects", "single-record"); + await fs.mkdir(projectDir, { recursive: true }); + const sessionFile = path.join(projectDir, "single-record.jsonl"); + const transcript = await fs.open(sessionFile, "w"); + try { + await transcript.writeFile('{"payload":"'); + const block = "x".repeat(1024 * 1024); + for (let index = 0; index < 128; index += 1) await transcript.writeFile(block); + await transcript.writeFile('","type":"custom-title","customTitle":"Latest title"}'); + } finally { + await transcript.close(); + } + const { stdout } = await promisify(execFile)( + process.execPath, + [ + "--max-old-space-size=96", + "--import", + import.meta.resolve("tsx"), + "--input-type=module", + "--eval", + ` + import { rejects } from "node:assert"; + import { ClaudeAgentClient } from ${JSON.stringify(new URL("./agent.ts", import.meta.url).href)}; + import pino from ${JSON.stringify(import.meta.resolve("pino"))}; + const client = new ClaudeAgentClient({ + logger: pino({ level: "silent" }), + runtimeSettings: { env: { CLAUDE_CONFIG_DIR: ${JSON.stringify(configDir)} } }, + }); + await rejects(client.listImportableSessions({ limit: 1 }), { + name: "ClaudeImportRecordTooLargeError", recordNumber: 1, maxRecordBytes: 4194304, + }); + console.log("Oversized record rejected"); + `, + ], + { timeout: 20_000 }, + ); + expect(stdout.trim()).toEqual("Oversized record rejected"); + } finally { + await fs.rm(configDir, { recursive: true, force: true }); + } + }); + + test("keeps metadata and UTF-8 in a record exactly at the byte limit", async () => { + const configDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-claude-import-limit-")); + try { + const projectDir = path.join(configDir, "projects", "boundary"); + await fs.mkdir(projectDir, { recursive: true }); + const prefix = '{"padding":"'; + // Split the four-byte turtle across a 64 KiB input chunk boundary. + const padding = "x".repeat(64 * 1024 - Buffer.byteLength(prefix) - 1); + const suffix = + '","type":"user","sessionId":"boundary-session","cwd":"/boundary","message":{"content":"Hello 🐢"}}'; + const remaining = 4 * 1024 * 1024 - Buffer.byteLength(prefix + padding + "🐢" + suffix); + await fs.writeFile( + path.join(projectDir, "boundary.jsonl"), + prefix + padding + "🐢" + "x".repeat(remaining) + suffix, + ); + const client = new ClaudeAgentClient({ + logger: createTestLogger(), + runtimeSettings: { env: { CLAUDE_CONFIG_DIR: configDir } }, + }); + expect( + (await client.listImportableSessions({ limit: 1 })).map((session) => ({ + providerHandleId: session.providerHandleId, + cwd: session.cwd, + title: session.title, + firstPromptPreview: session.firstPromptPreview, + lastPromptPreview: session.lastPromptPreview, + })), + ).toEqual([ + { + providerHandleId: "boundary-session", + cwd: "/boundary", + title: "Hello 🐢", + firstPromptPreview: "Hello 🐢", + lastPromptPreview: "Hello 🐢", + }, + ]); + } finally { + await fs.rm(configDir, { recursive: true, force: true }); + } + }); + + test("fails explicitly on an oversized record without returning stale metadata", async () => { + const configDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-claude-import-record-")); + try { + const projectDir = path.join(configDir, "projects", "oversized"); + await fs.mkdir(projectDir, { recursive: true }); + const sessionFile = path.join(projectDir, "oversized.jsonl"); + await fs.writeFile( + sessionFile, + `${JSON.stringify({ + type: "user", + sessionId: "oversized-session", + cwd: "/oversized-project", + message: { content: "First prompt" }, + })}\n`, + ); + const client = new ClaudeAgentClient({ + logger: createTestLogger(), + runtimeSettings: { env: { CLAUDE_CONFIG_DIR: configDir } }, + }); + expect((await client.listImportableSessions({ limit: 1 }))[0]!.title).toEqual("First prompt"); + // Metadata placed after the payload defeats prefix-only relevance checks. + await fs.appendFile( + sessionFile, + JSON.stringify({ + payload: "x".repeat(4 * 1024 * 1024), + type: "custom-title", + customTitle: "Latest title", + }), + ); + await expect(client.listImportableSessions({ limit: 1 })).rejects.toMatchObject({ + name: "ClaudeImportRecordTooLargeError", + filePath: sessionFile, + recordNumber: 2, + maxRecordBytes: 4 * 1024 * 1024, + }); + } finally { + await fs.rm(configDir, { recursive: true, force: true }); + } + }); + + test("rejects a cancelled listing before doing filesystem work", async () => { + const controller = new AbortController(); + controller.abort(new Error("Picker closed")); + const client = new ClaudeAgentClient({ logger: createTestLogger() }); + await expect(client.listImportableSessions({ signal: controller.signal })).rejects.toThrow( + "Picker closed", + ); + }); + test("uses the latest native custom title and leaves fixture mtimes unchanged", async () => { const tmpConfigDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-claude-import-")); const previousConfigDir = process.env.CLAUDE_CONFIG_DIR; @@ -3591,3 +3736,143 @@ describe("Claude question permission notifications", () => { expect(request.description).toBeUndefined(); }); }); + +describe("Claude import descriptor scheduling", () => { + function controlledReads() { + const started: string[] = []; + const reads = new Map< + string, + { resolve: (value: string) => void; reject: (reason: unknown) => void } + >(); + const signals = new Map(); + const sharedScheduler = createImportDescriptorScheduler( + (candidate: { cacheKey: string }, signal) => { + started.push(candidate.cacheKey); + signals.set(candidate.cacheKey, signal); + return new Promise((resolve, reject) => { + reads.set(candidate.cacheKey, { resolve, reject }); + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); + }, + ); + function scheduler(candidates: string[], signal: AbortSignal): Promise { + return sharedScheduler( + candidates.map((cacheKey) => ({ cacheKey })), + signal, + ); + } + return { scheduler, started, reads, signals }; + } + + test("shares a cold read across overlapping listings and leaves capacity for unrelated work", async () => { + const { scheduler, started, reads } = controlledReads(); + const signal = new AbortController().signal; + const overlapping = Array.from({ length: 4 }, () => scheduler(["shared"], signal)); + await vi.waitFor(() => expect(started).toEqual(["shared"])); + const unrelated = scheduler(["unrelated"], signal); + await vi.waitFor(() => expect(started).toEqual(["shared", "unrelated"])); + reads.get("unrelated")!.resolve("unrelated descriptor"); + await expect(unrelated).resolves.toEqual(["unrelated descriptor"]); + reads.get("shared")!.resolve("shared descriptor"); + await expect(Promise.all(overlapping)).resolves.toEqual( + Array.from({ length: 4 }, () => ["shared descriptor"]), + ); + // A settled entry must not become an accidental second cache. + const later = scheduler(["shared"], signal); + await vi.waitFor(() => expect(started).toEqual(["shared", "unrelated", "shared"])); + reads.get("shared")!.resolve("updated descriptor"); + await expect(later).resolves.toEqual(["updated descriptor"]); + }); + + test("cancelling the initiating listing keeps its other subscriber alive", async () => { + const { scheduler, started, reads } = controlledReads(); + const initiator = new AbortController(); + const first = scheduler(["shared", "unused"], initiator.signal); + const firstRejected = expect(first).rejects.toThrow("First picker closed"); + const second = scheduler(["shared"], new AbortController().signal); + await vi.waitFor(() => expect(started).toEqual(["shared"])); + initiator.abort(new Error("First picker closed")); + await firstRejected; + reads.get("shared")!.resolve("shared descriptor"); + await expect(second).resolves.toEqual(["shared descriptor"]); + expect(started).toEqual(["shared"]); + }); + + test("the final subscriber cancellation aborts the read and allows a fresh subscription", async () => { + const { scheduler, started, signals } = controlledReads(); + const firstController = new AbortController(); + const secondController = new AbortController(); + const first = scheduler(["shared"], firstController.signal); + const second = scheduler(["shared"], secondController.signal); + const firstRejected = expect(first).rejects.toThrow("First picker closed"); + const secondRejected = expect(second).rejects.toThrow("Final picker closed"); + await vi.waitFor(() => expect(started).toEqual(["shared"])); + firstController.abort(new Error("First picker closed")); + await firstRejected; + expect(signals.get("shared")!.aborted).toEqual(false); + secondController.abort(new Error("Final picker closed")); + await secondRejected; + expect(signals.get("shared")!.aborted).toEqual(true); + const retryController = new AbortController(); + const retry = scheduler(["shared"], retryController.signal); + const retryRejected = expect(retry).rejects.toThrow("Retry closed"); + await vi.waitFor(() => expect(started).toEqual(["shared", "shared"])); + retryController.abort(new Error("Retry closed")); + await retryRejected; + expect(signals.get("shared")!.aborted).toEqual(true); + }); + + test("gives a small later listing a turn before a large listing finishes", async () => { + const { scheduler, started, reads } = controlledReads(); + const signal = new AbortController().signal; + const large = scheduler( + Array.from({ length: 500 }, (_, index) => `large-${index}`), + signal, + ); + await vi.waitFor(() => expect(started).toEqual(["large-0"])); + const small = scheduler(["small"], signal); + await vi.waitFor(() => expect(started).toEqual(["large-0", "small"])); + reads.get("small")!.resolve("small descriptor"); + await expect(small).resolves.toEqual(["small descriptor"]); + expect(started).toEqual(["large-0", "small"]); + // Cancel the large listing so this test does not need to service 499 reads. + reads.get("large-0")!.reject(new Error("Finished fairness check")); + await expect(large).rejects.toThrow("Finished fairness check"); + }); + + test("bounds active reads and skips cancelled queued and remaining candidates", async () => { + const { scheduler, started, reads } = controlledReads(); + const controllers = Array.from({ length: 5 }, () => new AbortController()); + const listings = controllers.map((controller, index) => + scheduler([`${index}-first`, `${index}-next`], controller.signal), + ); + // Attach handlers before aborting any promises. + function errorMessage(error: Error): string { + return error.message; + } + const settled = listings.map((listing) => listing.catch(errorMessage)); + await vi.waitFor(() => expect(started).toEqual(["0-first", "1-first", "2-first", "3-first"])); + controllers[4]!.abort(new Error("Queued picker closed")); + await expect(settled[4]).resolves.toEqual("Queued picker closed"); + controllers[0]!.abort(new Error("Active picker closed")); + await expect(settled[0]).resolves.toEqual("Active picker closed"); + reads.get("1-first")!.reject(new Error("Stop 1")); + reads.get("2-first")!.reject(new Error("Stop 2")); + reads.get("3-first")!.reject(new Error("Stop 3")); + await expect(Promise.all(settled)).resolves.toEqual([ + "Active picker closed", + "Stop 1", + "Stop 2", + "Stop 3", + "Queued picker closed", + ]); + expect(started).toEqual(["0-first", "1-first", "2-first", "3-first"]); + // A cancelled queue entry must not consume a later listing's read slot. + const later = scheduler(["later"], new AbortController().signal); + await vi.waitFor(() => + expect(started).toEqual(["0-first", "1-first", "2-first", "3-first", "later"]), + ); + reads.get("later")!.resolve("later descriptor"); + await expect(later).resolves.toEqual(["later descriptor"]); + }); +}); diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 5a385232a5..86d82cba31 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -4,9 +4,8 @@ import { randomUUID } from "node:crypto"; import fs from "node:fs"; import { promises } from "node:fs"; import path from "node:path"; -import { createInterface } from "node:readline"; import { LRUCache } from "lru-cache"; -import pLimit from "p-limit"; +import { createImportDescriptorScheduler } from "./import-descriptor-scheduler.js"; import { type AgentDefinition, type CanUseTool, @@ -1509,12 +1508,9 @@ export class ClaudeAgentClient implements AgentClient { private readonly resolveBinary: () => Promise; private readonly resolveVersion: (signal?: AbortSignal) => Promise; private readonly rewindSdk: ClaudeRewindSdk; - // Share the limit across overlapping picker requests, not just within each listing. - private readonly importDescriptorLimit = pLimit(4); - private readonly importDescriptorReads = new Map< - string, - Promise - >(); + private readonly scheduleImportDescriptors = createImportDescriptorScheduler( + (candidate: ClaudeSessionCandidate, signal) => this.readImportDescriptor(candidate, signal), + ); private readonly importDescriptorCache = new LRUCache({ max: 500, maxSize: 8 * 1024 * 1024, @@ -1636,52 +1632,47 @@ export class ClaudeAgentClient implements AgentClient { } async listImportableSessions( - options?: ListImportableSessionsOptions, + options?: ListImportableSessionsOptions & { signal?: AbortSignal }, ): Promise { + // Leave time to deliver the provider error before the client RPC expires at 60 seconds. + const deadline = AbortSignal.timeout(45_000); + const signal = options?.signal ? AbortSignal.any([options.signal, deadline]) : deadline; + signal.throwIfAborted(); const configDir = claudeConfigDir(this.buildProviderEnv()); const sessionsRoot = options?.cwd ? claudeProjectDirSync(options.cwd, { configDir }) : path.join(configDir, "projects"); - if (!(await pathExists(sessionsRoot))) { - return []; - } + const rootExists = await pathExists(sessionsRoot); + signal.throwIfAborted(); + if (!rootExists) return []; const limit = options?.limit ?? 20; const scanLimit = Math.min(options?.scanLimit ?? limit * 3, 500); const candidates = await collectRecentClaudeSessions(sessionsRoot, scanLimit, { rootIsProjectDir: Boolean(options?.cwd), }); - const parsed = await Promise.all( - candidates.map((candidate) => this.readImportDescriptor(candidate)), - ); + const parsed = await this.scheduleImportDescriptors(candidates, signal); return parsed .filter((session): session is ImportableProviderSession => session !== null) - .slice(0, limit); + .slice(0, limit) + .map((session) => structuredClone(session)); } private async readImportDescriptor( candidate: ClaudeSessionCandidate, + signal: AbortSignal, ): Promise { + signal.throwIfAborted(); const cached = this.importDescriptorCache.get(candidate.cacheKey); if (cached) return structuredClone(cached); - let pending = this.importDescriptorReads.get(candidate.cacheKey); - if (!pending) { - pending = this.importDescriptorLimit(async () => { - const descriptor = await parseClaudeSessionDescriptor(candidate.path, candidate.mtime); - if (descriptor) { - const current = await fsPromises.stat(candidate.path).catch(() => null); - if (current && claudeSessionCacheKey(candidate.path, current) === candidate.cacheKey) { - // Copy the small descriptor so preview slices cannot retain a large transcript line. - this.importDescriptorCache.set(candidate.cacheKey, structuredClone(descriptor)); - } - } - return descriptor; - }).finally(() => { - this.importDescriptorReads.delete(candidate.cacheKey); - }); - this.importDescriptorReads.set(candidate.cacheKey, pending); + const descriptor = await parseClaudeSessionDescriptor(candidate.path, candidate.mtime, signal); + if (descriptor) { + const current = await fsPromises.stat(candidate.path).catch(() => null); + signal.throwIfAborted(); + if (current && claudeSessionCacheKey(candidate.path, current) === candidate.cacheKey) { + this.importDescriptorCache.set(candidate.cacheKey, structuredClone(descriptor)); + } } - const descriptor = await pending; return descriptor ? structuredClone(descriptor) : null; } @@ -6340,9 +6331,23 @@ function applyClaudeSessionEntryToAccumulator( } } +class ClaudeImportRecordTooLargeError extends Error { + constructor( + readonly filePath: string, + readonly recordNumber: number, + readonly maxRecordBytes: number, + ) { + super( + `Cannot list Claude sessions: transcript ${filePath} record ${recordNumber} exceeds the ${maxRecordBytes}-byte import limit. Shorten or remove this record before retrying.`, + ); + this.name = "ClaudeImportRecordTooLargeError"; + } +} + async function parseClaudeSessionDescriptor( filePath: string, mtime: Date, + signal: AbortSignal, ): Promise { const acc: ClaudeSessionDescriptorAccumulator = { sessionId: null, @@ -6354,23 +6359,52 @@ async function parseClaudeSessionDescriptor( lastPromptPreview: null, }; - const stream = fs.createReadStream(filePath, { encoding: "utf8" }); - const lines = createInterface({ input: stream, crlfDelay: Infinity }); - try { - for await (const line of lines) { - if (!line || !shouldParseClaudeSessionDescriptorLine(line, acc)) continue; - let entry: unknown; - try { - entry = JSON.parse(line); - } catch { - continue; - } - applyClaudeSessionEntryToAccumulator(entry, acc); + // Reject rather than skip: type/cwd/title can occur after a huge payload, so a + // prefix cannot establish that dropping the record preserves the descriptor. + const maxRecordBytes = 4 * 1024 * 1024; + const record = Buffer.allocUnsafe(maxRecordBytes); + let recordBytes = 0; + let recordNumber = 1; + const stream = fs.createReadStream(filePath, { highWaterMark: 64 * 1024, signal }); + function applyRecord(): void { + const line = record.toString("utf8", 0, recordBytes); + if (!line || !shouldParseClaudeSessionDescriptorLine(line, acc)) return; + let entry: unknown; + try { + entry = JSON.parse(line); + } catch { + return; } - } catch { - return null; + applyClaudeSessionEntryToAccumulator(entry, acc); + } + try { + for await (const chunk of stream) { + signal.throwIfAborted(); + let start = 0; + while (start < chunk.length) { + const newline = chunk.indexOf(10, start); + const end = newline === -1 ? chunk.length : newline; + const length = end - start; + if (recordBytes + length > maxRecordBytes) { + throw new ClaudeImportRecordTooLargeError(filePath, recordNumber, maxRecordBytes); + } + chunk.copy(record, recordBytes, start, end); + recordBytes += length; + if (newline === -1) break; + applyRecord(); + recordBytes = 0; + recordNumber += 1; + start = newline + 1; + } + } + if (recordBytes > 0) applyRecord(); + } catch (error) { + signal.throwIfAborted(); + if (error instanceof ClaudeImportRecordTooLargeError) throw error; + // Transcripts can disappear between candidate discovery and opening them. + if (error instanceof Error && "code" in error && error.code === "ENOENT") return null; + throw error; } finally { - lines.close(); stream.destroy(); } diff --git a/packages/server/src/server/agent/providers/claude/import-descriptor-scheduler.ts b/packages/server/src/server/agent/providers/claude/import-descriptor-scheduler.ts new file mode 100644 index 0000000000..eccc1a842d --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/import-descriptor-scheduler.ts @@ -0,0 +1,80 @@ +import pLimit from "p-limit"; + +interface ImportDescriptorCandidate { + cacheKey: string; +} + +interface PendingDescriptor { + controller: AbortController; + subscribers: number; + settled: boolean; + promise: Promise; +} + +export function createImportDescriptorScheduler< + Candidate extends ImportDescriptorCandidate, + Descriptor, +>( + read: (candidate: Candidate, signal: AbortSignal) => Promise, +): (candidates: Candidate[], signal: AbortSignal) => Promise { + const limit = pLimit(4); + const reads = new Map>(); + + function acquire(candidate: Candidate): PendingDescriptor { + let pending = reads.get(candidate.cacheKey); + if (!pending) { + const controller = new AbortController(); + const entry: PendingDescriptor = { + controller, + subscribers: 0, + settled: false, + promise: limit(() => { + controller.signal.throwIfAborted(); + return read(candidate, controller.signal); + }).finally(() => { + entry.settled = true; + if (reads.get(candidate.cacheKey) === entry) reads.delete(candidate.cacheKey); + }), + }; + reads.set(candidate.cacheKey, entry); + pending = entry; + } + pending.subscribers += 1; + return pending; + } + + function release(candidate: Candidate, pending: PendingDescriptor): void { + pending.subscribers -= 1; + // The initiating listing has no special ownership of a shared read. + if (pending.subscribers === 0 && !pending.settled) { + if (reads.get(candidate.cacheKey) === pending) reads.delete(candidate.cacheKey); + pending.controller.abort(); + } + } + + return async (candidates, signal) => { + signal.throwIfAborted(); + let onAbort: (() => void) | undefined; + const aborted = new Promise((_resolve, reject) => { + onAbort = () => reject(signal.reason); + signal.addEventListener("abort", onAbort, { once: true }); + }); + try { + const results: Descriptor[] = []; + for (const candidate of candidates) { + signal.throwIfAborted(); + // One outstanding read per listing gives later requests a turn in the + // shared FIFO instead of putting hundreds of candidates ahead of them. + const pending = acquire(candidate); + try { + results.push(await Promise.race([pending.promise, aborted])); + } finally { + release(candidate, pending); + } + } + return results; + } finally { + if (onAbort) signal.removeEventListener("abort", onAbort); + } + }; +} From 568aaeaf93f2ad6b6531e2a123eb9189539e9b4d Mon Sep 17 00:00:00 2001 From: Josh Lane Date: Tue, 6 Oct 2026 09:26:34 -0400 Subject: [PATCH 3/4] fix(claude): parallelize resilient session scans Use up to four workers per listing with a shared four-read limit and bounded FIFO queuing. Preserve cache-key deduplication, subscriber cancellation, and candidate order. Log and skip oversized or unreadable transcripts so healthy sessions remain visible. Propagate programming and abort errors. Detected-By: user --- .../agent/providers/claude/agent.test.ts | 131 ++++++++++++++---- .../server/agent/providers/claude/agent.ts | 26 ++-- .../claude/import-descriptor-scheduler.ts | 24 +++- 3 files changed, 138 insertions(+), 43 deletions(-) diff --git a/packages/server/src/server/agent/providers/claude/agent.test.ts b/packages/server/src/server/agent/providers/claude/agent.test.ts index 29cc75380c..435c53d0f6 100644 --- a/packages/server/src/server/agent/providers/claude/agent.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.test.ts @@ -3,6 +3,7 @@ import * as fs from "node:fs/promises"; import * as os from "node:os"; import * as path from "node:path"; import { promisify } from "node:util"; +import pino from "pino"; import { afterEach, describe, expect, test, vi } from "vitest"; import type { PermissionResult, SDKMessage, SDKUserMessage } from "@anthropic-ai/claude-agent-sdk"; @@ -1815,7 +1816,7 @@ describe("ClaudeAgentClient.listImportableSessions", () => { } }); - test("rejects a single record larger than the heap without allocating the full line", async () => { + test("skips a single record larger than the heap without allocating the full line", async () => { const configDir = await fs.mkdtemp( path.join(os.tmpdir(), "paseo-claude-import-single-record-"), ); @@ -1841,22 +1842,20 @@ describe("ClaudeAgentClient.listImportableSessions", () => { "--input-type=module", "--eval", ` - import { rejects } from "node:assert"; + import { deepStrictEqual } from "node:assert"; import { ClaudeAgentClient } from ${JSON.stringify(new URL("./agent.ts", import.meta.url).href)}; import pino from ${JSON.stringify(import.meta.resolve("pino"))}; const client = new ClaudeAgentClient({ logger: pino({ level: "silent" }), runtimeSettings: { env: { CLAUDE_CONFIG_DIR: ${JSON.stringify(configDir)} } }, }); - await rejects(client.listImportableSessions({ limit: 1 }), { - name: "ClaudeImportRecordTooLargeError", recordNumber: 1, maxRecordBytes: 4194304, - }); - console.log("Oversized record rejected"); + deepStrictEqual(await client.listImportableSessions({ limit: 1 }), []); + console.log("Oversized transcript skipped"); `, ], { timeout: 20_000 }, ); - expect(stdout.trim()).toEqual("Oversized record rejected"); + expect(stdout.trim()).toEqual("Oversized transcript skipped"); } finally { await fs.rm(configDir, { recursive: true, force: true }); } @@ -1903,7 +1902,7 @@ describe("ClaudeAgentClient.listImportableSessions", () => { } }); - test("fails explicitly on an oversized record without returning stale metadata", async () => { + test("logs and skips an oversized transcript while retaining healthy sessions", async () => { const configDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-claude-import-record-")); try { const projectDir = path.join(configDir, "projects", "oversized"); @@ -1918,11 +1917,27 @@ describe("ClaudeAgentClient.listImportableSessions", () => { message: { content: "First prompt" }, })}\n`, ); + const records: unknown[] = []; const client = new ClaudeAgentClient({ - logger: createTestLogger(), + logger: pino( + { level: "warn" }, + { write: (line: string) => records.push(JSON.parse(line)) }, + ), runtimeSettings: { env: { CLAUDE_CONFIG_DIR: configDir } }, }); expect((await client.listImportableSessions({ limit: 1 }))[0]!.title).toEqual("First prompt"); + const healthyFile = path.join(projectDir, "healthy.jsonl"); + await fs.writeFile( + healthyFile, + JSON.stringify({ + type: "user", + sessionId: "healthy-session", + cwd: "/healthy-project", + message: { content: "Healthy prompt" }, + }), + ); + const timestamp = new Date("2026-01-01T00:00:00Z"); + await fs.utimes(healthyFile, timestamp, timestamp); // Metadata placed after the payload defeats prefix-only relevance checks. await fs.appendFile( sessionFile, @@ -1932,12 +1947,29 @@ describe("ClaudeAgentClient.listImportableSessions", () => { customTitle: "Latest title", }), ); - await expect(client.listImportableSessions({ limit: 1 })).rejects.toMatchObject({ - name: "ClaudeImportRecordTooLargeError", - filePath: sessionFile, - recordNumber: 2, - maxRecordBytes: 4 * 1024 * 1024, - }); + await expect(client.listImportableSessions({ limit: 1 })).resolves.toEqual([ + { + providerHandleId: "healthy-session", + cwd: "/healthy-project", + title: "Healthy prompt", + firstPromptPreview: "Healthy prompt", + lastPromptPreview: "Healthy prompt", + lastActivityAt: timestamp, + }, + ]); + expect(records).toEqual([ + expect.objectContaining({ + level: pino.levels.values.warn, + path: sessionFile, + msg: "Skipping Claude transcript with unreadable import metadata", + err: expect.objectContaining({ + type: "ClaudeImportRecordTooLargeError", + filePath: sessionFile, + recordNumber: 2, + maxRecordBytes: 4 * 1024 * 1024, + }), + }), + ]); } finally { await fs.rm(configDir, { recursive: true, force: true }); } @@ -3787,7 +3819,7 @@ describe("Claude import descriptor scheduling", () => { test("cancelling the initiating listing keeps its other subscriber alive", async () => { const { scheduler, started, reads } = controlledReads(); const initiator = new AbortController(); - const first = scheduler(["shared", "unused"], initiator.signal); + const first = scheduler(["shared"], initiator.signal); const firstRejected = expect(first).rejects.toThrow("First picker closed"); const second = scheduler(["shared"], new AbortController().signal); await vi.waitFor(() => expect(started).toEqual(["shared"])); @@ -3822,29 +3854,70 @@ describe("Claude import descriptor scheduling", () => { expect(signals.get("shared")!.aborted).toEqual(true); }); - test("gives a small later listing a turn before a large listing finishes", async () => { - const { scheduler, started, reads } = controlledReads(); - const signal = new AbortController().signal; + test("runs four reads per listing and gives a later small listing a bounded turn", async () => { + const { scheduler, started, reads, signals } = controlledReads(); + const controller = new AbortController(); const large = scheduler( Array.from({ length: 500 }, (_, index) => `large-${index}`), - signal, + controller.signal, + ); + const largeRejected = expect(large).rejects.toThrow("Fairness check complete"); + await vi.waitFor(() => expect(started).toEqual(["large-0", "large-1", "large-2", "large-3"])); + const small = scheduler(["small"], new AbortController().signal); + reads.get("large-0")!.resolve("first descriptor"); + await vi.waitFor(() => + expect(started).toEqual(["large-0", "large-1", "large-2", "large-3", "small"]), ); - await vi.waitFor(() => expect(started).toEqual(["large-0"])); - const small = scheduler(["small"], signal); - await vi.waitFor(() => expect(started).toEqual(["large-0", "small"])); reads.get("small")!.resolve("small descriptor"); await expect(small).resolves.toEqual(["small descriptor"]); - expect(started).toEqual(["large-0", "small"]); - // Cancel the large listing so this test does not need to service 499 reads. - reads.get("large-0")!.reject(new Error("Finished fairness check")); - await expect(large).rejects.toThrow("Finished fairness check"); + controller.abort(new Error("Fairness check complete")); + await largeRejected; + expect(signals.get("large-1")!.aborted).toEqual(true); + expect(signals.get("large-2")!.aborted).toEqual(true); + expect(signals.get("large-3")!.aborted).toEqual(true); + expect(started.length).toBeLessThanOrEqual(6); }); - test("bounds active reads and skips cancelled queued and remaining candidates", async () => { + test("refills four workers while retaining candidate order after out-of-order completions", async () => { + const { scheduler, started, reads } = controlledReads(); + const listing = scheduler(["0", "1", "2", "3", "4", "5"], new AbortController().signal); + await vi.waitFor(() => expect(started).toEqual(["0", "1", "2", "3"])); + reads.get("3")!.resolve("descriptor-3"); + await vi.waitFor(() => expect(started).toEqual(["0", "1", "2", "3", "4"])); + reads.get("2")!.resolve("descriptor-2"); + await vi.waitFor(() => expect(started).toEqual(["0", "1", "2", "3", "4", "5"])); + reads.get("5")!.resolve("descriptor-5"); + reads.get("4")!.resolve("descriptor-4"); + reads.get("1")!.resolve("descriptor-1"); + reads.get("0")!.resolve("descriptor-0"); + await expect(listing).resolves.toEqual([ + "descriptor-0", + "descriptor-1", + "descriptor-2", + "descriptor-3", + "descriptor-4", + "descriptor-5", + ]); + }); + + test("an unexpected read failure cancels sibling workers without starting remaining candidates", async () => { + const { scheduler, started, reads, signals } = controlledReads(); + const listing = scheduler(["0", "1", "2", "3", "4", "5"], new AbortController().signal); + const rejected = expect(listing).rejects.toThrow("Unexpected read failure"); + await vi.waitFor(() => expect(started).toEqual(["0", "1", "2", "3"])); + reads.get("0")!.reject(new TypeError("Unexpected read failure")); + await rejected; + expect(signals.get("1")!.aborted).toEqual(true); + expect(signals.get("2")!.aborted).toEqual(true); + expect(signals.get("3")!.aborted).toEqual(true); + expect(started).toEqual(["0", "1", "2", "3"]); + }); + + test("bounds global reads and skips cancelled queued candidates", async () => { const { scheduler, started, reads } = controlledReads(); const controllers = Array.from({ length: 5 }, () => new AbortController()); const listings = controllers.map((controller, index) => - scheduler([`${index}-first`, `${index}-next`], controller.signal), + scheduler([`${index}-first`], controller.signal), ); // Attach handlers before aborting any promises. function errorMessage(error: Error): string { diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 86d82cba31..72a67b9631 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -1665,7 +1665,23 @@ export class ClaudeAgentClient implements AgentClient { const cached = this.importDescriptorCache.get(candidate.cacheKey); if (cached) return structuredClone(cached); - const descriptor = await parseClaudeSessionDescriptor(candidate.path, candidate.mtime, signal); + let descriptor: ImportableProviderSession | null; + try { + descriptor = await parseClaudeSessionDescriptor(candidate.path, candidate.mtime, signal); + } catch (error) { + signal.throwIfAborted(); + const isFileError = + error instanceof Error && + "syscall" in error && + "code" in error && + typeof error.code === "string"; + if (!(error instanceof ClaudeImportRecordTooLargeError) && !isFileError) throw error; + this.logger.warn( + { err: error, path: candidate.path }, + "Skipping Claude transcript with unreadable import metadata", + ); + return null; + } if (descriptor) { const current = await fsPromises.stat(candidate.path).catch(() => null); signal.throwIfAborted(); @@ -6338,7 +6354,7 @@ class ClaudeImportRecordTooLargeError extends Error { readonly maxRecordBytes: number, ) { super( - `Cannot list Claude sessions: transcript ${filePath} record ${recordNumber} exceeds the ${maxRecordBytes}-byte import limit. Shorten or remove this record before retrying.`, + `Claude transcript ${filePath} record ${recordNumber} exceeds the ${maxRecordBytes}-byte import limit. Import metadata for this transcript was not loaded.`, ); this.name = "ClaudeImportRecordTooLargeError"; } @@ -6398,12 +6414,6 @@ async function parseClaudeSessionDescriptor( } } if (recordBytes > 0) applyRecord(); - } catch (error) { - signal.throwIfAborted(); - if (error instanceof ClaudeImportRecordTooLargeError) throw error; - // Transcripts can disappear between candidate discovery and opening them. - if (error instanceof Error && "code" in error && error.code === "ENOENT") return null; - throw error; } finally { stream.destroy(); } diff --git a/packages/server/src/server/agent/providers/claude/import-descriptor-scheduler.ts b/packages/server/src/server/agent/providers/claude/import-descriptor-scheduler.ts index eccc1a842d..41766e2d9b 100644 --- a/packages/server/src/server/agent/providers/claude/import-descriptor-scheduler.ts +++ b/packages/server/src/server/agent/providers/claude/import-descriptor-scheduler.ts @@ -54,25 +54,37 @@ export function createImportDescriptorScheduler< return async (candidates, signal) => { signal.throwIfAborted(); + const controller = new AbortController(); + signal = AbortSignal.any([signal, controller.signal]); let onAbort: (() => void) | undefined; const aborted = new Promise((_resolve, reject) => { onAbort = () => reject(signal.reason); signal.addEventListener("abort", onAbort, { once: true }); }); - try { - const results: Descriptor[] = []; - for (const candidate of candidates) { + const results: Descriptor[] = []; + let nextIndex = 0; + async function worker(): Promise { + while (nextIndex < candidates.length) { signal.throwIfAborted(); - // One outstanding read per listing gives later requests a turn in the - // shared FIFO instead of putting hundreds of candidates ahead of them. + const index = nextIndex++; + const candidate = candidates[index]!; + // Each worker queues only its next read. Later listings enter the + // shared FIFO after at most four outstanding reads from this listing. const pending = acquire(candidate); try { - results.push(await Promise.race([pending.promise, aborted])); + results[index] = await Promise.race([pending.promise, aborted]); } finally { release(candidate, pending); } } + } + try { + const workers = Array.from({ length: Math.min(4, candidates.length) }, () => worker()); + await Promise.all(workers); return results; + } catch (error) { + controller.abort(error); + throw error; } finally { if (onAbort) signal.removeEventListener("abort", onAbort); } From afd338d65eba5ad10545f68094e34c86a7981501 Mon Sep 17 00:00:00 2001 From: Josh Lane Date: Tue, 6 Oct 2026 10:01:06 -0400 Subject: [PATCH 4/4] fix(claude): surface session import read failures Skip expected candidate read failures and return healthy sessions. Distinguish unreadable metadata from parsed transcripts, and raise AggregateError when every candidate is unreadable. Propagate systemic file errors and narrow discovery error suppression. Detected-By: user --- .../agent/providers/claude/agent.test.ts | 269 +++++++++++++++++- .../server/agent/providers/claude/agent.ts | 94 ++++-- 2 files changed, 340 insertions(+), 23 deletions(-) diff --git a/packages/server/src/server/agent/providers/claude/agent.test.ts b/packages/server/src/server/agent/providers/claude/agent.test.ts index 435c53d0f6..29c00b1ed7 100644 --- a/packages/server/src/server/agent/providers/claude/agent.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.test.ts @@ -1,5 +1,6 @@ import { execFile } from "node:child_process"; import * as fs from "node:fs/promises"; +import nativeFs from "node:fs"; import * as os from "node:os"; import * as path from "node:path"; import { promisify } from "node:util"; @@ -22,6 +23,7 @@ import { createImportDescriptorScheduler } from "./import-descriptor-scheduler.j import { claudeProjectDirSync } from "./project-dir.js"; import { streamSession } from "../test-utils/session-stream-adapter.js"; import type { + ImportableProviderSession, AgentPromptInput, AgentSession, AgentTimelineItem, @@ -1670,6 +1672,261 @@ describe("normalizeClaudeAskUserQuestionUpdatedInput", () => { }); describe("ClaudeAgentClient.listImportableSessions", () => { + async function descriptorFixture(count: number) { + const configDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-claude-import-errors-")); + const projectDir = path.join(configDir, "projects", "errors"); + await fs.mkdir(projectDir, { recursive: true }); + const files: string[] = []; + const descriptors: ImportableProviderSession[] = []; + const logs: unknown[] = []; + for (let index = 0; index < count; index += 1) { + const file = path.join(projectDir, `${index}.jsonl`); + const timestamp = new Date(Date.UTC(2026, 0, 1) + (count - index) * 1000); + await fs.writeFile( + file, + JSON.stringify({ + type: "user", + sessionId: `session-${index}`, + cwd: "/errors", + message: { content: `Prompt ${index}` }, + }), + ); + await fs.utimes(file, timestamp, timestamp); + files.push(file); + descriptors.push({ + providerHandleId: `session-${index}`, + cwd: "/errors", + title: `Prompt ${index}`, + firstPromptPreview: `Prompt ${index}`, + lastPromptPreview: `Prompt ${index}`, + lastActivityAt: timestamp, + }); + } + const options = { + logger: pino({ level: "warn" }, { write: (line: string) => logs.push(JSON.parse(line)) }), + runtimeSettings: { env: { CLAUDE_CONFIG_DIR: configDir } }, + }; + return { + files, + descriptors, + logs, + options, + cleanup: () => fs.rm(configDir, { recursive: true, force: true }), + }; + } + + function readError(code: string, filePath: string): Error { + return Object.assign(new Error(`${code}: cannot read ${filePath}`), { + code, + path: filePath, + syscall: "open", + }); + } + + test.each(["ENOENT", "ENOTDIR", "EPERM"])( + "isolates %s from one transcript while returning a healthy session", + async (code) => { + const fixture = await descriptorFixture(2); + const originalReadStream = nativeFs.createReadStream; + const error = readError(code, fixture.files[0]!); + const spy = vi.spyOn(nativeFs, "createReadStream").mockImplementation((file, options) => { + if (file === fixture.files[0]) throw error; + return originalReadStream(file, options); + }); + try { + const client = new ClaudeAgentClient(fixture.options); + await expect(client.listImportableSessions({ limit: 2 })).resolves.toEqual([ + fixture.descriptors[1], + ]); + expect(fixture.logs).toEqual([ + expect.objectContaining({ + path: fixture.files[0], + err: expect.objectContaining({ code }), + msg: "Skipping Claude transcript with unreadable import metadata", + }), + ]); + } finally { + spy.mockRestore(); + await fixture.cleanup(); + } + }, + ); + + const cannotEnforceFilePermissions = process.platform === "win32" || process.getuid?.() === 0; + const discoveryPermissionCases = [ + { label: "project enumeration", mode: 0, deniedSuffix: "", existingSuffix: "" }, + { + label: "candidate stat", + mode: 0o400, + deniedSuffix: "hidden.jsonl", + existingSuffix: "0.jsonl", + }, + ]; + test.skipIf(cannotEnforceFilePermissions).each(discoveryPermissionCases)( + "logs denied $label while discovering a healthy project", + async ({ mode, deniedSuffix }) => { + const fixture = await descriptorFixture(1); + const projectsRoot = path.dirname(path.dirname(fixture.files[0]!)); + const deniedProject = path.join(projectsRoot, "denied-project"); + await fs.mkdir(deniedProject); + try { + await fs.writeFile(path.join(deniedProject, "hidden.jsonl"), '{"type":"user"}'); + await fs.chmod(deniedProject, mode); + const client = new ClaudeAgentClient(fixture.options); + await expect(client.listImportableSessions({ limit: 2 })).resolves.toEqual( + fixture.descriptors, + ); + expect(fixture.logs).toEqual([ + expect.objectContaining({ + path: path.join(deniedProject, deniedSuffix), + err: expect.objectContaining({ code: "EACCES" }), + msg: "Skipping Claude session discovery path with denied access", + }), + ]); + } finally { + await fs.chmod(deniedProject, 0o700); + await fixture.cleanup(); + } + }, + ); + + test.skipIf(cannotEnforceFilePermissions).each(discoveryPermissionCases)( + "surfaces a provider error when $label denies discovery in every project", + async ({ mode, deniedSuffix, existingSuffix }) => { + const fixture = await descriptorFixture(1); + const existingProject = path.dirname(fixture.files[0]!); + const deniedProject = path.join(path.dirname(existingProject), "denied-project"); + await fs.mkdir(deniedProject); + try { + await fs.writeFile(path.join(deniedProject, "hidden.jsonl"), '{"type":"user"}'); + await fs.chmod(existingProject, mode); + await fs.chmod(deniedProject, mode); + const client = new ClaudeAgentClient(fixture.options); + await expect(client.listImportableSessions({ limit: 2 })).rejects.toMatchObject({ + name: "AggregateError", + message: expect.stringContaining("no transcripts could be discovered"), + errors: expect.arrayContaining([ + expect.objectContaining({ + code: "EACCES", + path: path.join(existingProject, existingSuffix), + }), + expect.objectContaining({ + code: "EACCES", + path: path.join(deniedProject, deniedSuffix), + }), + ]), + }); + expect(fixture.logs).toHaveLength(2); + expect(fixture.logs).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + path: path.join(existingProject, existingSuffix), + err: expect.objectContaining({ code: "EACCES" }), + }), + expect.objectContaining({ + path: path.join(deniedProject, deniedSuffix), + err: expect.objectContaining({ code: "EACCES" }), + }), + ]), + ); + } finally { + await fs.chmod(existingProject, 0o700); + await fs.chmod(deniedProject, 0o700); + await fixture.cleanup(); + } + }, + ); + + test.skipIf(cannotEnforceFilePermissions)( + "isolates an actual access-denied transcript while returning a healthy session", + async () => { + const fixture = await descriptorFixture(2); + try { + await fs.chmod(fixture.files[0]!, 0); + const client = new ClaudeAgentClient(fixture.options); + await expect(client.listImportableSessions({ limit: 2 })).resolves.toEqual([ + fixture.descriptors[1], + ]); + expect(fixture.logs).toEqual([ + expect.objectContaining({ + path: fixture.files[0], + err: expect.objectContaining({ code: "EACCES" }), + msg: "Skipping Claude transcript with unreadable import metadata", + }), + ]); + } finally { + await fixture.cleanup(); + } + }, + ); + + test.skipIf(cannotEnforceFilePermissions)( + "surfaces a provider error when every actual candidate is access denied", + async () => { + const fixture = await descriptorFixture(2); + try { + await Promise.all(fixture.files.map((file) => fs.chmod(file, 0))); + const client = new ClaudeAgentClient(fixture.options); + await expect(client.listImportableSessions({ limit: 2 })).rejects.toMatchObject({ + name: "AggregateError", + errors: fixture.files.map((file) => + expect.objectContaining({ code: "EACCES", path: file }), + ), + message: expect.stringContaining( + "all 2 candidate transcripts have unreadable import metadata", + ), + }); + expect(fixture.logs).toHaveLength(2); + } finally { + await fixture.cleanup(); + } + }, + ); + + test("returns an empty listing for readable transcripts without import metadata", async () => { + const fixture = await descriptorFixture(2); + try { + await Promise.all(fixture.files.map((file) => fs.writeFile(file, '{"type":"assistant"}'))); + const client = new ClaudeAgentClient(fixture.options); + await expect(client.listImportableSessions({ limit: 2 })).resolves.toEqual([]); + expect(fixture.logs).toEqual([]); + } finally { + await fixture.cleanup(); + } + }); + + test.each(["EMFILE", "ENFILE", "EIO", "ERR_INVALID_ARG_TYPE"])( + "propagates %s and cancels sibling streams instead of skipping transcripts", + async (code) => { + const fixture = await descriptorFixture(6); + const started: string[] = []; + const streams: nativeFs.ReadStream[] = []; + const originalReadStream = nativeFs.createReadStream; + const error = readError(code, fixture.files[0]!); + const spy = vi.spyOn(nativeFs, "createReadStream").mockImplementation((file, options) => { + started.push(String(file)); + if (file === fixture.files[0]) throw error; + const stream = originalReadStream(file, options); + streams.push(stream); + return stream; + }); + try { + const client = new ClaudeAgentClient(fixture.options); + await expect(client.listImportableSessions({ limit: 6 })).rejects.toBe(error); + expect(started).toEqual(fixture.files.slice(0, 4)); + expect(streams.map((stream) => stream.destroyed)).toEqual([true, true, true]); + function closedStates(): boolean[] { + return streams.map((stream) => stream.closed); + } + await vi.waitFor(() => expect(closedStates()).toEqual([true, true, true])); + expect(fixture.logs).toEqual([]); + } finally { + spy.mockRestore(); + await fixture.cleanup(); + } + }, + ); + test("refreshes descriptors after append, rewrite, replacement, and deletion", async () => { const configDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-claude-import-cache-")); try { @@ -1816,7 +2073,7 @@ describe("ClaudeAgentClient.listImportableSessions", () => { } }); - test("skips a single record larger than the heap without allocating the full line", async () => { + test("reports an unreadable single record larger than the heap without allocating the full line", async () => { const configDir = await fs.mkdtemp( path.join(os.tmpdir(), "paseo-claude-import-single-record-"), ); @@ -1842,20 +2099,22 @@ describe("ClaudeAgentClient.listImportableSessions", () => { "--input-type=module", "--eval", ` - import { deepStrictEqual } from "node:assert"; + import { rejects } from "node:assert"; import { ClaudeAgentClient } from ${JSON.stringify(new URL("./agent.ts", import.meta.url).href)}; import pino from ${JSON.stringify(import.meta.resolve("pino"))}; const client = new ClaudeAgentClient({ logger: pino({ level: "silent" }), runtimeSettings: { env: { CLAUDE_CONFIG_DIR: ${JSON.stringify(configDir)} } }, }); - deepStrictEqual(await client.listImportableSessions({ limit: 1 }), []); - console.log("Oversized transcript skipped"); + await rejects(client.listImportableSessions({ limit: 1 }), { + name: "AggregateError", message: /all 1 candidate transcripts have unreadable import metadata/, + }); + console.log("Oversized transcript reported"); `, ], { timeout: 20_000 }, ); - expect(stdout.trim()).toEqual("Oversized transcript skipped"); + expect(stdout.trim()).toEqual("Oversized transcript reported"); } finally { await fs.rm(configDir, { recursive: true, force: true }); } diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 72a67b9631..84495ab65e 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -414,6 +414,10 @@ interface ClaudeAgentClientOptions { rewindSdk?: ClaudeRewindSdk; } +type ClaudeImportDescriptorResult = + | { type: "parsed"; descriptor: ImportableProviderSession | null } + | { type: "unreadable"; error: Error }; + interface ClaudeAgentSessionOptions { defaults?: { agents?: Record }; runtimeSettings?: ProviderRuntimeSettings; @@ -1649,9 +1653,20 @@ export class ClaudeAgentClient implements AgentClient { const scanLimit = Math.min(options?.scanLimit ?? limit * 3, 500); const candidates = await collectRecentClaudeSessions(sessionsRoot, scanLimit, { rootIsProjectDir: Boolean(options?.cwd), + logger: this.logger, }); - const parsed = await this.scheduleImportDescriptors(candidates, signal); + const results = await this.scheduleImportDescriptors(candidates, signal); + const failures = results.filter((result) => result.type === "unreadable"); + if (failures.length > 0 && failures.length === results.length) { + const errors = failures.map((result) => result.error); + throw new AggregateError( + errors, + `Cannot list Claude sessions: all ${failures.length} candidate transcripts have unreadable import metadata. ${errors[0]!.message}`, + ); + } + const parsed = results.filter((result) => result.type === "parsed"); return parsed + .map((result) => result.descriptor) .filter((session): session is ImportableProviderSession => session !== null) .slice(0, limit) .map((session) => structuredClone(session)); @@ -1660,36 +1675,36 @@ export class ClaudeAgentClient implements AgentClient { private async readImportDescriptor( candidate: ClaudeSessionCandidate, signal: AbortSignal, - ): Promise { + ): Promise { signal.throwIfAborted(); const cached = this.importDescriptorCache.get(candidate.cacheKey); - if (cached) return structuredClone(cached); + if (cached) return { type: "parsed", descriptor: structuredClone(cached) }; let descriptor: ImportableProviderSession | null; try { descriptor = await parseClaudeSessionDescriptor(candidate.path, candidate.mtime, signal); } catch (error) { signal.throwIfAborted(); - const isFileError = - error instanceof Error && - "syscall" in error && - "code" in error && - typeof error.code === "string"; - if (!(error instanceof ClaudeImportRecordTooLargeError) && !isFileError) throw error; + const isExpectedError = + error instanceof ClaudeImportRecordTooLargeError || isExpectedClaudeImportFileError(error); + if (!(error instanceof Error) || !isExpectedError) throw error; this.logger.warn( { err: error, path: candidate.path }, "Skipping Claude transcript with unreadable import metadata", ); - return null; + return { type: "unreadable", error }; } if (descriptor) { - const current = await fsPromises.stat(candidate.path).catch(() => null); + const current = await fsPromises.stat(candidate.path).catch((error: unknown) => { + if (!isExpectedClaudeImportFileError(error)) throw error; + return null; + }); signal.throwIfAborted(); if (current && claudeSessionCacheKey(candidate.path, current) === candidate.cacheKey) { this.importDescriptorCache.set(candidate.cacheKey, structuredClone(descriptor)); } } - return descriptor ? structuredClone(descriptor) : null; + return { type: "parsed", descriptor: descriptor ? structuredClone(descriptor) : null }; } async importSession(input: ImportProviderSessionInput, context: ImportProviderSessionContext) { @@ -6222,27 +6237,62 @@ function claudeSessionCacheKey(filePath: string, stats: fs.Stats): string { return `${filePath}\0${stats.dev}:${stats.ino}:${stats.size}:${stats.mtimeMs}:${stats.ctimeMs}`; } +function isMissingClaudeImportFile(error: unknown): boolean { + return ( + error instanceof Error && + "code" in error && + (error.code === "ENOENT" || error.code === "ENOTDIR") + ); +} + +function isExpectedClaudeImportFileError(error: unknown): boolean { + if (!(error instanceof Error) || !("code" in error)) return false; + return ( + error.code === "ENOENT" || + error.code === "ENOTDIR" || + error.code === "EACCES" || + error.code === "EPERM" + ); +} + async function pathExists(target: string): Promise { try { await fsPromises.access(target); return true; - } catch { + } catch (error) { + if (!isMissingClaudeImportFile(error)) throw error; return false; } } +interface ClaudeSessionDiscoveryOptions { + rootIsProjectDir: boolean; + logger: Logger; +} + async function collectRecentClaudeSessions( root: string, limit: number, - options?: { rootIsProjectDir?: boolean }, + options: ClaudeSessionDiscoveryOptions, ): Promise { + const permissionErrors: Error[] = []; + function recordDiscoveryFailure(error: unknown, targetPath: string): void { + if (isMissingClaudeImportFile(error)) return; + if (!(error instanceof Error) || !isExpectedClaudeImportFileError(error)) throw error; + permissionErrors.push(error); + options.logger.warn( + { err: error, path: targetPath }, + "Skipping Claude session discovery path with denied access", + ); + } let rootEntries: string[]; try { rootEntries = await fsPromises.readdir(root); - } catch { + } catch (error) { + if (!isMissingClaudeImportFile(error)) throw error; return []; } - const fileEntries = options?.rootIsProjectDir + const fileEntries = options.rootIsProjectDir ? rootEntries.filter((file) => file.endsWith(".jsonl")).map((file) => path.join(root, file)) : ( await Promise.all( @@ -6255,7 +6305,8 @@ async function collectRecentClaudeSessions( return files .filter((file) => file.endsWith(".jsonl")) .map((file) => path.join(projectPath, file)); - } catch { + } catch (error) { + recordDiscoveryFailure(error, projectPath); return [] as string[]; } }), @@ -6271,7 +6322,8 @@ async function collectRecentClaudeSessions( mtime: fileStats.mtime, cacheKey: claudeSessionCacheKey(fullPath, fileStats), }; - } catch { + } catch (error) { + recordDiscoveryFailure(error, fullPath); return null; } }), @@ -6279,6 +6331,12 @@ async function collectRecentClaudeSessions( const candidates: ClaudeSessionCandidate[] = statResults.filter( (entry): entry is ClaudeSessionCandidate => entry !== null, ); + if (candidates.length === 0 && permissionErrors.length > 0) { + throw new AggregateError( + permissionErrors, + `Cannot discover Claude sessions: access was denied to ${permissionErrors.length} paths and no transcripts could be discovered. ${permissionErrors[0]!.message}`, + ); + } return candidates.sort((a, b) => b.mtime.getTime() - a.mtime.getTime()).slice(0, limit); }