From 62d0630d7c3e26752fb66b2b5a53630ef6f96dc2 Mon Sep 17 00:00:00 2001 From: Harsh Batheja Date: Sun, 17 May 2026 00:51:45 +0530 Subject: [PATCH] =?UTF-8?q?fix(pipeline):=20non-blocking=20command=20execu?= =?UTF-8?q?tor=20=E2=80=94=20release=20engine=20lock=20during=20child=20ru?= =?UTF-8?q?ntime?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Greptile P1: command stages held the engine's serialization lock for their entire execution (up to DEFAULT_COMMAND_TIMEOUT_MS = 10 min). Every other operation — cancelRun, tick, sibling stages, concurrent dispatch — queued behind lockTail for that whole window. The agent executor avoids this with a startStage/pollStage/cancelStage handoff; the command executor now matches. **Executor refactor** (`packages/core/src/pipeline/executors/command.ts`): - New `startStage(input)` spawns the child and returns a `RunningCommandStage` handle immediately. Synthetic failures (fork refusal / spawn error) produce a handle whose `outcome` is already populated and `done` already resolved. - New `pollStage(handle)` is synchronous: returns the cached outcome or `{ status: "running" }`. - New `cancelStage(handle)` is idempotent: clears timers, SIGTERMs the process tree, schedules SIGKILL after `COMMAND_KILL_GRACE_MS`, and settles the outcome synchronously so callers awaiting `done` unblock without waiting on `close`. - `CommandExecutorSpawnError` added (mirrors `AgentExecutorSpawnError`) for kind-mismatch — the engine catches and dispatches STAGE_FAILED. - `run()` is now a thin sugar: `startStage(input)` then `await done`. Test ergonomics unchanged. **Engine wiring** (`packages/core/src/pipeline/engine.ts`): - Added `commandInflight: Map` (parallel to the agent `inflight` map). - `runCommandStage` now awaits the fast `startStage`, then either dispatches an already-settled outcome (refusal / spawn fail) inline OR stores the handle in `commandInflight` and returns. **The lock releases immediately after spawn.** - `tick()` polls `commandInflight` after agent stages and dispatches STAGE_COMPLETED / STAGE_FAILED on terminal outcomes. - `CANCEL_STAGE` handler extended to check `commandInflight` and call `cancelStage`. **Tests**: - pipeline-command-executor: +6 tests for `startStage`/`pollStage`/ `cancelStage` directly, including idempotency, spawn-fast, sleep-30 cancellation, kind-mismatch throw, fork-refusal handle. - pipeline-engine-executors: regression test proving `cancelRun` completes in <500ms against two parallel `sleep 5` stages — would have timed out in the pre-refactor code. **Bonus** (round-3 P2 deferred from previous review): - Added regression test for the stdout-overflow path including stderr in the error message (`packages/core/src/pipeline/executors/command.ts` line 268). The earlier round-3 fix was correct but had no test coverage; now it does. All 1372 core tests pass. Typecheck clean. Zero lint errors. --- .../pipeline-command-executor.test.ts | 165 +++++- .../pipeline-engine-executors.test.ts | 100 +++- packages/core/src/index.ts | 2 + packages/core/src/pipeline/engine.ts | 124 +++-- .../core/src/pipeline/executors/command.ts | 510 ++++++++++++------ packages/core/src/pipeline/executors/index.ts | 2 + packages/core/src/pipeline/index.ts | 2 + 7 files changed, 686 insertions(+), 219 deletions(-) diff --git a/packages/core/src/__tests__/pipeline-command-executor.test.ts b/packages/core/src/__tests__/pipeline-command-executor.test.ts index 494acd54c..46c4d8d53 100644 --- a/packages/core/src/__tests__/pipeline-command-executor.test.ts +++ b/packages/core/src/__tests__/pipeline-command-executor.test.ts @@ -15,6 +15,7 @@ import { asRunId, asStageRunId, createCommandExecutor, + CommandExecutorSpawnError, DEFAULT_COMMAND_STDERR_CAP_BYTES, formatForkRefusalMessage, type CommandStartInput, @@ -46,18 +47,17 @@ function makeInput(overrides: Partial = {}): CommandStartInpu // Platform-neutral guards: never spawn a subprocess. describe("command executor — guards (cross-platform)", () => { - it("rejects non-command stages with a typed failure", async () => { + it("throws CommandExecutorSpawnError via run() for non-command stages", async () => { const exec = createCommandExecutor(); - const outcome = await exec.run( - makeInput({ - stage: makeCommandStage({ - executor: { kind: "agent", plugin: "claude-code", mode: "code" }, + await expect( + exec.run( + makeInput({ + stage: makeCommandStage({ + executor: { kind: "agent", plugin: "claude-code", mode: "code" }, + }), }), - }), - ); - expect(outcome.status).toBe("failed"); - if (outcome.status !== "failed") throw new Error("unreachable"); - expect(outcome.errorMessage).toContain("command executor cannot run"); + ), + ).rejects.toThrow(CommandExecutorSpawnError); }); it("refuses to run a fork PR when stage.allowFork is unset", async () => { @@ -305,6 +305,151 @@ posixDescribe("command executor — stderr cap", () => { // is well under 2x the cap (the raw cap plus the marker suffix). expect(outcome.errorMessage.length).toBeLessThan(DEFAULT_COMMAND_STDERR_CAP_BYTES * 2); }); + + it("includes stderr in the error message when stdout overflows the cap", async () => { + // Regression: round 3 fixed a bug where the stdout-overflow path dropped + // stderr entirely, leaving operators with no diagnostic signal. The error + // message should contain BOTH the "stdout exceeded" marker AND any stderr + // the command produced. + const exec = createCommandExecutor(); + // 4 MB + 1 KB of stdout (well over DEFAULT_COMMAND_STDOUT_CAP_BYTES), + // plus a short stderr hint. The shell pipeline below is GNU-coreutils- + // safe: `yes` is line-buffered, `head -c` is byte-bounded. + const stage = makeCommandStage({ + executor: { + kind: "command", + command: "/bin/sh", + args: [ + "-c", + 'yes x | head -c 4200000; echo "stderr-hint" >&2', + ], + }, + }); + + const outcome = await exec.run(makeInput({ stage })); + + expect(outcome.status).toBe("failed"); + if (outcome.status !== "failed") throw new Error("unreachable"); + expect(outcome.errorMessage).toContain("stdout exceeded"); + expect(outcome.errorMessage).toContain("stderr-hint"); + }); +}); + +// --------------------------------------------------------------------------- +// startStage / pollStage / cancelStage — new non-blocking API +// --------------------------------------------------------------------------- + +// Cross-platform: kind-mismatch throws without spawning a subprocess. +describe("command executor — startStage (cross-platform)", () => { + it("throws CommandExecutorSpawnError for non-command stages", async () => { + const exec = createCommandExecutor(); + await expect( + exec.startStage( + makeInput({ + stage: makeCommandStage({ + executor: { kind: "agent", plugin: "claude-code", mode: "code" }, + }), + }), + ), + ).rejects.toThrow(CommandExecutorSpawnError); + }); + + it("returns a handle with refused outcome for fork PRs (no subprocess)", async () => { + const onRefuse = vi.fn(); + const exec = createCommandExecutor({ onRefuse }); + const stage = makeCommandStage({ + executor: { kind: "command", command: "/bin/sh", args: ["-c", "echo should-not-run"] }, + }); + const handle = await exec.startStage(makeInput({ stage, isFromFork: true })); + + // Outcome is already populated (no subprocess was spawned). + expect(handle.outcome).not.toBeNull(); + expect(handle.outcome?.status).toBe("failed"); + if (handle.outcome?.status !== "failed") throw new Error("unreachable"); + expect(handle.outcome.refused).toBe(true); + expect(handle.outcome.errorMessage).toBe(formatForkRefusalMessage(stage.name)); + + // done resolves immediately. + const resolved = await handle.done; + expect(resolved.status).toBe("failed"); + + // onRefuse was called. + expect(onRefuse).toHaveBeenCalledTimes(1); + }); +}); + +posixDescribe("command executor — startStage (POSIX subprocess)", () => { + it("startStage returns a handle without waiting for the child to exit", async () => { + const exec = createCommandExecutor(); + const stage = makeCommandStage({ + executor: { kind: "command", command: "/bin/sh", args: ["-c", "sleep 1"] }, + }); + + const start = Date.now(); + const handle = await exec.startStage(makeInput({ stage })); + const elapsed = Date.now() - start; + + // Must return well before the child exits (target: <100ms on any CI host). + expect(elapsed).toBeLessThan(100); + // While the child is alive, outcome is null and pollStage says "running". + expect(handle.outcome).toBeNull(); + expect(exec.pollStage(handle).status).toBe("running"); + + // Clean up: cancel so the sleep doesn't run for a full second. + await exec.cancelStage(handle); + }); + + it("pollStage returns running while the child is alive, then the terminal outcome once it exits", async () => { + const exec = createCommandExecutor(); + const stage = makeCommandStage({ + executor: { kind: "command", command: "/bin/sh", args: ["-c", "true"] }, + }); + + const handle = await exec.startStage(makeInput({ stage })); + + // Await the child to finish. + const outcome = await handle.done; + + // pollStage now returns the same terminal outcome. + expect(exec.pollStage(handle).status).toBe("completed"); + expect(outcome.status).toBe("completed"); + expect(handle.outcome).not.toBeNull(); + expect(handle.outcome?.status).toBe("completed"); + }); + + it("cancelStage kills the child and resolves done with a cancelled failure", async () => { + const exec = createCommandExecutor(); + const stage = makeCommandStage({ + executor: { kind: "command", command: "/bin/sh", args: ["-c", "sleep 30"] }, + }); + + const handle = await exec.startStage(makeInput({ stage })); + expect(handle.outcome).toBeNull(); + + const cancelStart = Date.now(); + await exec.cancelStage(handle); + const cancelElapsed = Date.now() - cancelStart; + + // Cancel must resolve quickly (well before the 30s sleep). + expect(cancelElapsed).toBeLessThan(3_000); + + const outcome = await handle.done; + expect(outcome.status).toBe("failed"); + if (outcome.status !== "failed") throw new Error("unreachable"); + expect(outcome.errorMessage).toContain("cancelled"); + }); + + it("cancelStage is idempotent — calling twice does not error", async () => { + const exec = createCommandExecutor(); + const stage = makeCommandStage({ + executor: { kind: "command", command: "/bin/sh", args: ["-c", "sleep 30"] }, + }); + + const handle = await exec.startStage(makeInput({ stage })); + await exec.cancelStage(handle); + // Second call must not throw. + await expect(exec.cancelStage(handle)).resolves.toBeUndefined(); + }); }); posixDescribe("command executor — timeout enforcement", () => { diff --git a/packages/core/src/__tests__/pipeline-engine-executors.test.ts b/packages/core/src/__tests__/pipeline-engine-executors.test.ts index eacd38614..e59391536 100644 --- a/packages/core/src/__tests__/pipeline-engine-executors.test.ts +++ b/packages/core/src/__tests__/pipeline-engine-executors.test.ts @@ -33,8 +33,11 @@ import { createPipelineEngine, createPipelineStore, formatForkRefusalMessage, + isTerminalLoopState, type AgentStageExecutor, type Pipeline, + type PipelineEngine, + type RunId, type RunningAgentStage, type Stage, type StageOutcome, @@ -43,6 +46,30 @@ import { import { createPluginRegistry } from "../plugin-registry.js"; import type { Agent, PluginManifest, PluginModule } from "../types.js"; +/** + * Poll the engine until the run reaches a terminal loop state (done / stalled + * / terminated). Command stages now use the non-blocking pattern — they're in + * commandInflight after startRun and only complete after tick() harvests the + * exited child. Fast subprocesses (true, echo) typically finish in one or two + * ticks; we cap at 200 iterations (≈10s) to avoid hanging on flaky hosts. + */ +async function tickUntilDone( + engine: PipelineEngine, + runId: RunId, + maxTicks = 200, + intervalMs = 50, +): Promise { + for (let i = 0; i < maxTicks; i++) { + const runState = engine.state().runs[runId]; + if (runState && isTerminalLoopState(runState.loopState)) return; + await engine.tick(); + // Brief yield so the child process has a chance to exit between ticks. + await new Promise((r) => setTimeout(r, intervalMs)); + } + // One final tick in case the process exited during the last sleep. + await engine.tick(); +} + const posixDescribe = isWindows() ? describe.skip : describe; let storeRoot: string; @@ -153,6 +180,8 @@ posixDescribe("engine + command executor", () => { headSha: "sha-aaa", }); + await tickUntilDone(engine, runId); + const run = store.loadRun(runId); expect(run?.stages["lint"]?.status).toBe("succeeded"); const stageRunId = run?.stages["lint"]?.stageRunId; @@ -188,6 +217,10 @@ posixDescribe("engine + command executor", () => { isFromFork: true, }); + // Fork refusal settles synchronously inside startRun (no subprocess), + // so a single tick is sufficient to confirm the terminal state. + await engine.tick(); + const run = store.loadRun(runId); expect(run?.stages["scan"]?.status).toBe("failed"); expect(run?.stages["scan"]?.errorMessage).toBe(formatForkRefusalMessage("scan")); @@ -221,6 +254,8 @@ posixDescribe("engine + command executor", () => { isFromFork: true, }); + await tickUntilDone(engine, runId); + const run = store.loadRun(runId); expect(run?.stages["scan"]?.status).toBe("succeeded"); }); @@ -281,6 +316,8 @@ posixDescribe("engine + builtin/router", () => { headSha: "sha-aaa", }); + await tickUntilDone(engine, runId); + const run = store.loadRun(runId); expect(run?.stages["lint"]?.status).toBe("succeeded"); expect(run?.stages["deliver"]?.status).toBe("succeeded"); @@ -328,13 +365,15 @@ posixDescribe("engine + builtin/router", () => { ], }; - await engine.startRun({ + const runId = await engine.startRun({ pipeline, projectId: "proj-a", sessionId: "ses-run-owner", headSha: "sha-aaa", }); + await tickUntilDone(engine, runId); + expect(sendToSession).toHaveBeenCalledWith("ses-run-owner", expect.any(String)); }); }); @@ -399,6 +438,8 @@ posixDescribe("engine + builtin/compose", () => { headSha: "sha-aaa", }); + await tickUntilDone(engine, runId); + const run = store.loadRun(runId); expect(run?.stages["merge"]?.status).toBe("succeeded"); expect(run?.loopState).toBe("done"); @@ -462,6 +503,8 @@ posixDescribe("engine + builtin executors — guardrails", () => { headSha: "sha-aaa", }); + await tickUntilDone(engine, runId); + const run = store.loadRun(runId); expect(run?.stages["dummy"]?.status).toBe("succeeded"); expect(run?.stages["deliver"]?.status).toBe("failed"); @@ -493,6 +536,9 @@ posixDescribe("engine + builtin executors — guardrails", () => { headSha: "sha-aaa", }); + // No subprocess — failure is instant. One tick is enough to confirm. + await engine.tick(); + const run = store.loadRun(runId); expect(run?.stages["lint"]?.status).toBe("failed"); expect(run?.stages["lint"]?.errorMessage).toContain("command executor not configured"); @@ -548,9 +594,61 @@ posixDescribe("engine + builtin executors — guardrails", () => { headSha: "sha-aaa", }); + await tickUntilDone(engine, runId); + const run = store.loadRun(runId); expect(run?.stages["deliver"]?.status).toBe("failed"); expect(run?.stages["deliver"]?.errorMessage).toContain("simulated executor crash"); expect(run?.loopState).toBe("stalled"); }); }); + +posixDescribe("engine + command executor — non-blocking handoff", () => { + it("does not block cancelRun / tick / parallel stages during a long-running command stage", async () => { + // Pipeline: two parallel `sleep 5` command stages (no dependsOn). + // After startRun returns, both should be in commandInflight (running). + // cancelRun should complete in <500ms (not have to wait for sleep to finish). + // After cancel, the run state should be 'terminated' and stages 'outdated' or 'failed'. + const registry = createPluginRegistry(); + registry.register(makeAgentPlugin()); + const store = createPipelineStore(storeRoot); + + const engine = createPipelineEngine({ + store, + registry, + agentExecutor: noopAgentExecutor(), + commandExecutor: createCommandExecutor(), + }); + + const pipeline: Pipeline = { + id: asPipelineId("pl-nonblock"), + name: "nonblock", + stages: [ + makeCommandStage("stage-a", "sleep 5"), + makeCommandStage("stage-b", "sleep 5"), + ], + }; + + const runId = await engine.startRun({ + pipeline, + projectId: "proj-a", + sessionId: "ses-1", + headSha: "sha-nb", + }); + + // After startRun the stages must be running (lock was released immediately + // after spawn — the children are alive in commandInflight). + const runAfterStart = engine.state().runs[runId]; + expect(runAfterStart?.loopState).toBe("running"); + + // cancelRun must complete far faster than the 5s sleep. + const cancelStart = Date.now(); + await engine.cancelRun(runId); + const cancelElapsed = Date.now() - cancelStart; + expect(cancelElapsed).toBeLessThan(500); + + // The run must be in a terminal state. + const runAfterCancel = engine.state().runs[runId]; + expect(isTerminalLoopState(runAfterCancel?.loopState ?? "running")).toBe(true); + }, 10_000); // generous wall-clock budget; the cancel itself must be <500ms +}); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index faaccf444..b69f62436 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -444,6 +444,7 @@ export { validatePipelineDag, // v1.2: command executor, builtin executors, findings parser createCommandExecutor, + CommandExecutorSpawnError, formatForkRefusalMessage, DEFAULT_COMMAND_TIMEOUT_MS, DEFAULT_COMMAND_STDOUT_CAP_BYTES, @@ -522,6 +523,7 @@ export type { CommandStartInput, CommandOutcome, CommandExecutorDeps, + RunningCommandStage, // v1.2: builtin executor types BuiltinExecutor, BuiltinRunInput, diff --git a/packages/core/src/pipeline/engine.ts b/packages/core/src/pipeline/engine.ts index 6626d45d9..07a3d1371 100644 --- a/packages/core/src/pipeline/engine.ts +++ b/packages/core/src/pipeline/engine.ts @@ -56,7 +56,12 @@ import { type StartStageInput, } from "./executors/agent.js"; import type { BuiltinExecutor } from "./executors/builtin-router.js"; -import type { CommandStageExecutor, CommandStartInput } from "./executors/command.js"; +import { + CommandExecutorSpawnError, + type CommandStageExecutor, + type CommandStartInput, + type RunningCommandStage, +} from "./executors/command.js"; export interface PipelineEngineDeps { store: PipelineStore; @@ -142,8 +147,10 @@ export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine { } = deps; let state: EngineState = deps.initialState ?? emptyEngineState(); - /** stageRunId → executor handle for stages we own. */ + /** stageRunId → executor handle for agent stages. */ const inflight = new Map(); + /** stageRunId → executor handle for command stages. */ + const commandInflight = new Map(); /** * Side-table for projectId/issueId/isFromFork, keyed by RunId. The persisted * RunState shape was locked by v0.1 and doesn't carry these, so the engine @@ -262,11 +269,20 @@ export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine { } case "CANCEL_STAGE": { - const handle = inflight.get(effect.stageRunId); - if (handle) { + const agentHandle = inflight.get(effect.stageRunId); + if (agentHandle) { inflight.delete(effect.stageRunId); try { - await agentExecutor.cancelStage(handle); + await agentExecutor.cancelStage(agentHandle); + } catch { + // Best-effort — handle may already be gone. + } + } + const cmdHandle = commandInflight.get(effect.stageRunId); + if (cmdHandle && commandExecutor) { + commandInflight.delete(effect.stageRunId); + try { + await commandExecutor.cancelStage(cmdHandle); } catch { // Best-effort — handle may already be gone. } @@ -351,41 +367,54 @@ export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine { isFromFork: meta?.isFromFork ?? false, }; - // Wrap the executor call so an unexpected throw (executor bug, - // platform error not surfaced as a CommandOutcome) still terminates - // the stage rather than escaping `executeEffect` and leaving the - // stage permanently `running`. - let outcome; + // startStage returns immediately after spawn (or instantly for synthetic + // failures). The lock is released as soon as this function returns — + // tick() polls commandInflight in a separate lock acquisition. + let handle; try { - outcome = await commandExecutor.run(input); + handle = await commandExecutor.startStage(input); } catch (err) { + // CommandExecutorSpawnError for kind mismatch — should not happen in + // practice since the engine only routes kind="command" stages here. await dispatchInline({ type: "STAGE_FAILED", now: now(), runId, stageName: stage.name, errorMessage: - err instanceof Error ? err.message : `command executor threw: ${String(err)}`, + err instanceof CommandExecutorSpawnError + ? err.message + : `command executor failed to start: ${err instanceof Error ? err.message : String(err)}`, }); return; } - if (outcome.status === "completed") { - await dispatchInline({ - type: "STAGE_COMPLETED", - now: now(), - runId, - stageName: stage.name, - artifacts: outcome.artifacts, - }); - } else { - await dispatchInline({ - type: "STAGE_FAILED", - now: now(), - runId, - stageName: stage.name, - errorMessage: outcome.errorMessage, - }); + + // If the outcome is already populated (fork refusal / sync spawn error), + // dispatch inline immediately without adding to commandInflight. + if (handle.outcome !== null) { + const outcome = handle.outcome; + if (outcome.status === "completed") { + await dispatchInline({ + type: "STAGE_COMPLETED", + now: now(), + runId, + stageName: stage.name, + artifacts: outcome.artifacts, + }); + } else { + await dispatchInline({ + type: "STAGE_FAILED", + now: now(), + runId, + stageName: stage.name, + errorMessage: outcome.errorMessage, + }); + } + return; } + + // Child is running. Store the handle for tick() to poll. + commandInflight.set(stageRunId, handle); } async function runBuiltinStage( @@ -480,9 +509,11 @@ export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine { async function tick(): Promise { return withLock(async () => { - if (inflight.size === 0) return; - const handles = [...inflight.values()]; - for (const handle of handles) { + if (inflight.size === 0 && commandInflight.size === 0) return; + + // Poll agent stages (async, may yield). + const agentHandles = [...inflight.values()]; + for (const handle of agentHandles) { const outcome = await agentExecutor.pollStage(handle); if (outcome.status === "running") continue; @@ -506,6 +537,37 @@ export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine { }); } } + + // Poll command stages (synchronous read of handle.outcome via pollStage). + // commandInflight is only populated when commandExecutor is defined, so + // this guard is belt-and-suspenders rather than a real runtime path. + if (commandExecutor) { + const commandHandles = [...commandInflight.values()]; + for (const handle of commandHandles) { + const outcome = commandExecutor.pollStage(handle); + if (outcome.status === "running") continue; + + commandInflight.delete(handle.stageRunId); + + if (outcome.status === "completed") { + await dispatchInline({ + type: "STAGE_COMPLETED", + now: now(), + runId: handle.runId, + stageName: handle.stageName, + artifacts: outcome.artifacts, + }); + } else { + await dispatchInline({ + type: "STAGE_FAILED", + now: now(), + runId: handle.runId, + stageName: handle.stageName, + errorMessage: outcome.errorMessage, + }); + } + } + } }); } diff --git a/packages/core/src/pipeline/executors/command.ts b/packages/core/src/pipeline/executors/command.ts index f42c5911c..091837a06 100644 --- a/packages/core/src/pipeline/executors/command.ts +++ b/packages/core/src/pipeline/executors/command.ts @@ -25,6 +25,11 @@ * logged via a `command.refused_fork` observation effect (the engine relays * it to the standard observation log) so operators can see why a stage was * skipped. + * + * Non-blocking design: `startStage` returns a handle as soon as the child + * process spawns (or instantly for synthetic failures like fork refusal / + * kind mismatch). The engine polls via `pollStage` inside `tick()` so the + * serialization lock is not held for the duration of the child process. */ import { spawn } from "node:child_process"; @@ -66,11 +71,77 @@ export interface CommandStartInput { } export type CommandOutcome = + | { status: "running" } | { status: "completed"; artifacts: ArtifactInput[] } | { status: "failed"; errorMessage: string; refused?: boolean }; +/** + * Handle for a command stage that's running. Mirrors `RunningAgentStage` from + * the agent executor. The engine holds onto this and threads it back into + * `pollStage` / `cancelStage`. + */ +export interface RunningCommandStage { + runId: RunId; + stageRunId: StageRunId; + stageName: string; + startedAt: number; + /** + * Set when the child exits, the timeout fires, spawn fails, or refusal + * triggers. `null` while the child is still running. + */ + outcome: Exclude | null; + /** + * Resolves with the terminal outcome once `outcome` is populated. Safe to + * `await` from tests or one-shot callers; the engine does NOT await this + * inside the lock. + */ + done: Promise>; + /** + * Idempotent. Clears timers and kills the child process tree if alive. + * Settles `outcome` and resolves `done` if they haven't been settled yet. + */ + cancel: () => void; +} + export interface CommandStageExecutor { - run(input: CommandStartInput): Promise; + /** + * Spawn the child and return a handle. Resolves AS SOON AS spawn returns + * (or instantly for synthetic failures like fork refusal / kind mismatch). + * The handle's `outcome` field is `null` while the child is running, and + * populated when the child exits / times out / errors. + * + * Throws `CommandExecutorSpawnError` only for `stage.executor.kind !== + * "command"` — all other failures are captured inside the handle. + */ + startStage(input: CommandStartInput): Promise; + /** + * Synchronous read of the current state. Returns `{ status: "running" }` + * if `handle.outcome` is still `null`. + */ + pollStage(handle: RunningCommandStage): CommandOutcome; + /** + * Best-effort cancel. Calls `handle.cancel()` internally. Idempotent. + */ + cancelStage(handle: RunningCommandStage): Promise; + /** + * Sugar: `startStage` + `await handle.done`. Suitable for tests / one-shot + * callers that don't need the non-blocking split. + * + * Re-throws `CommandExecutorSpawnError` if `startStage` throws. + */ + run(input: CommandStartInput): Promise>; +} + +/** + * Thrown when `startStage` is called with a stage whose `executor.kind` is + * not `"command"`. Engine catches this and dispatches STAGE_FAILED. + * Follows the same pattern as `AgentExecutorSpawnError` in agent.ts. + */ +export class CommandExecutorSpawnError extends Error { + constructor(message: string, options?: { cause?: unknown }) { + super(message, options); + this.name = "CommandExecutorSpawnError"; + } } /** @@ -86,9 +157,9 @@ export const DEFAULT_COMMAND_STDOUT_CAP_BYTES = 4 * 1024 * 1024; */ export const DEFAULT_COMMAND_STDERR_CAP_BYTES = 64 * 1024; /** - * Grace period between SIGTERM and SIGKILL when a stage times out. Long - * enough for a well-behaved child to flush stdout/stderr, short enough that - * the pipeline still recovers in bounded time. + * Grace period between SIGTERM and SIGKILL when a stage times out or is + * cancelled. Long enough for a well-behaved child to flush stdout/stderr, + * short enough that the pipeline still recovers in bounded time. */ export const COMMAND_KILL_GRACE_MS = 2_000; @@ -123,184 +194,269 @@ export function createCommandExecutor(deps: CommandExecutorDeps = {}): CommandSt const now = deps.now ?? Date.now; const defaultTimeoutMs = deps.defaultTimeoutMs ?? DEFAULT_COMMAND_TIMEOUT_MS; - return { - run(input: CommandStartInput): Promise { - const stage = input.stage; - if (stage.executor.kind !== "command") { - return Promise.resolve({ - status: "failed", - errorMessage: `command executor cannot run stage "${stage.name}" with executor.kind=${stage.executor.kind}`, - }); - } + async function startStage(input: CommandStartInput): Promise { + const stage = input.stage; - if (input.isFromFork && stage.allowFork !== true) { - const message = formatForkRefusalMessage(stage.name); - onRefuse(stage, message); - return Promise.resolve({ status: "failed", errorMessage: message, refused: true }); - } + if (stage.executor.kind !== "command") { + throw new CommandExecutorSpawnError( + `command executor cannot run stage "${stage.name}" with executor.kind=${stage.executor.kind}`, + ); + } - const executor = stage.executor; - const workspaceRoot = input.workspaceRoot ?? process.cwd(); - const cwd = executor.cwd ? join(workspaceRoot, executor.cwd) : workspaceRoot; - const env = buildChildEnv(input, executor.env); - const stdoutCap = DEFAULT_COMMAND_STDOUT_CAP_BYTES; - const timeoutMs = stage.timeoutMs ?? defaultTimeoutMs; - const startedAt = now(); + // Fork refusal: synthetic instant failure, no child spawned. + if (input.isFromFork && stage.allowFork !== true) { + const message = formatForkRefusalMessage(stage.name); + onRefuse(stage, message); + const terminalOutcome: Exclude = { + status: "failed", + errorMessage: message, + refused: true, + }; + return { + runId: input.runId, + stageRunId: input.stageRunId, + stageName: stage.name, + startedAt: now(), + outcome: terminalOutcome, + done: Promise.resolve(terminalOutcome), + cancel: () => undefined, + }; + } - return new Promise((resolve) => { - let child; - try { - child = spawnFn(executor.command, executor.args ?? [], { - cwd, - env, - stdio: ["ignore", "pipe", "pipe"], - // Windows installs node-based CLIs as `.cmd` shims that `spawn` - // can't resolve without going through cmd.exe. See - // docs/CROSS_PLATFORM.md. - shell: isWindows(), - // Detach the child into its own process group on Unix so a - // timeout kill can reach grandchildren (e.g. `sh -c 'sleep 30'` - // — without a group kill, sleep inherits the stdio pipes and - // `close` never fires). No effect on Windows; taskkill /T - // walks the tree by PID. - detached: !isWindows(), - }); - } catch (err) { - resolve({ - status: "failed", - errorMessage: `failed to spawn command "${executor.command}": ${ - err instanceof Error ? err.message : String(err) - }`, - }); - return; - } + const executor = stage.executor; + const workspaceRoot = input.workspaceRoot ?? process.cwd(); + const cwd = executor.cwd ? join(workspaceRoot, executor.cwd) : workspaceRoot; + const env = buildChildEnv(input, executor.env); + const stdoutCap = DEFAULT_COMMAND_STDOUT_CAP_BYTES; + const timeoutMs = stage.timeoutMs ?? defaultTimeoutMs; + const startedAt = now(); - const stdoutChunks: Buffer[] = []; - const stderrChunks: Buffer[] = []; - let stdoutBytes = 0; - let stderrBytes = 0; - let truncated = false; - let stderrTruncated = false; - let settled = false; - let timedOut = false; - let killTimer: NodeJS.Timeout | null = null; - let forceTimer: NodeJS.Timeout | null = null; - - const clearTimers = () => { - if (killTimer) clearTimeout(killTimer); - if (forceTimer) clearTimeout(forceTimer); - killTimer = null; - forceTimer = null; - }; - - const settle = (outcome: CommandOutcome) => { - if (settled) return; - settled = true; - clearTimers(); - resolve(outcome); - }; - - if (timeoutMs > 0 && Number.isFinite(timeoutMs)) { - killTimer = setTimeout(() => { - if (settled) return; - timedOut = true; - // Best-effort graceful shutdown of the whole process tree, - // escalating to SIGKILL after a short grace window so a wedged - // child (or shell grandchild) can't hold the pipeline. Fire and - // forget — `close` resolves the outer promise. - if (child.pid !== undefined) { - void killProcessTree(child.pid, "SIGTERM"); - } - forceTimer = setTimeout(() => { - if (child.pid !== undefined) { - void killProcessTree(child.pid, "SIGKILL"); - } - }, COMMAND_KILL_GRACE_MS); - }, timeoutMs); - } - - child.stdout?.on("data", (chunk: Buffer) => { - stdoutBytes += chunk.length; - if (stdoutBytes <= stdoutCap) { - stdoutChunks.push(chunk); - } else if (!truncated) { - truncated = true; - // Keep everything up to the cap; drop overflow rather than OOM. - const overflow = stdoutBytes - stdoutCap; - if (chunk.length > overflow) { - stdoutChunks.push(chunk.subarray(0, chunk.length - overflow)); - } - } - }); - child.stderr?.on("data", (chunk: Buffer) => { - stderrBytes += chunk.length; - if (stderrBytes <= DEFAULT_COMMAND_STDERR_CAP_BYTES) { - stderrChunks.push(chunk); - } else if (!stderrTruncated) { - stderrTruncated = true; - // Keep everything up to the cap; drop overflow rather than OOM. - const overflow = stderrBytes - DEFAULT_COMMAND_STDERR_CAP_BYTES; - if (chunk.length > overflow) { - stderrChunks.push(chunk.subarray(0, chunk.length - overflow)); - } - } - }); - child.on("error", (err) => { - settle({ - status: "failed", - errorMessage: `command "${executor.command}" failed: ${err.message}`, - }); - }); - child.on("close", (code, signal) => { - const stderrRaw = Buffer.concat(stderrChunks).toString("utf-8").trim(); - const stderr = stderrTruncated ? `${stderrRaw}\n[stderr truncated]` : stderrRaw; - if (timedOut) { - const elapsed = now() - startedAt; - settle({ - status: "failed", - errorMessage: `command "${executor.command}" timed out after ${timeoutMs}ms (ran for ${elapsed}ms)${stderr ? `: ${stderr}` : ""}`, - }); - return; - } - if (truncated) { - settle({ - status: "failed", - errorMessage: `command "${executor.command}" stdout exceeded ${stdoutCap} bytes${stderr ? `: ${stderr}` : ""}`, - }); - return; - } - if (signal) { - settle({ - status: "failed", - errorMessage: `command "${executor.command}" terminated by signal ${signal}${stderr ? `: ${stderr}` : ""}`, - }); - return; - } - if (code !== 0) { - settle({ - status: "failed", - errorMessage: `command "${executor.command}" exited ${code}${stderr ? `: ${stderr}` : ""}`, - }); - return; - } - const stdout = Buffer.concat(stdoutChunks).toString("utf-8"); - let artifacts: ArtifactInput[]; - try { - artifacts = parseStdoutFindings(stdout); - } catch (err) { - settle({ - status: "failed", - errorMessage: `command "${executor.command}" produced unparseable findings: ${ - err instanceof Error ? err.message : String(err) - }`, - }); - return; - } - settle({ status: "completed", artifacts }); - }); + // Attempt to spawn the child. Synchronous spawn errors produce an instant + // failed handle without throwing from startStage. + let child; + try { + child = spawnFn(executor.command, executor.args ?? [], { + cwd, + env, + stdio: ["ignore", "pipe", "pipe"], + // Windows installs node-based CLIs as `.cmd` shims that `spawn` + // can't resolve without going through cmd.exe. See + // docs/CROSS_PLATFORM.md. + shell: isWindows(), + // Detach the child into its own process group on Unix so a + // timeout kill can reach grandchildren (e.g. `sh -c 'sleep 30'` + // — without a group kill, sleep inherits the stdio pipes and + // `close` never fires). No effect on Windows; taskkill /T + // walks the tree by PID. + detached: !isWindows(), }); - }, - }; + } catch (err) { + const terminalOutcome: Exclude = { + status: "failed", + errorMessage: `failed to spawn command "${executor.command}": ${ + err instanceof Error ? err.message : String(err) + }`, + }; + return { + runId: input.runId, + stageRunId: input.stageRunId, + stageName: stage.name, + startedAt, + outcome: terminalOutcome, + done: Promise.resolve(terminalOutcome), + cancel: () => undefined, + }; + } + + // Child spawned. Build the handle with shared mutable state. + const stdoutChunks: Buffer[] = []; + const stderrChunks: Buffer[] = []; + let stdoutBytes = 0; + let stderrBytes = 0; + let truncated = false; + let stderrTruncated = false; + let settled = false; + let cancelled = false; + let timedOut = false; + let killTimer: NodeJS.Timeout | null = null; + let forceTimer: NodeJS.Timeout | null = null; + + // `resolveOutcome` is populated by the Promise constructor below. + let resolveOutcome!: ( + outcome: Exclude, + ) => void; + + const done = new Promise>( + (resolve) => { + resolveOutcome = resolve; + }, + ); + + // `handle.outcome` starts null; settle() populates it and resolves `done`. + const handle: RunningCommandStage = { + runId: input.runId, + stageRunId: input.stageRunId, + stageName: stage.name, + startedAt, + outcome: null, + done, + cancel: () => { + if (settled || cancelled) return; + cancelled = true; + clearTimers(); + if (child.pid !== undefined) { + void killProcessTree(child.pid, "SIGTERM"); + forceTimer = setTimeout(() => { + if (child.pid !== undefined) { + void killProcessTree(child.pid, "SIGKILL"); + } + }, COMMAND_KILL_GRACE_MS); + } + // Settle immediately with cancelled failure so pollStage / done + // see the terminal outcome without waiting for the close event. + settle({ + status: "failed", + errorMessage: `command "${executor.command}" was cancelled`, + }); + }, + }; + + const clearTimers = () => { + if (killTimer) clearTimeout(killTimer); + if (forceTimer) clearTimeout(forceTimer); + killTimer = null; + forceTimer = null; + }; + + const settle = (outcome: Exclude) => { + if (settled) return; + settled = true; + clearTimers(); + handle.outcome = outcome; + resolveOutcome(outcome); + }; + + if (timeoutMs > 0 && Number.isFinite(timeoutMs)) { + killTimer = setTimeout(() => { + if (settled) return; + timedOut = true; + // Best-effort graceful shutdown of the whole process tree, + // escalating to SIGKILL after a short grace window so a wedged + // child (or shell grandchild) can't hold the pipeline. Fire and + // forget — `close` resolves the outer promise. + if (child.pid !== undefined) { + void killProcessTree(child.pid, "SIGTERM"); + } + forceTimer = setTimeout(() => { + if (child.pid !== undefined) { + void killProcessTree(child.pid, "SIGKILL"); + } + }, COMMAND_KILL_GRACE_MS); + }, timeoutMs); + } + + child.stdout?.on("data", (chunk: Buffer) => { + stdoutBytes += chunk.length; + if (stdoutBytes <= stdoutCap) { + stdoutChunks.push(chunk); + } else if (!truncated) { + truncated = true; + // Keep everything up to the cap; drop overflow rather than OOM. + const overflow = stdoutBytes - stdoutCap; + if (chunk.length > overflow) { + stdoutChunks.push(chunk.subarray(0, chunk.length - overflow)); + } + } + }); + child.stderr?.on("data", (chunk: Buffer) => { + stderrBytes += chunk.length; + if (stderrBytes <= DEFAULT_COMMAND_STDERR_CAP_BYTES) { + stderrChunks.push(chunk); + } else if (!stderrTruncated) { + stderrTruncated = true; + // Keep everything up to the cap; drop overflow rather than OOM. + const overflow = stderrBytes - DEFAULT_COMMAND_STDERR_CAP_BYTES; + if (chunk.length > overflow) { + stderrChunks.push(chunk.subarray(0, chunk.length - overflow)); + } + } + }); + child.on("error", (err) => { + settle({ + status: "failed", + errorMessage: `command "${executor.command}" failed: ${err.message}`, + }); + }); + child.on("close", (code, signal) => { + // If we already settled (e.g. cancel() was called), don't overwrite. + if (settled) return; + const stderrRaw = Buffer.concat(stderrChunks).toString("utf-8").trim(); + const stderr = stderrTruncated ? `${stderrRaw}\n[stderr truncated]` : stderrRaw; + if (timedOut) { + const elapsed = now() - startedAt; + settle({ + status: "failed", + errorMessage: `command "${executor.command}" timed out after ${timeoutMs}ms (ran for ${elapsed}ms)${stderr ? `: ${stderr}` : ""}`, + }); + return; + } + if (truncated) { + settle({ + status: "failed", + errorMessage: `command "${executor.command}" stdout exceeded ${stdoutCap} bytes${stderr ? `: ${stderr}` : ""}`, + }); + return; + } + if (signal) { + settle({ + status: "failed", + errorMessage: `command "${executor.command}" terminated by signal ${signal}${stderr ? `: ${stderr}` : ""}`, + }); + return; + } + if (code !== 0) { + settle({ + status: "failed", + errorMessage: `command "${executor.command}" exited ${code}${stderr ? `: ${stderr}` : ""}`, + }); + return; + } + const stdout = Buffer.concat(stdoutChunks).toString("utf-8"); + let artifacts: ArtifactInput[]; + try { + artifacts = parseStdoutFindings(stdout); + } catch (err) { + settle({ + status: "failed", + errorMessage: `command "${executor.command}" produced unparseable findings: ${ + err instanceof Error ? err.message : String(err) + }`, + }); + return; + } + settle({ status: "completed", artifacts }); + }); + + return handle; + } + + function pollStage(handle: RunningCommandStage): CommandOutcome { + return handle.outcome ?? { status: "running" }; + } + + async function cancelStage(handle: RunningCommandStage): Promise { + handle.cancel(); + } + + async function run( + input: CommandStartInput, + ): Promise> { + // Let CommandExecutorSpawnError propagate to the caller (engine catches it). + const handle = await startStage(input); + return handle.done; + } + + return { startStage, pollStage, cancelStage, run }; } function buildChildEnv( diff --git a/packages/core/src/pipeline/executors/index.ts b/packages/core/src/pipeline/executors/index.ts index fee464a97..17211e50e 100644 --- a/packages/core/src/pipeline/executors/index.ts +++ b/packages/core/src/pipeline/executors/index.ts @@ -11,6 +11,7 @@ export { export { createCommandExecutor, + CommandExecutorSpawnError, formatForkRefusalMessage, DEFAULT_COMMAND_TIMEOUT_MS, DEFAULT_COMMAND_STDOUT_CAP_BYTES, @@ -20,6 +21,7 @@ export { type CommandStageExecutor, type CommandStartInput, type CommandOutcome, + type RunningCommandStage, } from "./command.js"; export { parseFindingsJsonl, coerceArtifactInput } from "./findings-parser.js"; diff --git a/packages/core/src/pipeline/index.ts b/packages/core/src/pipeline/index.ts index dcfafdbfc..90f41e619 100644 --- a/packages/core/src/pipeline/index.ts +++ b/packages/core/src/pipeline/index.ts @@ -40,6 +40,7 @@ export { type StageOutcome, type StartStageInput, createCommandExecutor, + CommandExecutorSpawnError, formatForkRefusalMessage, DEFAULT_COMMAND_TIMEOUT_MS, DEFAULT_COMMAND_STDOUT_CAP_BYTES, @@ -49,6 +50,7 @@ export { type CommandStageExecutor, type CommandStartInput, type CommandOutcome, + type RunningCommandStage, parseFindingsJsonl, coerceArtifactInput, createBuiltinRouterExecutor,