fix(claude): stream and cache session import descriptors

2026-10-23 09:42:53 +00:00 · 2026-10-02 14:10:20 -07:00
parent d831c7bf33
commit 9b61a6779d
2 changed files with 212 additions and 17 deletions
@@ -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;
@@ -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<string>;
  private readonly resolveVersion: (signal?: AbortSignal) => Promise<string>;
  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<ImportableProviderSession | null>
+  >();
+  private readonly importDescriptorCache = new LRUCache<string, ImportableProviderSession>({
+    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<ImportableProviderSession | null> {
+    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<T>(): AsyncMessageInput<T> {
 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<boolean> {
@@ -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<ImportableProviderSession | null> {
-  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;