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

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<string, AbortSignal>();
+    const sharedScheduler = createImportDescriptorScheduler(
+      (candidate: { cacheKey: string }, signal) => {
+        started.push(candidate.cacheKey);
+        signals.set(candidate.cacheKey, signal);
+        return new Promise<string>((resolve, reject) => {
+          reads.set(candidate.cacheKey, { resolve, reject });
+          signal.addEventListener("abort", () => reject(signal.reason), { once: true });
+        });
+      },
+    );
+    function scheduler(candidates: string[], signal: AbortSignal): Promise<string[]> {
+      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"]);
+  });
+});

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<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 scheduleImportDescriptors = createImportDescriptorScheduler(
+    (candidate: ClaudeSessionCandidate, signal) => this.readImportDescriptor(candidate, signal),
+  );
   private readonly importDescriptorCache = new LRUCache<string, ImportableProviderSession>({
     max: 500,
     maxSize: 8 * 1024 * 1024,
@@ -1636,52 +1632,47 @@ export class ClaudeAgentClient implements AgentClient {
   }
 
   async listImportableSessions(
-    options?: ListImportableSessionsOptions,
+    options?: ListImportableSessionsOptions & { signal?: AbortSignal },
   ): Promise<ImportableProviderSession[]> {
+    // 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<ImportableProviderSession | null> {
+    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<ImportableProviderSession | null> {
   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();
   }
 

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<Descriptor> {
+  controller: AbortController;
+  subscribers: number;
+  settled: boolean;
+  promise: Promise<Descriptor>;
+}
+
+export function createImportDescriptorScheduler<
+  Candidate extends ImportDescriptorCandidate,
+  Descriptor,
+>(
+  read: (candidate: Candidate, signal: AbortSignal) => Promise<Descriptor>,
+): (candidates: Candidate[], signal: AbortSignal) => Promise<Descriptor[]> {
+  const limit = pLimit(4);
+  const reads = new Map<string, PendingDescriptor<Descriptor>>();
+
+  function acquire(candidate: Candidate): PendingDescriptor<Descriptor> {
+    let pending = reads.get(candidate.cacheKey);
+    if (!pending) {
+      const controller = new AbortController();
+      const entry: PendingDescriptor<Descriptor> = {
+        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<Descriptor>): 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<never>((_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);
+    }
+  };
+}