test(providers): reproduce plugin migration, steering, and chunk failures

* fix(server): keep overlapping directory searches within their scan budget

Construct child paths and branch iterators as the round-robin walk consumes them. Eager sibling expansion retained hundreds of thousands of unvisited children per search and exhausted the heap when requests overlapped.

* test(server): stop directory-search children before fixture cleanup

---------

Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>

packages/plugin/src/server/acp.test.ts

@@ -1246,6 +1246,49 @@ describe("runAcpProvider streamed chunks without messageId (#4699)", () => {
     messageId?: string;
   }
 
+  it("keeps reasoning separate from a reply sharing its native message ID", async () => {
+    const harness = connectorHarness({
+      promptUpdates: () => [
+        textChunk("private reasoning", "agent_thought_chunk", "shared-id"),
+        textChunk("public reply", "agent_message_chunk", "shared-id"),
+      ],
+    });
+    const registration = runAcpProvider({
+      id: "chunk-acp",
+      label: "Chunk ACP",
+      connector: harness.connector,
+    });
+    const connection = await registration.connect({
+      versions: [1],
+      capabilities: ["prompt.message"],
+    });
+    const events: ProviderEvent[] = [];
+    connection.onEvent((event) => events.push(event));
+    try {
+      await connection.send(openInput());
+      await waitForEvent(events, (event) => event.type === "session.ready");
+      await connection.send({
+        type: "session.prompt",
+        sessionId: "session-1",
+        prompt: {
+          clientMessageId: "same-id",
+          delivery: "auto",
+          input: { type: "message", content: [{ type: "text", text: "hello" }] },
+        },
+      });
+      await waitForEvent(
+        events,
+        (event) => event.type === "session.turn" && event.state === "completed",
+      );
+      const reply = events.findLast(
+        (event) => event.type === "timeline.item" && event.item.type === "assistant_message",
+      );
+      expect(reply).toMatchObject({ item: { text: "public reply" } });
+    } finally {
+      await connection.close();
+    }
+  });
+
   it("continues one timeline item per turn instead of one per chunk", async () => {
     const chunksByPrompt: Record<string, Chunk[]> = {
       first: [

packages/server/src/server/agent/plugin-provider.test.ts

@@ -287,6 +287,45 @@ function expectNestedChildren(events: AgentStreamEvent[]) {
 }
 
 describe("PluginAgentClientRegistry", () => {
+  test("returns unavailable when steering a provider without prompt.steer capability", async () => {
+    const harness = createProviderHarness();
+    const registry = new PluginAgentClientRegistry(createTestLogger());
+    registry.replace([harness.registration]);
+    const session = await registry.clients()[harness.registration.id]!.createSession({
+      provider: harness.registration.id,
+      cwd: "/tmp",
+    });
+    try {
+      await expect(
+        session.steerActiveTurn!("follow up", { expectedTurnId: "turn-1" }),
+      ).resolves.toEqual({ status: "unavailable" });
+      expect(harness.inputs.filter((input) => input.type === "session.prompt")).toHaveLength(0);
+    } finally {
+      await session.close();
+      registry.replace([]);
+    }
+  });
+
+  test("resumes a native ACP handle after the same provider ID moves to a plugin", async () => {
+    const harness = createProviderHarness();
+    const registry = new PluginAgentClientRegistry(createTestLogger());
+    registry.replace([harness.registration]);
+    try {
+      const session = await registry.clients()[harness.registration.id]!.resumeSession(
+        {
+          provider: harness.registration.id,
+          sessionId: "native-session-id",
+          nativeHandle: "native-session-id",
+          metadata: { provider: "acp", cwd: "/tmp" },
+        },
+        { provider: harness.registration.id, cwd: "/tmp" },
+      );
+      await session.close();
+    } finally {
+      registry.replace([]);
+    }
+  });
+
   test("stores only agent options while the plugin receives merged defaults", async () => {
     const logger = createTestLogger();
     const harness = createProviderHarness();

packages/server/src/utils/directory-suggestions.test.ts

@@ -1,5 +1,5 @@
 import { createHook } from "node:async_hooks";
-import { execFileSync } from "node:child_process";
+import { execFileSync, spawn, type ChildProcess } from "node:child_process";
 import {
   mkdtempSync,
   mkdirSync,
@@ -1066,6 +1066,16 @@ describe("relative typed-entry configuration", () => {
   });
 });
 
+interface SearchProcessExit {
+  code: number | null;
+  signal: string | null;
+}
+
+interface SearchProcess {
+  child: ChildProcess;
+  exited: Promise<SearchProcessExit>;
+}
+
 // Reading a tree is cheap; re-deriving each entry's path state is not. A scan that asks whether
 // every ancestor of every entry is inside the root, or is Git-ignored, costs multiples of the
 // read and is what made large home directories exceed the client request timeout.
@@ -1073,6 +1083,7 @@ describe("home-tree scan cost", () => {
   const DEPTH = 8;
   const CHAINS = 40;
   let scanRoot: string;
+  let searchProcess: SearchProcess | null = null;
 
   beforeEach(() => {
     scanRoot = realpathSync.native(mkdtempSync(path.join(tmpdir(), "directory-scan-cost-")));
@@ -1083,8 +1094,16 @@ describe("home-tree scan cost", () => {
     }
   });
 
-  afterEach(() => {
-    rmSync(scanRoot, { recursive: true, force: true });
+  afterEach(async () => {
+    try {
+      if (searchProcess) {
+        searchProcess.child.kill();
+        await searchProcess.exited;
+      }
+    } finally {
+      searchProcess = null;
+      rmSync(scanRoot, { recursive: true, force: true });
+    }
   });
 
   // Nothing matches, so the scan walks the whole tree instead of stopping at a confident result.
@@ -1102,6 +1121,78 @@ describe("home-tree scan cost", () => {
     return stopPathContainmentMetrics();
   }
 
+  it("finishes overlapping broad-tree searches within a bounded heap", async () => {
+    const broadRoot = path.join(scanRoot, "broad");
+    mkdirSync(broadRoot);
+    for (let branch = 0; branch < 30; branch += 1) {
+      const branchRoot = path.join(broadRoot, `branch-${branch}`);
+      mkdirSync(branchRoot);
+      for (let group = 0; group < 30; group += 1) {
+        const groupRoot = path.join(branchRoot, `group-${group}`);
+        mkdirSync(groupRoot);
+        for (let leaf = 0; leaf < 30; leaf += 1) {
+          mkdirSync(path.join(groupRoot, `leaf-${leaf}`));
+        }
+      }
+    }
+    // A smaller real tree and explicit heap budget keep the million-directory repro cheap in CI.
+    const moduleUrl = new URL("./directory-suggestions.ts", import.meta.url).href;
+    const source = `
+      const { searchDirectoryEntries } = await import(${JSON.stringify(moduleUrl)});
+      const results = await Promise.all(Array.from({ length: 14 }, (_, index) =>
+        searchDirectoryEntries({
+          root: ${JSON.stringify(broadRoot)},
+          query: "z".repeat(index + 1),
+          pathFormat: "absolute",
+          includeFiles: false,
+          includeDirectories: true,
+          pathQueryPolicy: "rooted",
+          rootAliases: ["~"],
+          blankQueryBehavior: "none",
+          confidentResultScanThreshold: 5000,
+          maxEntriesScanned: 2000,
+          limit: 30,
+        })
+      ));
+      process.stdout.write(JSON.stringify(results));
+    `;
+    const child = spawn(
+      process.execPath,
+      [
+        "--max-old-space-size=128",
+        "--conditions=source",
+        "--import",
+        "tsx",
+        "--input-type=module",
+        "--eval",
+        source,
+      ],
+      { stdio: ["ignore", "pipe", "pipe"], timeout: 20_000 },
+    );
+    let stdout = "";
+    let stderr = "";
+    child.stdout.setEncoding("utf8");
+    child.stderr.setEncoding("utf8");
+    child.stdout.on("data", (chunk: string) => {
+      stdout += chunk;
+    });
+    child.stderr.on("data", (chunk: string) => {
+      stderr += chunk;
+    });
+    const exited = new Promise<SearchProcessExit>((resolve, reject) => {
+      child.on("error", reject);
+      child.on("close", (code, signal) => resolve({ code, signal }));
+    });
+    searchProcess = { child, exited };
+    const exit = await exited;
+    expect({ ...exit, outOfMemory: stderr.includes("heap out of memory") }).toEqual({
+      code: 0,
+      signal: null,
+      outOfMemory: false,
+    });
+    expect(JSON.parse(stdout)).toEqual(Array.from({ length: 14 }, () => []));
+  });
+
   it("derives no containment for a tree that cannot leave the root", async () => {
     const checks = await countContainmentChecks(scanWholeTree);
 

packages/server/src/utils/directory-suggestions.ts

@@ -63,6 +63,13 @@ interface WalkStart {
   resolvedPath: string;
 }
 
+interface WalkChildrenOptions {
+  parent: WalkStart;
+  depth: number;
+  input: SearchInput;
+  visited: Set<string>;
+}
+
 interface TraversedEntry extends ChildEntry {
   visiblePath: string;
   depth: number;
@@ -318,17 +325,18 @@ async function searchChildren(input: SearchInput): Promise<RankedEntry[]> {
   const discoveryInput = isInsideExcludedPath(parent, input)
     ? { ...input, excludedDiscoveryPaths: new Set<string>() }
     : input;
-  const entries = await readChildren(parent);
-  return entries.flatMap((entry) => {
+  const ranked: RankedEntry[] = [];
+  for await (const entry of readChildren(parent)) {
     if (!isPathInsideRoot(input.root, entry.resolvedPath) || !shouldDiscover(entry, discoveryInput))
-      return [];
+      continue;
     const candidate: TraversedEntry = {
       ...entry,
       visiblePath: path.join(visibleParent, entry.name),
       depth: 1,
     };
-    return shouldSuggest(candidate, input) ? [rank(candidate, input)] : [];
-  });
+    if (shouldSuggest(candidate, input)) ranked.push(rank(candidate, input));
+  }
+  return ranked;
 }
 
 async function searchTree(
@@ -336,21 +344,8 @@ async function searchTree(
   start: WalkStart = { visiblePath: input.root, resolvedPath: input.root },
 ): Promise<RankedEntry[]> {
   if (!(input.maxEntriesScanned > 0)) return [];
-  const roots = (await readChildren(start.resolvedPath)).filter((entry) =>
-    staysInsideRoot(entry, input.root),
-  );
   const visited = new Set<string>([start.resolvedPath]);
-  const branches = roots.flatMap((entry) =>
-    shouldDiscover(entry, input)
-      ? [
-          walkBranch(
-            { ...entry, visiblePath: path.join(start.visiblePath, entry.name), depth: 1 },
-            input,
-            visited,
-          ),
-        ]
-      : [],
-  );
+  const branches = walkChildren({ parent: start, depth: 1, input, visited });
   const ranked: RankedEntry[] = [];
   let scanned = 0;
   const threshold = input.confidentResultScanThreshold;
@@ -381,38 +376,39 @@ async function* walkBranch(
   )
     return;
   visited.add(entry.resolvedPath);
-  const children = (await readChildren(entry.resolvedPath)).filter((child) =>
-    staysInsideRoot(child, input.root),
-  );
-  const branches = children.flatMap((child) =>
-    shouldDiscover(child, input)
-      ? [
-          walkBranch(
-            {
-              ...child,
-              visiblePath: path.join(entry.visiblePath, child.name),
-              depth: entry.depth + 1,
-            },
-            input,
-            visited,
-          ),
-        ]
-      : [],
-  );
-  yield* roundRobin(branches);
+  yield* roundRobin(walkChildren({ parent: entry, depth: entry.depth + 1, input, visited }));
 }
 
-async function* roundRobin<T>(branches: Array<AsyncGenerator<T>>): AsyncGenerator<T> {
-  let active = branches;
-  while (active.length) {
+// Create a child's path and iterator only when its branch gets a turn. Eagerly constructing every
+// sibling's subtree retained hundreds of thousands of unvisited children beyond the scan budget.
+async function* walkChildren({
+  parent,
+  depth,
+  input,
+  visited,
+}: WalkChildrenOptions): AsyncGenerator<AsyncGenerator<TraversedEntry>> {
+  for await (const child of readChildren(parent.resolvedPath)) {
+    if (!staysInsideRoot(child, input.root) || !shouldDiscover(child, input)) continue;
+    yield walkBranch(
+      { ...child, visiblePath: path.join(parent.visiblePath, child.name), depth },
+      input,
+      visited,
+    );
+  }
+}
+
+async function* roundRobin<T>(branches: AsyncIterable<AsyncGenerator<T>>): AsyncGenerator<T> {
+  let active: AsyncIterable<AsyncGenerator<T>> | Array<AsyncGenerator<T>> = branches;
+  while (true) {
     const nextRound: Array<AsyncGenerator<T>> = [];
-    for (const branch of active) {
+    for await (const branch of active) {
       const next = await branch.next();
       if (!next.done) {
         nextRound.push(branch);
         yield next.value;
       }
     }
+    if (!nextRound.length) return;
     active = nextRound;
   }
 }
@@ -742,9 +738,9 @@ async function resolveDirectory(inputPath: string): Promise<string | null> {
   return info?.isDirectory() ? resolved : null;
 }
 
-async function readChildren(directory: string): Promise<ChildEntry[]> {
+async function* readChildren(directory: string): AsyncGenerator<ChildEntry> {
   const directoryInfo = await stat(directory).catch(() => null);
-  if (!directoryInfo?.isDirectory()) return [];
+  if (!directoryInfo?.isDirectory()) return;
 
   const cached = CAN_VALIDATE_DIRECTORY_CACHE_FROM_METADATA
     ? directoryListCache.get(directory)
@@ -774,9 +770,10 @@ async function readChildren(directory: string): Promise<ChildEntry[]> {
     }
   }
 
-  return (await Promise.all(rawEntries.map((entry) => resolveChild(directory, entry))))
-    .filter((entry): entry is ChildEntry => entry !== null)
-    .sort((left, right) => left.name.localeCompare(right.name));
+  for (const rawEntry of rawEntries) {
+    const entry = await resolveChild(directory, rawEntry);
+    if (entry) yield entry;
+  }
 }
 
 async function loadGitIgnoredPaths(root: string): Promise<Set<string>> {