fix: release host terminals after usage scans

.github/workflows/ci.yml

@@ -11,7 +11,10 @@ permissions:
 
 jobs:
   check-test-build:
-    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        os: [ubuntu-latest, macos-latest]
+    runs-on: ${{ matrix.os }}
     timeout-minutes: 15
     steps:
       - name: Check out repository

lib/opencode-go.test.ts

@@ -73,11 +73,10 @@ printf '%s' "$FAKE_HTTP"
 
   let path = `${binDir}:${process.env.PATH ?? ""}`;
   if (nodeOnly) {
-    for (const [name, target] of [
-      ["cat", "/usr/bin/cat"],
-      ["mktemp", "/usr/bin/mktemp"],
-      ["rm", "/usr/bin/rm"],
-    ] as const) {
+    for (const name of ["cat", "mktemp", "rm"] as const) {
+      const resolved = spawnSync("/bin/sh", ["-c", `command -v ${name}`], { encoding: "utf8" });
+      const target = resolved.stdout.trim();
+      if (resolved.status !== 0 || !target) throw new Error(`could not locate ${name} on PATH`);
       symlinkSync(target, join(binDir, name));
     }
     const nodePath = join(binDir, "node");

package.json

@@ -1,6 +1,6 @@
 {
   "name": "bb-plugin-usage",
-  "version": "0.3.6",
+  "version": "0.3.7",
   "type": "module",
   "scripts": {
     "build": "bb plugin build",

server.test.ts

@@ -9,7 +9,7 @@ vi.mock("@bb/plugin-sdk", () => ({
 
 import plugin, {
   dashboardRecordsSql, extractOpenCodeJson, jsonAgentRoots, loadProviderLimits, loadStoredOpenCodeGoLimits,
-  openCodeCommand, openCodeSql, runHostCommand, syncOpenCode, syncOpenCodeGo,
+  openCodeCommand, openCodeSql, runHostCommand, runWithConcurrency, syncOpenCode, syncOpenCodeGo,
 } from "./server";
 
 function localDay(ts: number): string {
@@ -97,7 +97,7 @@ describe("sync RPC", () => {
   it("actually dispatches an Antigravity scan through syncAll, not just through direct scan() calls", async () => {
     // Regression test for the exact gap flagged in review on
     // https://github.com/MayankBansal12/bb-plugin-usage/pull/21: AGENTS and
-    // jsonAgentRoots knew about "antigravity", but syncAll()'s Promise.all
+    // jsonAgentRoots knew about "antigravity", but syncAll()'s collector list
     // never called syncJsonAgent(..., "antigravity", ...), so no scan ever
     // ran for it in production even though the unit tests (which call
     // scan()/parseHostUsageAggregates directly) all passed. This drives the
@@ -106,7 +106,7 @@ describe("sync RPC", () => {
     const db = new Database(":memory:");
     let handlers: { sync: () => unknown } | undefined;
 
-    // The command is a shell wrapper around `node -e eval(gunzip(base64(...)))`
+    // The command runs `node -e eval(gunzip(base64(...)))` through the host shell
     // where the gzipped payload is the generated collector script with
     // agentId/roots baked in as a literal object — decode it the same way
     // to tell which JSON-agent sync this particular terminal is for.
@@ -130,6 +130,7 @@ describe("sync RPC", () => {
     }
 
     const commandsByTerminalId = new Map<string, string>();
+    const runningTerminalIds = new Set<string>();
 
     const bb = {
       settings: { define: vi.fn(() => ({ get: async () => ({ piSessionRoots: "", primeSessionRoots: "" }) })) },
@@ -153,7 +154,11 @@ describe("sync RPC", () => {
             commandsByTerminalId.set(id, input.start.command);
             return { id, status: "starting" };
           }),
-          get: vi.fn(async (args: { terminalId: string }) => ({ id: args.terminalId, status: "running" })),
+          get: vi.fn(async (args: { terminalId: string }) => {
+            if (runningTerminalIds.has(args.terminalId)) return { id: args.terminalId, status: "exited", exitCode: 0 };
+            runningTerminalIds.add(args.terminalId);
+            return { id: args.terminalId, status: "running", exitCode: null };
+          }),
           output: vi.fn(async (args: { terminalId: string }) => {
             const command = commandsByTerminalId.get(args.terminalId) ?? "";
             const agentId = agentIdFromCommand(command);
@@ -171,6 +176,7 @@ describe("sync RPC", () => {
               : fakeHostScanOutput(agentId ?? "codex", []); // every other agent: empty, uninteresting scan
             return { chunks: [{ seq: 1, dataBase64: Buffer.from(text).toString("base64") }], truncated: false };
           }),
+          input: vi.fn(async () => undefined),
           close: vi.fn(async () => undefined),
         },
       },
@@ -360,16 +366,19 @@ describe("provider limit loading", () => {
 });
 
 describe("host command output", () => {
-  it("collects output while the terminal is still running, then closes it", async () => {
+  it("collects output while running, releases the bounded handshake, and closes cleanly", async () => {
     const text = "query result\n__BB_HOST_COMMAND_DONE__:0\n";
     const create = vi.fn(async (input: unknown) => ({ id: "terminal-1", status: "starting", input }));
-    const get = vi.fn(async () => ({ id: "terminal-1", status: "running" }));
+    const get = vi.fn()
+      .mockResolvedValueOnce({ id: "terminal-1", status: "running", exitCode: null })
+      .mockResolvedValueOnce({ id: "terminal-1", status: "exited", exitCode: 0 });
     const output = vi.fn(async () => ({
       chunks: [{ seq: 1, dataBase64: Buffer.from(text).toString("base64") }],
       truncated: false,
     }));
+    const input = vi.fn(async () => undefined);
     const close = vi.fn(async () => undefined);
-    const bb = { sdk: { terminals: { create, get, output, close } } } as unknown as BbPluginApi;
+    const bb = { sdk: { terminals: { create, get, output, input, close } } } as unknown as BbPluginApi;
 
     await expect(runHostCommand(
       bb,
@@ -379,22 +388,33 @@ describe("host command output", () => {
       { title: "Usage test", timeoutMs: 1_000, pollMs: 1 },
     )).resolves.toBe(text);
 
-    expect(get).toHaveBeenCalledOnce();
     expect(output).toHaveBeenCalledOnce();
-    expect(close).toHaveBeenCalledWith({ terminalId: "terminal-1", mode: "force" });
+    expect(input).toHaveBeenCalledWith({ terminalId: "terminal-1", dataBase64: "Cg==" });
+    expect(close).toHaveBeenCalledWith({ terminalId: "terminal-1", mode: "if-clean" });
     expect(create.mock.calls[0]?.[0]).toMatchObject({
       start: { mode: "command", command: expect.stringContaining("__BB_HOST_COMMAND_DONE__") },
     });
+    expect(create.mock.calls[0]?.[0]).toEqual(expect.objectContaining({
+      start: expect.objectContaining({ command: expect.stringContaining("read -r -t 30") }),
+    }));
+    expect(create.mock.calls[0]?.[0]).not.toEqual(expect.objectContaining({
+      start: expect.objectContaining({ command: expect.stringContaining("while :") }),
+    }));
   });
 
-  it("surfaces a command diagnostic before closing the held terminal", async () => {
-    const text = "__BB_USAGE_ERROR__:OpenCode query failed\n__BB_HOST_COMMAND_DONE__:1\n";
+  it("surfaces a command diagnostic after releasing a normal non-zero exit", async () => {
+    const text = "__BB_USAGE_ERROR__:OpenCode query failed\n__BB_HOST_COMMAND_DONE__:127\n";
     const close = vi.fn(async () => undefined);
+    const input = vi.fn(async () => undefined);
+    const get = vi.fn()
+      .mockResolvedValueOnce({ id: "terminal-1", status: "running", exitCode: null })
+      .mockResolvedValueOnce({ id: "terminal-1", status: "exited", exitCode: 127 });
     const bb = {
       sdk: { terminals: {
         create: vi.fn(async () => ({ id: "terminal-1", status: "starting" })),
-        get: vi.fn(async () => ({ id: "terminal-1", status: "running" })),
+        get,
         output: vi.fn(async () => ({ chunks: [{ seq: 1, dataBase64: Buffer.from(text).toString("base64") }], truncated: false })),
+        input,
         close,
       } },
     } as unknown as BbPluginApi;
@@ -406,16 +426,21 @@ describe("host command output", () => {
       new AbortController().signal,
       { title: "Usage test", timeoutMs: 1_000, pollMs: 1 },
     )).rejects.toThrow("OpenCode query failed");
-    expect(close).toHaveBeenCalledOnce();
+    expect(input).toHaveBeenCalledOnce();
+    expect(close).toHaveBeenCalledWith({ terminalId: "terminal-1", mode: "if-clean" });
   });
 
   it("surfaces bounded terminal output when a command has no structured diagnostic", async () => {
     const text = "CLI compatibility error\n__BB_HOST_COMMAND_DONE__:1\n";
+    const get = vi.fn()
+      .mockResolvedValueOnce({ id: "terminal-1", status: "running", exitCode: null })
+      .mockResolvedValueOnce({ id: "terminal-1", status: "exited", exitCode: 1 });
     const bb = {
       sdk: { terminals: {
         create: vi.fn(async () => ({ id: "terminal-1", status: "starting" })),
-        get: vi.fn(async () => ({ id: "terminal-1", status: "running" })),
+        get,
         output: vi.fn(async () => ({ chunks: [{ seq: 1, dataBase64: Buffer.from(text).toString("base64") }], truncated: false })),
+        input: vi.fn(async () => undefined),
         close: vi.fn(async () => undefined),
       } },
     } as unknown as BbPluginApi;
@@ -429,13 +454,14 @@ describe("host command output", () => {
     )).rejects.toThrow("CLI compatibility error");
   });
 
-  it("times out and closes a stalled machine terminal", async () => {
+  it("times out and force-closes a stalled machine terminal", async () => {
     const close = vi.fn(async () => undefined);
     const bb = {
       sdk: { terminals: {
         create: vi.fn(async () => ({ id: "terminal-1", status: "starting" })),
-        get: vi.fn(async () => ({ id: "terminal-1", status: "running" })),
+        get: vi.fn(async () => ({ id: "terminal-1", status: "running", exitCode: null })),
         output: vi.fn(async () => ({ chunks: [], truncated: false })),
+        input: vi.fn(async () => undefined),
         close,
       } },
     } as unknown as BbPluginApi;
@@ -449,6 +475,50 @@ describe("host command output", () => {
     )).rejects.toThrow("timed out");
     expect(close).toHaveBeenCalledWith({ terminalId: "terminal-1", mode: "force" });
   });
+
+  it("logs terminal cleanup failures", async () => {
+    const warn = vi.fn();
+    const text = "__BB_HOST_COMMAND_DONE__:0\n";
+    const get = vi.fn()
+      .mockResolvedValueOnce({ id: "terminal-1", status: "running", exitCode: null })
+      .mockResolvedValueOnce({ id: "terminal-1", status: "exited", exitCode: 0 });
+    const bb = {
+      sdk: { terminals: {
+        create: vi.fn(async () => ({ id: "terminal-1", status: "starting" })),
+        get,
+        output: vi.fn(async () => ({ chunks: [{ seq: 1, dataBase64: Buffer.from(text).toString("base64") }], truncated: false })),
+        input: vi.fn(async () => undefined),
+        close: vi.fn(async () => { throw new Error("close failed"); }),
+      } },
+      log: { warn },
+    } as unknown as BbPluginApi;
+
+    await expect(runHostCommand(
+      bb,
+      { id: "host-1", name: "Machine" },
+      "true",
+      new AbortController().signal,
+      { title: "Usage test", timeoutMs: 1_000, pollMs: 1 },
+    )).resolves.toBe(text);
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining("terminal cleanup failed: close failed"));
+  });
+});
+
+describe("collector concurrency", () => {
+  it("runs no more than the configured number of tasks at once", async () => {
+    let active = 0;
+    let maximumActive = 0;
+    const tasks = Array.from({ length: 9 }, (_, index) => async () => {
+      active += 1;
+      maximumActive = Math.max(maximumActive, active);
+      await Promise.resolve();
+      active -= 1;
+      return index;
+    });
+
+    await expect(runWithConcurrency(tasks, 3)).resolves.toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8]);
+    expect(maximumActive).toBe(3);
+  });
 });
 
 describe("OpenCode query", () => {
@@ -532,7 +602,7 @@ describe("OpenCode query", () => {
       db as unknown as ReturnType<BbPluginApi["storage"]["database"]>,
       { id: "host-1", name: "Machine" },
       new AbortController().signal,
-      async () => "__BB_USAGE_BEGIN__\n[{}]\n__BB_USAGE_END__:0\n__BB_HOST_COMMAND_DONE__:0\n",
+      async () => "__BB_USAGE_BEGIN__\n[{}]\n__BB_USAGE_END__:0\n",
     )).resolves.toBeUndefined();
 
     expect(db.prepare("SELECT COUNT(*) count FROM usage_events").get()).toEqual({ count: 1 });

server.ts

@@ -427,15 +427,32 @@ function delay(ms: number) {
 
 type HostCommandOptions = { title: string; timeoutMs: number; pollMs?: number };
 
-function heldHostCommand(command: string) {
-  return `( ${command} ); bb_usage_status=$?; printf '\\n%s:%s\\n' '__BB_HOST_COMMAND_DONE__' "$bb_usage_status"; while :; do sleep 3600; done`;
+const HOST_COMMAND_DONE = "__BB_HOST_COMMAND_DONE__";
+const HOST_COMMAND_RELEASE_TIMEOUT_SECONDS = 30;
+
+function releasableHostCommand(command: string) {
+  return `( ${command} ); bb_usage_status=$?; printf '\\n%s:%s\\n' '${HOST_COMMAND_DONE}' "$bb_usage_status"; read -r -t ${HOST_COMMAND_RELEASE_TIMEOUT_SECONDS} bb_usage_release || true; exit "$bb_usage_status"`;
 }
 
 function terminalOutputText(output: Awaited<ReturnType<BbPluginApi["sdk"]["terminals"]["output"]>>) {
   return output.chunks.sort((a, b) => a.seq - b.seq)
     .map((chunk) => Buffer.from(chunk.dataBase64, "base64").toString("utf8")).join("");
 }
 
+export async function runWithConcurrency<T>(tasks: readonly (() => Promise<T>)[], limit: number): Promise<T[]> {
+  if (!Number.isInteger(limit) || limit < 1) throw new Error("Concurrency limit must be a positive integer.");
+  const results = new Array<T>(tasks.length);
+  let nextTask = 0;
+  async function worker() {
+    while (nextTask < tasks.length) {
+      const index = nextTask++;
+      results[index] = await tasks[index]!();
+    }
+  }
+  await Promise.all(Array.from({ length: Math.min(limit, tasks.length) }, () => worker()));
+  return results;
+}
+
 export async function runHostCommand(
   bb: BbPluginApi,
   machine: Machine,
@@ -448,8 +465,10 @@ export async function runHostCommand(
     cols: 120,
     rows: 24,
     title: options.title,
-    start: { mode: "command", command: heldHostCommand(command) },
+    start: { mode: "command", command: releasableHostCommand(command) },
   });
+  let exited = false;
+  let completion: { text: string; exitCode: number } | null = null;
   try {
     const deadline = Date.now() + options.timeoutMs;
     while (Date.now() < deadline) {
@@ -463,24 +482,34 @@ export async function runHostCommand(
         });
         if (output.truncated) throw new Error(`${options.title} exceeded the 900 KB output limit.`);
         const text = terminalOutputText(output);
-        const completion = text.match(/__BB_HOST_COMMAND_DONE__:(\d+)/);
-        if (completion) {
-          const exitCode = Number(completion[1]);
-          if (exitCode !== 0) {
-            const diagnostic = text.match(/__BB_USAGE_ERROR__:(.+)/)?.[1]?.trim()
-              ?? text.replace(/__BB_HOST_COMMAND_DONE__:\d+/g, "").trim().slice(-300);
-            throw new Error(diagnostic || `${options.title} exited with code ${exitCode}.`);
-          }
-          return text;
+        const match = text.match(new RegExp(`${HOST_COMMAND_DONE}:(\\d+)`));
+        if (match && !completion) {
+          completion = { text, exitCode: Number(match[1]) };
+          await bb.sdk.terminals.input({
+            terminalId: terminal.id,
+            dataBase64: Buffer.from("\n").toString("base64"),
+          }).catch((error) => {
+            bb.log.debug(`${options.title} terminal release signal failed: ${errorMessage(error)}`);
+          });
+        }
+      } else if (state.status === "exited") {
+        exited = true;
+        if (!completion) throw new Error(`${options.title} stopped before its output could be collected.`);
+        const { text, exitCode } = completion;
+        if (exitCode !== 0) {
+          const diagnostic = text.match(/__BB_USAGE_ERROR__:(.+)/)?.[1]?.trim()
+            ?? text.replace(new RegExp(`${HOST_COMMAND_DONE}:\\d+`, "g"), "").trim().slice(-300);
+          throw new Error(diagnostic || `${options.title} exited with code ${exitCode}.`);
         }
-      } else if (state.status !== "starting" && state.status !== "disconnected") {
-        throw new Error(`${options.title} stopped before its output could be collected.`);
+        return text;
       }
       await delay(options.pollMs ?? 200);
     }
     throw new Error(`${options.title} timed out after ${Math.ceil(options.timeoutMs / 1000)} seconds.`);
   } finally {
-    await bb.sdk.terminals.close({ terminalId: terminal.id, mode: "force" }).catch(() => { /* already closed by the host */ });
+    await bb.sdk.terminals.close({ terminalId: terminal.id, mode: exited ? "if-clean" : "force" }).catch((error) => {
+      bb.log.warn(`${options.title} terminal cleanup failed: ${errorMessage(error)}`);
+    });
   }
 }
 
@@ -784,17 +813,17 @@ export default async function plugin(bb: BbPluginApi) {
           for (const agent of AGENTS) upsertState(db, machine.id, agent.id, "unavailable", countForMachine(db, machine.id, agent.id), message, false);
           continue;
         }
-        await Promise.all([
-          syncJsonAgent(bb, db, machine, home, "codex", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
-          syncJsonAgent(bb, db, machine, home, "claude", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
-          syncJsonAgent(bb, db, machine, home, "fx", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
-          syncJsonAgent(bb, db, machine, home, "grok", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
-          syncJsonAgent(bb, db, machine, home, "pi", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
-          syncJsonAgent(bb, db, machine, home, "prime", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
-          syncJsonAgent(bb, db, machine, home, "antigravity", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
-          syncOpenCode(bb, db, machine, timeoutSignal(OPENCODE_SYNC_TIMEOUT_MS, serviceSignal)),
-          syncOpenCodeGo(bb, db, machine, timeoutSignal(OPENCODE_GO_SYNC_TIMEOUT_MS, serviceSignal)),
-        ]);
+        await runWithConcurrency([
+          () => syncJsonAgent(bb, db, machine, home, "codex", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
+          () => syncJsonAgent(bb, db, machine, home, "claude", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
+          () => syncJsonAgent(bb, db, machine, home, "fx", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
+          () => syncJsonAgent(bb, db, machine, home, "grok", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
+          () => syncJsonAgent(bb, db, machine, home, "pi", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
+          () => syncJsonAgent(bb, db, machine, home, "prime", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
+          () => syncJsonAgent(bb, db, machine, home, "antigravity", collectorSettings, timeoutSignal(JSON_AGENT_SYNC_TIMEOUT_MS, serviceSignal)),
+          () => syncOpenCode(bb, db, machine, timeoutSignal(OPENCODE_SYNC_TIMEOUT_MS, serviceSignal)),
+          () => syncOpenCodeGo(bb, db, machine, timeoutSignal(OPENCODE_GO_SYNC_TIMEOUT_MS, serviceSignal)),
+        ], 3);
       }
       return new Date().toISOString();
     });