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;