From 4ef4a95a429ec2f76c6a1710e3e3ee88a90d1818 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Thu, 22 Oct 2026 12:07:39 +0200 Subject: [PATCH] Keep Claude session listing from loading whole histories into memory (#7626) * fix(claude): stream and cache session import descriptors * 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 * 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 * 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 * fix(claude): keep large sessions importable while streaming history Preserve import metadata for transcripts containing image records above 4 MiB. Stream complete records instead of rejecting otherwise listable sessions, and cover overlapping requests across aliased provider roots under a constrained heap. Co-authored-by: Josh Lane * test(app): wait for active agent controls and submitted turns Observe the mode control before testing Command Center keyboard selection. Observe a running turn before waiting for its completion, so the rewind test cannot mistake the previous idle state and end marker for the new turn. --------- Co-authored-by: Josh Lane Co-authored-by: Josh Lane Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> --- .../e2e/browser/agent-message-rewind.spec.ts | 2 + .../helpers/command-center-agent-controls.ts | 3 + .../claude/agent.import-large-record.test.ts | 53 ++ .../agent/providers/claude/agent.test.ts | 584 ++++++++++++++++++ .../server/agent/providers/claude/agent.ts | 181 +++++- .../claude/import-descriptor-scheduler.ts | 92 +++ 6 files changed, 890 insertions(+), 25 deletions(-) create mode 100644 packages/server/src/server/agent/providers/claude/agent.import-large-record.test.ts create mode 100644 packages/server/src/server/agent/providers/claude/import-descriptor-scheduler.ts diff --git a/packages/app/e2e/browser/agent-message-rewind.spec.ts b/packages/app/e2e/browser/agent-message-rewind.spec.ts index 4875cc31c1..8d4e773259 100644 --- a/packages/app/e2e/browser/agent-message-rewind.spec.ts +++ b/packages/app/e2e/browser/agent-message-rewind.spec.ts @@ -21,8 +21,10 @@ async function completeSubmittedTurn( const userMessage = page.getByTestId("user-message").filter({ hasText: prompt }); await expect(userMessage).toBeVisible(); + await expectAgentReadyToInterrupt(page); const finish = await agent.client.waitForFinish(agent.agentId, 30_000); expect(finish.status).toBe("idle"); + await expectAgentIdle(page); await expect(page.getByText("(end of synthetic stream)", { exact: true }).last()).toBeVisible(); await expect(userMessage).toHaveAttribute("aria-busy", "false"); return userMessage; diff --git a/packages/app/e2e/support/helpers/command-center-agent-controls.ts b/packages/app/e2e/support/helpers/command-center-agent-controls.ts index 6b51b84609..f819cc98e9 100644 --- a/packages/app/e2e/support/helpers/command-center-agent-controls.ts +++ b/packages/app/e2e/support/helpers/command-center-agent-controls.ts @@ -18,6 +18,9 @@ export async function openCommandCenterForAgent( workspace: Pick, ): Promise { await openAgentRoute(page, { workspaceId: workspace.workspaceId, agentId: workspace.agentId }); + await expect( + page.getByRole("button", { name: /^Select agent mode \(/ }).filter({ visible: true }), + ).toBeVisible({ timeout: 30_000 }); const panel = await openCommandCenter(page); return panel.getByTestId("command-center-input"); } diff --git a/packages/server/src/server/agent/providers/claude/agent.import-large-record.test.ts b/packages/server/src/server/agent/providers/claude/agent.import-large-record.test.ts new file mode 100644 index 0000000000..f0fb26ecc3 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/agent.import-large-record.test.ts @@ -0,0 +1,53 @@ +import { test, expect } from "vitest"; +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; +import pino from "pino"; +import { ClaudeAgentClient } from "./agent.js"; + +test("lists a session containing a large image record and retains the final prompt", async () => { + const configDir = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-import-image-")); + try { + const project = path.join(configDir, "projects", "image-project"); + await fs.mkdir(project, { recursive: true }); + const first = { + type: "user", + sessionId: "image-session", + cwd: "/image-project", + message: { content: "Describe the image" }, + }; + const image = { + type: "user", + sessionId: "image-session", + cwd: "/image-project", + message: { + content: [ + { + type: "image", + source: { type: "base64", media_type: "image/png", data: "a".repeat(5 * 1024 * 1024) }, + }, + ], + }, + }; + const last = { type: "user", message: { content: "What changed?" } }; + await fs.writeFile( + path.join(project, "image-session.jsonl"), + [first, image, last].map((record) => JSON.stringify(record)).join("\n"), + ); + const client = new ClaudeAgentClient({ + logger: pino({ level: "silent" }), + runtimeSettings: { env: { CLAUDE_CONFIG_DIR: configDir } }, + }); + const sessions = await client.listImportableSessions({ limit: 1 }); + expect(sessions).toEqual([ + expect.objectContaining({ + providerHandleId: "image-session", + title: "Describe the image", + firstPromptPreview: "Describe the image", + lastPromptPreview: "What changed?", + }), + ]); + } finally { + await fs.rm(configDir, { recursive: true, force: true }); + } +}); 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 d6b869f1e0..2b2c725269 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,9 @@ +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 pino from "pino"; import { afterEach, describe, expect, test, vi } from "vitest"; import type { PermissionResult, SDKMessage, SDKUserMessage } from "@anthropic-ai/claude-agent-sdk"; @@ -15,9 +18,11 @@ 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 { + ImportableProviderSession, AgentPromptInput, AgentSession, AgentTimelineItem, @@ -1740,6 +1745,404 @@ 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 }), + }; + } + + 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("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], + ], + ); + + 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), + ).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.each([ + { providers: 1, requestsPerProvider: 4 }, + { providers: 5, requestsPerProvider: 3 }, + ])( + "discovers large histories with $providers providers and $requestsPerProvider requests each", + async ({ providers, requestsPerProvider }) => { + 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"))}; + import { mkdir, symlink } from "node:fs/promises"; + import path from "node:path"; + const clients = []; + for (let index = 0; index < ${providers}; index++) { + const providerConfigDir = path.join(${JSON.stringify(configDir)}, "alias-" + index); + await mkdir(providerConfigDir); + await symlink(path.join(${JSON.stringify(configDir)}, "projects"), path.join(providerConfigDir, "projects"), "junction"); + clients.push(new ClaudeAgentClient({ + logger: pino({ level: "silent" }), + runtimeSettings: { env: { CLAUDE_CONFIG_DIR: providerConfigDir } }, + })); + } + const listings = await Promise.all(clients.flatMap(client => + Array.from({ length: ${requestsPerProvider} }, () => 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("keeps metadata and UTF-8 across read chunk boundaries", 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("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; @@ -3695,3 +4098,184 @@ 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"], 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("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}`), + 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"]), + ); + reads.get("small")!.resolve("small descriptor"); + await expect(small).resolves.toEqual(["small descriptor"]); + 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("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`], 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 3d59851038..681988a813 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -4,6 +4,8 @@ import { randomUUID } from "node:crypto"; import fs from "node:fs"; import { promises } from "node:fs"; import path from "node:path"; +import { LRUCache } from "lru-cache"; +import { createImportDescriptorScheduler } from "./import-descriptor-scheduler.js"; import { type AgentDefinition, type CanUseTool, @@ -467,6 +469,10 @@ interface ClaudeAgentClientOptions { rewindSdk?: ClaudeRewindSdk; } +type ClaudeImportDescriptorResult = + | { type: "parsed"; descriptor: ImportableProviderSession | null } + | { type: "unreadable"; error: Error }; + interface ClaudeAgentSessionOptions { defaults?: { agents?: Record }; runtimeSettings?: ProviderRuntimeSettings; @@ -1571,6 +1577,15 @@ export class ClaudeAgentClient implements AgentClient { private readonly resolveBinary: () => Promise; private readonly resolveVersion: (signal?: AbortSignal) => Promise; private readonly rewindSdk: ClaudeRewindSdk; + private readonly scheduleImportDescriptors = createImportDescriptorScheduler( + (candidate: ClaudeSessionCandidate, signal) => this.readImportDescriptor(candidate, signal), + ); + 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.discoverModels = options.discoverModels ?? discoverClaudeModels; @@ -1714,26 +1729,73 @@ 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), + logger: this.logger, }); - const parsed = await Promise.all( - candidates.map((candidate) => parseClaudeSessionDescriptor(candidate.path, candidate.mtime)), - ); + 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); + .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 { type: "parsed", descriptor: structuredClone(cached) }; + + let descriptor: ImportableProviderSession | null; + try { + descriptor = await parseClaudeSessionDescriptor(candidate.path, candidate.mtime, signal); + } catch (error) { + signal.throwIfAborted(); + if (!(error instanceof Error) || !isExpectedClaudeImportFileError(error)) throw error; + this.logger.warn( + { err: error, path: candidate.path }, + "Skipping Claude transcript with unreadable import metadata", + ); + return { type: "unreadable", error }; + } + if (descriptor) { + 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 { type: "parsed", descriptor: descriptor ? structuredClone(descriptor) : null }; } async importSession(input: ImportProviderSessionInput, context: ImportProviderSessionContext) { @@ -6540,29 +6602,69 @@ 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}`; +} + +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( @@ -6575,7 +6677,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[]; } }), @@ -6585,8 +6688,14 @@ async function collectRecentClaudeSessions( fileEntries.map(async (fullPath) => { try { const fileStats = await fsPromises.stat(fullPath); - return { path: fullPath, mtime: fileStats.mtime }; - } catch { + if (!fileStats.isFile()) return null; + return { + path: fullPath, + mtime: fileStats.mtime, + cacheKey: claudeSessionCacheKey(fullPath, fileStats), + }; + } catch (error) { + recordDiscoveryFailure(error, fullPath); return null; } }), @@ -6594,6 +6703,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); } @@ -6665,14 +6780,8 @@ function applyClaudeSessionEntryToAccumulator( async function parseClaudeSessionDescriptor( filePath: string, mtime: Date, + signal: AbortSignal, ): Promise { - let content: string; - try { - content = await fsPromises.readFile(filePath, "utf8"); - } catch { - return null; - } - const acc: ClaudeSessionDescriptorAccumulator = { sessionId: null, cwd: null, @@ -6683,16 +6792,38 @@ async function parseClaudeSessionDescriptor( lastPromptPreview: null, }; - for (const line of content.split(/\r?\n/)) { - if (!line || !shouldParseClaudeSessionDescriptorLine(line, acc)) continue; + // Keep only the current record, including image records larger than a read chunk. + const record: Buffer[] = []; + const stream = fs.createReadStream(filePath, { highWaterMark: 64 * 1024, signal }); + function applyRecord(): void { + const line = Buffer.concat(record).toString("utf8"); + if (!line || !shouldParseClaudeSessionDescriptorLine(line, acc)) return; let entry: unknown; try { entry = JSON.parse(line); } catch { - continue; + return; } 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; + record.push(chunk.subarray(start, end)); + if (newline === -1) break; + applyRecord(); + record.length = 0; + start = newline + 1; + } + } + if (record.length > 0) applyRecord(); + } finally { + stream.destroy(); + } const { sessionId, cwd } = acc; 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..41766e2d9b --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/import-descriptor-scheduler.ts @@ -0,0 +1,92 @@ +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(); + 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 }); + }); + const results: Descriptor[] = []; + let nextIndex = 0; + async function worker(): Promise { + while (nextIndex < candidates.length) { + signal.throwIfAborted(); + 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[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); + } + }; +}