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); + } + }; +}