fix(cli): arm rig exec --wait's Ctrl-C before announcing the run (#2601)

rig exec run --wait installed its SIGINT listener only when polling began, after it had printed that the run had started, so a Ctrl-C landing between the two met the default action and killed the CLI with no exit code (the run itself, in its own session, was never affected). armWait now installs the listener before each path announces the run: a run the row already has, a caller that lost the claim to another, and a fresh start right after the connector is spawned, where the start-wait loop also stops when the waiter leaves. report removes the listener in a finally, and waitForEnd takes the waiter instead of arming its own. A tests-only afterAnnounce seam stalls the CLI just after the announcement; the new cell sends a Ctrl-C to the whole process group into that stall and expects exit 3, the stopped-waiting line and the run completing. Limits: a Ctrl-C before the connector is spawned still meets the default; an exception before report would skip the listener cleanup, though the CLI exits on it; the pre-announcement path is traced at source, not separately exercised; a run that has already ended still exits 0 or 2 by its outcome.

packages/cli/src/commands/exec.ts

@@ -360,10 +360,13 @@ export interface ExecDeps {
   startWaitMs?: number;
   /** Tests only: acts between the row-claim read and its publish, as a concurrent caller would. */
   beforeClaim?: () => void;
+  /** Tests only: runs right after the CLI has told its caller about the run, as a process descheduled there would. */
+  afterAnnounce?: () => void;
 }
 
 const sleep = (ms: number) => new Promise((done) => setTimeout(done, ms));
 const exitFor = (run: ExecRun) => (run.outcome === "completed" ? 0 : 2);
+type Waiter = { left: boolean; leave: () => void };
 
 export function execCommand(deps: ExecDeps = {}): Command {
   const cmd = new Command("exec").description("Run a queue row's task on a named harness (Claude Code or Codex over ACP) in a connector that outlives the command, with the outcome kept on the row");
@@ -376,21 +379,25 @@ export function execCommand(deps: ExecDeps = {}): Command {
     ?? ((runId: string) => [process.execPath, fileURLToPath(new URL("../index.js", import.meta.url)), "exec", "connector", runId]);
   const pollMs = deps.pollMs ?? 1000;
 
-  /** Polls the run file. Leaving (Ctrl-C) stops only the wait; the connector is in its own session and keeps running. */
-  const waitForEnd = async (runId: string): Promise<ExecRun | "left"> => {
-    let left = false;
-    const leave = () => { left = true; };
-    process.on("SIGINT", leave);
-    try {
-      for (let polls = 1; ; polls++) {
-        const run = readRun(runId);
-        if (!run) throw new Error(`the record of ${runId} is gone`);
-        if (run.endedAt) return run;
-        if (left) return "left";
-        if (polls % 5 === 0) { const checked = await reconcileRun(run, queueFor); if (checked.endedAt) return checked; }
-        await sleep(pollMs);
-      }
-    } finally { process.removeListener("SIGINT", leave); }
+  /** The wait's own Ctrl-C, armed before the CLI tells its caller anything about the run, so a Ctrl-C from then on
+   *  only stops the wait: armed later, one landing just after the announcement met the default and killed the CLI. The
+   *  connector is in its own session either way. */
+  const armWait = (wait: boolean | undefined): Waiter | undefined => {
+    if (!wait) return undefined;
+    const waiter: Waiter = { left: false, leave: () => { waiter.left = true; } };
+    process.on("SIGINT", waiter.leave);
+    return waiter;
+  };
+  /** Polls the run file until it ends, or until the waiter leaves. */
+  const waitForEnd = async (runId: string, waiter: Waiter): Promise<ExecRun | "left"> => {
+    for (let polls = 1; ; polls++) {
+      const run = readRun(runId);
+      if (!run) throw new Error(`the record of ${runId} is gone`);
+      if (run.endedAt) return run;
+      if (waiter.left) return "left";
+      if (polls % 5 === 0) { const checked = await reconcileRun(run, queueFor); if (checked.endedAt) return checked; }
+      await sleep(pollMs);
+    }
   };
   const finish = (result: ExecRun | "left", runId: string) => {
     if (result === "left") {
@@ -403,9 +410,11 @@ export function execCommand(deps: ExecDeps = {}): Command {
   };
   /** Every path that holds the row's run ends here: a run that has ended exits by its outcome (0 completed, 2 any
    *  other), --wait waits for one still going (3 if the waiter leaves), and without --wait a run still going exits 0. */
-  const report = async (run: ExecRun, wait: boolean | undefined) => {
-    if (run.endedAt) process.exitCode = exitFor(run);
-    else if (wait) finish(await waitForEnd(run.runId), run.runId);
+  const report = async (run: ExecRun, waiter: Waiter | undefined) => {
+    try {
+      if (run.endedAt) process.exitCode = exitFor(run);
+      else if (waiter) finish(await waitForEnd(run.runId, waiter), run.runId);
+    } finally { if (waiter) process.removeListener("SIGINT", waiter.leave); }
   };
 
   cmd
@@ -478,8 +487,9 @@ Guide: rig context get reference/acp-agents.md`)
           return;
         }
         if (!opts.again) {
+          const waiter = armWait(opts.wait);
           console.log(`${qitemId} already has a run, so the task isn't sent twice:\n${describeRun(run)}\nOnce it has ended, --again starts another turn on this row.`);
-          await report(run, opts.wait);
+          await report(run, waiter);
           return;
         }
       }
@@ -502,8 +512,9 @@ Guide: rig context get reference/acp-agents.md`)
         rmSync(runPath(runId), { force: true });
         const winnerId = rowRun(qitemId);
         const winner = winnerId ? readRun(winnerId) : undefined;
+        const waiter = winner ? armWait(opts.wait) : undefined;
         console.log(`${qitemId} was started by another caller as ${winnerId ?? "another run"}, so the task isn't sent twice${winner ? `:\n${describeRun(winner)}` : ""}`);
-        if (winner) await report(winner, opts.wait);
+        if (winner) await report(winner, waiter);
         else process.exitCode = 1;
         return;
       }
@@ -518,10 +529,11 @@ Guide: rig context get reference/acp-agents.md`)
         child.once("exit", (code, signal) => { connectorExit = `exited (code ${code ?? "none"}, signal ${signal ?? "none"})`; });
         child.unref();
       } finally { closeSync(log); }
+      const waiter = armWait(opts.wait);
       let current: ExecRun = run;
       for (const deadline = Date.now() + (deps.startWaitMs ?? 10_000); Date.now() < deadline; await sleep(50)) {
         current = readRun(runId) ?? current;
-        if (current.state !== "requested" || spawnError || connectorExit) break;
+        if (current.state !== "requested" || spawnError || connectorExit || waiter?.left) break;
       }
       if (spawnError && current.state === "requested") {
         current = (await settleRun(current, { state: "finished", outcome: "failed", summary: `the connector couldn't be started: ${spawnError.message}`, endedAt: now() })).run;
@@ -530,7 +542,8 @@ Guide: rig context get reference/acp-agents.md`)
       }
       console.log(`[rig exec] ${runId} for ${qitemId} on ${harness.name}: ${current.endedAt ? `ended, outcome ${current.outcome}: ${current.summary}` : current.state === "started" ? "started" : "requested; its connector hasn't recorded itself yet"}`);
       console.log(`  rig exec status ${runId} shows it; rig exec cancel ${runId} cancels it.`);
-      await report(current, opts.wait);
+      deps.afterAnnounce?.();
+      await report(current, waiter);
     });
 
   cmd

packages/cli/test/exec.test.ts

@@ -44,7 +44,7 @@ function setup(body = "scenario:reply PONG", env: Record<string, string> = {}) {
     ...process.env, OPENRIG_HOME: home, OPENRIG_SESSION_NAME: SEAT, EXEC_QUEUE_FILE: queueFile,
     PATH: [join(home, "node_modules", ".bin"), dirname(process.execPath), process.env.PATH].join(delimiter), ...env,
   };
-  for (const name of ["FAKE_ACP_LOAD", "FAKE_ACP_MISSING", "FAKE_ACP_COMPANION_PID_FILE", "FAKE_ACP_COST", "FAKE_ACP_PROMPT_USAGE", "FAKE_ACP_IGNORE_EOF", "EXEC_TEST_CONNECTOR", "EXEC_TEST_WINNER", "EXEC_TEST_HOOK", "OPENRIG_EXEC_CONNECTOR"]) if (!(name in env)) delete runEnv[name];
+  for (const name of ["FAKE_ACP_LOAD", "FAKE_ACP_MISSING", "FAKE_ACP_COMPANION_PID_FILE", "FAKE_ACP_COST", "FAKE_ACP_PROMPT_USAGE", "FAKE_ACP_IGNORE_EOF", "EXEC_TEST_CONNECTOR", "EXEC_TEST_WINNER", "EXEC_TEST_HOOK", "EXEC_TEST_STALL_MS", "OPENRIG_EXEC_CONNECTOR"]) if (!(name in env)) delete runEnv[name];
   return { home, env: runEnv, queueFile };
 }
 type Home = ReturnType<typeof setup>;
@@ -165,6 +165,16 @@ describe("rig exec", () => {
     expect(notes(h).join("\n")).not.toContain("cancel requested");
   }, 40_000);
 
+  it("a Ctrl-C that lands just after the run is announced, while the CLI is stalled there, still stops only the wait", async () => {
+    const h = withEnv(setup("scenario:slow"), { EXEC_TEST_STALL_MS: "1500" });
+    const waiter = start(h, ["run", "qitem-task", "--harness", "codex", "--wait"], { group: true });
+    await until(() => (/for qitem-task on codex: started/.test(waiter.out()) ? true : undefined));
+    process.kill(-waiter.child.pid!, "SIGINT"); // arrives during the stall, before the CLI's next step
+    expect(await waiter.exited).toBe(3);
+    expect(waiter.out()).toMatch(/stopped waiting; exec-\S+ keeps running/);
+    expect((await ended(h, startedId(waiter.out())!)).outcome).toBe("completed");
+  }, 40_000);
+
   it("starts a row's task once: a second start or a retry returns the same run, and --again starts another after it ends", async () => {
     const h = setup();
     const args = ["run", "qitem-task", "--harness", "claude"];

packages/cli/test/fixtures/exec-cli-runner.ts

@@ -1,7 +1,8 @@
 // One `rig exec …` invocation in its own process, which then exits as the CLI does, over the test's file queue.
 // EXEC_TEST_CONNECTOR=die starts a connector that exits before recording itself; =instant one that records a completed
 // outcome before the CLI's first look. EXEC_TEST_WINNER=<outcome> has another caller claim the row, with a run that
-// ended with that outcome, between this caller's claim read and its publish.
+// ended with that outcome, between this caller's claim read and its publish. EXEC_TEST_STALL_MS stalls the CLI right
+// after it announces the run, as a process descheduled there would be.
 import { writeFileSync } from "node:fs";
 import { join } from "node:path";
 import { fileURLToPath } from "node:url";
@@ -13,12 +14,14 @@ const instant = fileURLToPath(new URL("./exec-instant-connector.ts", import.meta
 const queue = fileQueue(process.env.EXEC_QUEUE_FILE!);
 const connector = process.env.EXEC_TEST_CONNECTOR;
 const winner = process.env.EXEC_TEST_WINNER;
+const stall = Number(process.env.EXEC_TEST_STALL_MS ?? 0);
 await execCommand({
   queue: async () => queue,
   connectorCommand: (runId) => connector === "die" ? [process.execPath, "-e", "process.exit(3)"]
     : [process.execPath, "--import", "tsx", connector === "instant" ? instant : entry, runId],
   pollMs: 100,
   startWaitMs: 20_000,
+  afterAnnounce: stall ? () => { for (const until = Date.now() + stall; Date.now() < until;) { /* busy: no timer or signal callback runs */ } } : undefined,
   beforeClaim: winner ? () => {
     const runs = join(process.env.OPENRIG_HOME!, "acp", "runs");
     writeFileSync(join(runs, "exec-winner.json"), JSON.stringify({