bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(pipeline): non-blocking command executor — release engine lock during child runtime 

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<StageRunId, RunningCommandStage>` (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.
This commit is contained in:
Harsh Batheja 2026-05-17 00:51:45 +05:30
parent 1e81033200
commit 62d0630d7c
7 changed files with 686 additions and 219 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

165
packages/core/src/__tests__/pipeline-command-executor.test.ts
View File

@ -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<CommandStartInput> = {}): 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", () => {

100
packages/core/src/__tests__/pipeline-engine-executors.test.ts
View File

@ -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<void> {
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<void>((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
});

2
packages/core/src/index.ts
View File

@ -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,

124
packages/core/src/pipeline/engine.ts
View File

@ -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, RunningAgentStage>();
/** stageRunId → executor handle for command stages. */
const commandInflight = new Map<StageRunId, RunningCommandStage>();
/**
* 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<void> {
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,
});
}
}
}
});
}

510
packages/core/src/pipeline/executors/command.ts
View File

@ -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<CommandOutcome, { status: "running" }> | 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<Exclude<CommandOutcome, { status: "running" }>>;
/**
* 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<CommandOutcome>;
/**
* 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<RunningCommandStage>;
/**
* 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<void>;
/**
* 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<Exclude<CommandOutcome, { status: "running" }>>;
}
/**
* 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<CommandOutcome> {
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<RunningCommandStage> {
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<CommandOutcome, { status: "running" }> = {
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<CommandOutcome>((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<CommandOutcome, { status: "running" }> = {
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<CommandOutcome, { status: "running" }>,
) => void;
const done = new Promise<Exclude<CommandOutcome, { status: "running" }>>(
(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<CommandOutcome, { status: "running" }>) => {
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<void> {
handle.cancel();
}
async function run(
input: CommandStartInput,
): Promise<Exclude<CommandOutcome, { status: "running" }>> {
// Let CommandExecutorSpawnError propagate to the caller (engine catches it).
const handle = await startStage(input);
return handle.done;
}
return { startStage, pollStage, cancelStage, run };
}
function buildChildEnv(

2
packages/core/src/pipeline/executors/index.ts
View File

@ -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";

2
packages/core/src/pipeline/index.ts
View File

@ -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,