From 0b36e14fb2a8fda32167456e21875081e41dbebf Mon Sep 17 00:00:00 2001 From: dev-driver Date: Fri, 6 Nov 2026 09:36:13 +0000 Subject: [PATCH] fix(cli): arm rig exec --wait's Ctrl-C before announcing the run rig exec run --wait installed its own Ctrl-C listener only when it began polling, after printing that the run had started. A Ctrl-C landing in between met the default action and killed the CLI with no exit code, instead of stopping the wait with exit 3 as documented; the detached run itself was never affected. A test that sends the Ctrl-C on that printed line failed this way once under CI load. The wait is now armed before the CLI says anything about the run, on all three paths (a fresh start, a run the row already has, and a caller that lost the claim), and for a fresh start right after the connector is spawned, so a Ctrl-C from then on only stops the wait. A test seam stalls the CLI right after it announces the run, and the new case sends the Ctrl-C into that stall; it failed the same way before this change. --- packages/cli/src/commands/exec.ts | 57 ++++++++++++------- packages/cli/test/exec.test.ts | 12 +++- packages/cli/test/fixtures/exec-cli-runner.ts | 5 +- 3 files changed, 50 insertions(+), 24 deletions(-) diff --git a/packages/cli/src/commands/exec.ts b/packages/cli/src/commands/exec.ts index 167606ca98..0af0845c69 100644 --- a/packages/cli/src/commands/exec.ts +++ b/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 => { - 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 => { + 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 diff --git a/packages/cli/test/exec.test.ts b/packages/cli/test/exec.test.ts index 612ed21c45..0fc6421ae0 100644 --- a/packages/cli/test/exec.test.ts +++ b/packages/cli/test/exec.test.ts @@ -44,7 +44,7 @@ function setup(body = "scenario:reply PONG", env: Record = {}) { ...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; @@ -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"]; diff --git a/packages/cli/test/fixtures/exec-cli-runner.ts b/packages/cli/test/fixtures/exec-cli-runner.ts index ff0b0d173e..3579c45543 100644 --- a/packages/cli/test/fixtures/exec-cli-runner.ts +++ b/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= 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({