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
2026-10-23 09:42:53 +00:00 · 2026-10-06 10:01:06 -04:00
parent 568aaeaf93
commit afd338d65e
2 changed files with 340 additions and 23 deletions
@@ -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 });
    }
@@ -414,6 +414,10 @@ interface ClaudeAgentClientOptions {
  rewindSdk?: ClaudeRewindSdk;
 }

+type ClaudeImportDescriptorResult =
+  | { type: "parsed"; descriptor: ImportableProviderSession | null }
+  | { type: "unreadable"; error: Error };
+
 interface ClaudeAgentSessionOptions {
  defaults?: { agents?: Record<string, AgentDefinition> };
  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<ImportableProviderSession | null> {
+  ): Promise<ClaudeImportDescriptorResult> {
    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<boolean> {
  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<ClaudeSessionCandidate[]> {
+  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);
 }