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

feat(pipeline): v1.2 — command executor + builtin router + builtin compose 

Adds three new stage executor kinds to the pipeline engine:

- `command`: shell-based stages parsed via stdout JSON/JSONL. Refuses
  fork PRs unless the stage opts in via Stage.allowFork; the refusal is
  surfaced as a STAGE_FAILED with the documented message. Not
  talk-to-able (per locked decision 10).
- `builtin/router`: replaces the original SEND_TO_AGENT reducer command;
  reads upstream findings, formats a payload, and delivers it to a
  target session via BuiltinTaskContext.sendToSession.
- `builtin/compose`: merges findings from multiple upstream stages into
  a single composite json artifact for downstream consumption.

Other surface changes:
- PRInfo gains `isFromFork: boolean`; scm-github and scm-gitlab populate
  it from `isCrossRepository` / source-vs-target project ids.
- Stage gains optional `allowFork` flag.
- New BuiltinTaskContext interface exposes only the capabilities
  builtins need (readSiblingArtifacts, sendToSession).
- Shared findings-parser extracted from agent.ts so command + agent
  stages validate ArtifactInput records the same way.
- Engine dispatches START_STAGE by executor kind; missing
  command/builtin executors fail the stage with a clear error rather
  than hanging.

Closes #1631
This commit is contained in:
Harsh Batheja 2026-05-16 23:00:56 +05:30
parent 9144c0ec06
commit 058b3bdf14
24 changed files with 2147 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

151
packages/core/src/__tests__/pipeline-builtin-compose.test.ts Normal file
View File

@ -0,0 +1,151 @@
/**
* Tests for the builtin/compose executor.
*/
import { describe, expect, it, vi } from "vitest";
import {
asArtifactId,
asRunId,
asStageRunId,
createBuiltinComposeExecutor,
type Artifact,
type BuiltinRunInput,
type BuiltinTaskContext,
type Stage,
} from "../pipeline/index.js";
function makeStage(overrides: Partial<Stage> = {}): Stage {
return {
name: "compose",
trigger: { on: ["manual"] },
executor: { kind: "builtin/compose", fromStages: ["lint", "scan"] },
task: {},
dependsOn: ["lint", "scan"],
...overrides,
};
}
function makeFinding(stage: string, title: string): Artifact {
return {
artifactId: asArtifactId(`art-${stage}-${title}`),
pipelineRunId: asRunId("run-1"),
stageRunId: asStageRunId(`sr-${stage}`),
stageName: stage,
status: "open",
createdAt: new Date().toISOString(),
kind: "finding",
filePath: "src/x.ts",
startLine: 1,
endLine: 1,
title,
description: "d",
category: "general",
severity: "info",
confidence: 0.5,
} as Artifact;
}
function makeCtx(): {
ctx: BuiltinTaskContext;
read: ReturnType<typeof vi.fn>;
} {
const read = vi.fn(async (_: string): Promise<Artifact[]> => []);
const ctx: BuiltinTaskContext = {
runId: asRunId("run-1"),
stageRunId: asStageRunId("sr-1"),
stageName: "compose",
sessionId: "session-self",
pipelineName: "default",
readSiblingArtifacts: read,
sendToSession: vi.fn(async () => undefined),
};
return { ctx, read };
}
function makeInput(ctx: BuiltinTaskContext, overrides: Partial<BuiltinRunInput> = {}): BuiltinRunInput {
return {
runId: asRunId("run-1"),
stageRunId: asStageRunId("sr-1"),
stage: makeStage(),
loopRound: 1,
ctx,
...overrides,
};
}
describe("builtin/compose — guards", () => {
it("rejects non-builtin-compose stages", async () => {
const { ctx } = makeCtx();
const exec = createBuiltinComposeExecutor();
const outcome = await exec.run(
makeInput(ctx, {
stage: makeStage({ executor: { kind: "command", command: "true" } }),
}),
);
expect(outcome.status).toBe("failed");
});
});
describe("builtin/compose — bundling", () => {
it("merges findings from two upstream stages into a single composite artifact", async () => {
const { ctx, read } = makeCtx();
const lintFindings = [makeFinding("lint", "lint-A"), makeFinding("lint", "lint-B")];
const scanFindings = [makeFinding("scan", "scan-A")];
read.mockImplementation(async (stage: string) =>
stage === "lint" ? lintFindings : stage === "scan" ? scanFindings : [],
);
const exec = createBuiltinComposeExecutor();
const outcome = await exec.run(makeInput(ctx));
expect(outcome.status).toBe("completed");
if (outcome.status !== "completed") throw new Error("unreachable");
expect(outcome.artifacts).toHaveLength(1);
const [composite] = outcome.artifacts;
expect(composite.kind).toBe("json");
if (composite.kind !== "json") throw new Error("unreachable");
expect(composite.data).toMatchObject({
builtin: "compose",
pipelineName: "default",
sourceStages: ["lint", "scan"],
loopRound: 1,
bundles: [
{ stage: "lint", count: 2, artifacts: lintFindings },
{ stage: "scan", count: 1, artifacts: scanFindings },
],
});
});
it("still emits a composite when upstream stages produced no findings", async () => {
const { ctx, read } = makeCtx();
read.mockResolvedValue([]);
const exec = createBuiltinComposeExecutor();
const outcome = await exec.run(makeInput(ctx));
expect(outcome.status).toBe("completed");
if (outcome.status !== "completed") throw new Error("unreachable");
expect(outcome.artifacts).toHaveLength(1);
const composite = outcome.artifacts[0];
if (composite.kind !== "json") throw new Error("unreachable");
expect(composite.data).toMatchObject({
bundles: [
{ stage: "lint", count: 0, artifacts: [] },
{ stage: "scan", count: 0, artifacts: [] },
],
});
});
it("surfaces readSiblingArtifacts errors as failed", async () => {
const { ctx, read } = makeCtx();
read.mockRejectedValueOnce(new Error("store unavailable"));
const exec = createBuiltinComposeExecutor();
const outcome = await exec.run(makeInput(ctx));
expect(outcome.status).toBe("failed");
if (outcome.status !== "failed") throw new Error("unreachable");
expect(outcome.errorMessage).toContain("store unavailable");
});
});

195
packages/core/src/__tests__/pipeline-builtin-router.test.ts Normal file
View File

@ -0,0 +1,195 @@
/**
* Tests for the builtin/router executor.
*
* Uses an in-memory BuiltinTaskContext mock — the executor is pure over its
* context so no subprocess / session manager is needed at this layer. The
* end-to-end "delivers findings to a target session" check lives in
* pipeline-builtin-router.integration.test.ts.
*/
import { describe, expect, it, vi } from "vitest";
import {
asArtifactId,
asRunId,
asStageRunId,
createBuiltinRouterExecutor,
type Artifact,
type BuiltinRunInput,
type BuiltinTaskContext,
type Stage,
} from "../pipeline/index.js";
function makeStage(overrides: Partial<Stage> = {}): Stage {
return {
name: "router",
trigger: { on: ["manual"] },
executor: {
kind: "builtin/router",
fromStages: ["lint", "scan"],
target: { kind: "self" },
},
task: {},
dependsOn: ["lint", "scan"],
...overrides,
};
}
function makeFinding(overrides: Partial<Artifact> = {}): Artifact {
return {
artifactId: asArtifactId("art-1"),
pipelineRunId: asRunId("run-1"),
stageRunId: asStageRunId("sr-up-1"),
stageName: "lint",
status: "open",
createdAt: new Date().toISOString(),
kind: "finding",
filePath: "src/foo.ts",
startLine: 1,
endLine: 2,
title: "missing return",
description: "fn missing return",
category: "correctness",
severity: "warning",
confidence: 0.7,
...overrides,
} as Artifact;
}
function makeCtx(overrides: Partial<BuiltinTaskContext> = {}): {
ctx: BuiltinTaskContext;
send: ReturnType<typeof vi.fn>;
read: ReturnType<typeof vi.fn>;
} {
const send = vi.fn(async () => undefined);
const read = vi.fn(async (_stage: string): Promise<Artifact[]> => []);
const ctx: BuiltinTaskContext = {
runId: asRunId("run-1"),
stageRunId: asStageRunId("sr-1"),
stageName: "router",
sessionId: "session-self",
pipelineName: "default",
readSiblingArtifacts: read,
sendToSession: send,
...overrides,
};
return { ctx, send, read };
}
function makeInput(ctx: BuiltinTaskContext, overrides: Partial<BuiltinRunInput> = {}): BuiltinRunInput {
return {
runId: asRunId("run-1"),
stageRunId: asStageRunId("sr-1"),
stage: makeStage(),
loopRound: 2,
ctx,
...overrides,
};
}
describe("builtin/router — guards", () => {
it("rejects non-builtin-router stages", async () => {
const { ctx } = makeCtx();
const exec = createBuiltinRouterExecutor();
const outcome = await exec.run(
makeInput(ctx, {
stage: makeStage({ executor: { kind: "agent", plugin: "claude-code", mode: "code" } }),
}),
);
expect(outcome.status).toBe("failed");
});
});
describe("builtin/router — delivery", () => {
it("reads siblings, formats payload, delivers to the resolved target", async () => {
const findings = [makeFinding({ title: "issue A" })];
const scans = [makeFinding({ stageName: "scan", title: "issue B" })];
const { ctx, send, read } = makeCtx({
sessionId: "session-self",
});
read.mockImplementation(async (stage: string) =>
stage === "lint" ? findings : stage === "scan" ? scans : [],
);
const exec = createBuiltinRouterExecutor();
const outcome = await exec.run(makeInput(ctx));
expect(outcome.status).toBe("completed");
if (outcome.status !== "completed") throw new Error("unreachable");
expect(read).toHaveBeenCalledWith("lint");
expect(read).toHaveBeenCalledWith("scan");
expect(send).toHaveBeenCalledTimes(1);
const [target, payload] = send.mock.calls[0];
expect(target).toBe("session-self");
expect(payload).toContain("Pipeline routing: default → router");
expect(payload).toContain("Loop round: 2");
expect(payload).toContain("## From stage: lint (1 artifact)");
expect(payload).toContain("issue A");
expect(payload).toContain("## From stage: scan (1 artifact)");
expect(payload).toContain("issue B");
expect(outcome.artifacts).toHaveLength(1);
expect(outcome.artifacts[0]).toMatchObject({
kind: "json",
data: {
builtin: "router",
targetSessionId: "session-self",
bundles: [
{ stage: "lint", count: 1 },
{ stage: "scan", count: 1 },
],
},
});
});
it("delivers an empty payload when upstream stages produced no artifacts", async () => {
const { ctx, send, read } = makeCtx();
read.mockResolvedValue([]);
const exec = createBuiltinRouterExecutor();
const outcome = await exec.run(makeInput(ctx));
expect(outcome.status).toBe("completed");
if (outcome.status !== "completed") throw new Error("unreachable");
const payload = send.mock.calls[0]?.[1] as string;
expect(payload).toContain("_no findings_");
expect(outcome.artifacts[0]).toMatchObject({
data: { bundles: [{ count: 0 }, { count: 0 }] },
});
});
it("routes to a literal session id when target.kind === 'session'", async () => {
const { ctx, send } = makeCtx();
const exec = createBuiltinRouterExecutor();
const outcome = await exec.run(
makeInput(ctx, {
stage: makeStage({
executor: {
kind: "builtin/router",
fromStages: ["lint"],
target: { kind: "session", sessionId: "ses-target" },
},
}),
}),
);
expect(outcome.status).toBe("completed");
expect(send).toHaveBeenCalledWith("ses-target", expect.any(String));
});
it("surfaces sendToSession errors as failed", async () => {
const { ctx, send } = makeCtx();
send.mockRejectedValueOnce(new Error("session not found"));
const exec = createBuiltinRouterExecutor();
const outcome = await exec.run(makeInput(ctx));
expect(outcome.status).toBe("failed");
if (outcome.status !== "failed") throw new Error("unreachable");
expect(outcome.errorMessage).toContain("session not found");
expect(outcome.errorMessage).toContain("session-self");
});
});

268
packages/core/src/__tests__/pipeline-command-executor.test.ts Normal file
View File

@ -0,0 +1,268 @@
/**
* Tests for the command executor — shell-based pipeline stages.
*
* The executor is exercised against real subprocesses via /bin/sh -c so we
* cover the actual spawn → stdout-capture → JSONL-parse path end-to-end.
* Fork-PR refusal and executor-kind guards run in-process with no subprocess.
*/
import { describe, expect, it, vi } from "vitest";
import {
asRunId,
asStageRunId,
createCommandExecutor,
formatForkRefusalMessage,
type CommandStartInput,
type Stage,
} from "../pipeline/index.js";
function makeCommandStage(overrides: Partial<Stage> = {}): Stage {
return {
name: "lint",
trigger: { on: ["pr.opened"] },
executor: { kind: "command", command: "echo", args: [] },
task: {},
...overrides,
};
}
function makeInput(overrides: Partial<CommandStartInput> = {}): CommandStartInput {
return {
pipelineName: "default",
runId: asRunId("run-1"),
stageRunId: asStageRunId("sr-1"),
stage: makeCommandStage(),
loopRound: 1,
...overrides,
};
}
describe("command executor — guards", () => {
it("rejects non-command stages with a typed failure", async () => {
const exec = createCommandExecutor();
const outcome = await 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");
});
it("refuses to run a fork PR when stage.allowFork is unset", 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 outcome = await exec.run(makeInput({ stage, isFromFork: true }));
expect(outcome.status).toBe("failed");
if (outcome.status !== "failed") throw new Error("unreachable");
expect(outcome.refused).toBe(true);
expect(outcome.errorMessage).toBe(formatForkRefusalMessage(stage.name));
expect(onRefuse).toHaveBeenCalledTimes(1);
expect(onRefuse).toHaveBeenCalledWith(stage, formatForkRefusalMessage(stage.name));
});
it("refuses to run a fork PR when stage.allowFork is explicitly false", async () => {
const exec = createCommandExecutor();
const stage = makeCommandStage({
allowFork: false,
executor: { kind: "command", command: "/bin/sh", args: ["-c", "echo hi"] },
});
const outcome = await exec.run(makeInput({ stage, isFromFork: true }));
expect(outcome.status).toBe("failed");
if (outcome.status !== "failed") throw new Error("unreachable");
expect(outcome.refused).toBe(true);
});
it("runs a fork PR when stage.allowFork is explicitly true", async () => {
const exec = createCommandExecutor();
const stage = makeCommandStage({
allowFork: true,
executor: { kind: "command", command: "/bin/sh", args: ["-c", "echo -n ''"] },
});
const outcome = await exec.run(makeInput({ stage, isFromFork: true }));
expect(outcome).toEqual({ status: "completed", artifacts: [] });
});
it("runs non-fork PRs without checking allowFork", async () => {
const exec = createCommandExecutor();
const stage = makeCommandStage({
executor: { kind: "command", command: "/bin/sh", args: ["-c", "true"] },
});
const outcome = await exec.run(makeInput({ stage, isFromFork: false }));
expect(outcome.status).toBe("completed");
});
});
describe("command executor — stdout findings", () => {
it("parses JSONL stdout into ArtifactInput records", async () => {
const exec = createCommandExecutor();
const finding = {
kind: "finding",
filePath: "src/foo.ts",
startLine: 1,
endLine: 2,
title: "t",
description: "d",
category: "general",
severity: "info",
confidence: 0.5,
};
const jsonArtifact = { kind: "json", data: { ok: true } };
// Single-quoted in sh so JSON.stringify's double quotes survive. JSON
// never emits literal single quotes so the embedding is unambiguous.
const script = `echo '${JSON.stringify(finding)}'; echo '${JSON.stringify(jsonArtifact)}'`;
const stage = makeCommandStage({
executor: { kind: "command", command: "/bin/sh", args: ["-c", script] },
});
const outcome = await exec.run(makeInput({ stage }));
expect(outcome.status).toBe("completed");
if (outcome.status !== "completed") throw new Error("unreachable");
expect(outcome.artifacts).toHaveLength(2);
expect(outcome.artifacts[0]).toMatchObject({ kind: "finding", title: "t" });
expect(outcome.artifacts[1]).toMatchObject({ kind: "json", data: { ok: true } });
});
it("parses a single JSON array stdout into ArtifactInput records", async () => {
const exec = createCommandExecutor();
const arr = [{ kind: "json", data: { a: 1 } }, { kind: "json", data: { b: 2 } }];
const stage = makeCommandStage({
executor: {
kind: "command",
command: "/bin/sh",
args: ["-c", `printf '%s' ${JSON.stringify(JSON.stringify(arr))}`],
},
});
const outcome = await exec.run(makeInput({ stage }));
expect(outcome.status).toBe("completed");
if (outcome.status !== "completed") throw new Error("unreachable");
expect(outcome.artifacts).toEqual(arr);
});
it("treats empty stdout as zero artifacts", async () => {
const exec = createCommandExecutor();
const stage = makeCommandStage({
executor: { kind: "command", command: "/bin/sh", args: ["-c", "true"] },
});
const outcome = await exec.run(makeInput({ stage }));
expect(outcome).toEqual({ status: "completed", artifacts: [] });
});
it("fails on non-zero exit codes and surfaces stderr in the error", async () => {
const exec = createCommandExecutor();
const stage = makeCommandStage({
executor: {
kind: "command",
command: "/bin/sh",
args: ["-c", "echo boom >&2; exit 3"],
},
});
const outcome = await exec.run(makeInput({ stage }));
expect(outcome.status).toBe("failed");
if (outcome.status !== "failed") throw new Error("unreachable");
expect(outcome.errorMessage).toContain("exited 3");
expect(outcome.errorMessage).toContain("boom");
});
it("fails when stdout is invalid JSON", async () => {
const exec = createCommandExecutor();
const stage = makeCommandStage({
executor: {
kind: "command",
command: "/bin/sh",
args: ["-c", "echo 'not json {{{'"],
},
});
const outcome = await exec.run(makeInput({ stage }));
expect(outcome.status).toBe("failed");
if (outcome.status !== "failed") throw new Error("unreachable");
expect(outcome.errorMessage).toContain("unparseable findings");
});
it("fails when a finding has confidence out of [0, 1]", async () => {
const exec = createCommandExecutor();
const bad = {
kind: "finding",
filePath: "x.ts",
startLine: 1,
endLine: 1,
title: "t",
description: "d",
category: "c",
severity: "info",
confidence: 5,
};
const stage = makeCommandStage({
executor: {
kind: "command",
command: "/bin/sh",
args: ["-c", `printf '%s' ${JSON.stringify(JSON.stringify(bad))}`],
},
});
const outcome = await exec.run(makeInput({ stage }));
expect(outcome.status).toBe("failed");
if (outcome.status !== "failed") throw new Error("unreachable");
expect(outcome.errorMessage).toContain("confidence");
});
});
describe("command executor — environment", () => {
it("threads AO_PIPELINE_* env vars into the child process", async () => {
const exec = createCommandExecutor();
const stage = makeCommandStage({
executor: {
kind: "command",
command: "/bin/sh",
args: [
"-c",
'printf \'{"kind":"json","data":{"stage":"%s","run":"%s"}}\' "$AO_PIPELINE_STAGE_NAME" "$AO_PIPELINE_RUN_ID"',
],
},
});
const outcome = await exec.run(
makeInput({
runId: asRunId("run-xyz"),
stage,
}),
);
expect(outcome.status).toBe("completed");
if (outcome.status !== "completed") throw new Error("unreachable");
expect(outcome.artifacts[0]).toMatchObject({
kind: "json",
data: { stage: "lint", run: "run-xyz" },
});
});
it("lets stage.env overrides win over default env", async () => {
const exec = createCommandExecutor();
const stage = makeCommandStage({
executor: {
kind: "command",
command: "/bin/sh",
args: ["-c", 'printf \'{"kind":"json","data":{"v":"%s"}}\' "$MY_VAR"'],
env: { MY_VAR: "from-stage" },
},
});
const outcome = await exec.run(makeInput({ stage }));
expect(outcome.status).toBe("completed");
if (outcome.status !== "completed") throw new Error("unreachable");
expect(outcome.artifacts[0]).toMatchObject({ kind: "json", data: { v: "from-stage" } });
});
});

492
packages/core/src/__tests__/pipeline-engine-executors.test.ts Normal file
View File

@ -0,0 +1,492 @@
/**
* Engine-level integration tests for the v1.2 executors.
*
* Verifies that command, builtin/router, and builtin/compose stages are
* wired into the engine end-to-end:
* - Multi-stage DAG: lint (command) → compose → router → target session.
* - Fork-PR refusal at the engine boundary (command stage marks the run
* stalled with the documented refusal message).
* - Router stage actually invokes the sendToSession callback with a
* payload that contains the upstream findings.
* - Compose stage merges two upstream stages into a single composite.
*
* Agent executor is mocked so we can drive the engine without sessions.
*/
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import {
asPipelineId,
createBuiltinComposeExecutor,
createBuiltinRouterExecutor,
createCommandExecutor,
createPipelineEngine,
createPipelineStore,
formatForkRefusalMessage,
type AgentStageExecutor,
type Pipeline,
type RunningAgentStage,
type Stage,
type StageOutcome,
type StartStageInput,
} from "../pipeline/index.js";
import { createPluginRegistry } from "../plugin-registry.js";
import type { Agent, PluginManifest, PluginModule } from "../types.js";
let storeRoot: string;
beforeEach(() => {
storeRoot = mkdtempSync(join(tmpdir(), "pipeline-engine-exec-"));
});
afterEach(() => {
rmSync(storeRoot, { recursive: true, force: true });
});
function makeAgentPlugin(): PluginModule<Agent> {
const manifest: PluginManifest = {
name: "codex",
slot: "agent",
description: "test",
version: "0.0.0",
supportedTaskModes: ["review", "code", "answer"],
};
return {
manifest,
create: () =>
({
name: "codex",
processName: "codex",
getLaunchCommand: () => "true",
getEnvironment: () => ({}),
detectActivity: () => "idle",
getActivityState: async () => null,
isProcessRunning: async () => true,
getSessionInfo: async () => null,
}) as Agent,
};
}
function noopAgentExecutor(): AgentStageExecutor {
return {
async startStage(input: StartStageInput): Promise<RunningAgentStage> {
return {
runId: input.runId,
stageRunId: input.stageRunId,
stageName: input.stage.name,
sessionId: "mock-ses",
workspacePath: "/tmp/mock",
startedAt: Date.now(),
input,
};
},
async pollStage(): Promise<StageOutcome> {
return { status: "running" };
},
async cancelStage(): Promise<void> {
// no-op
},
};
}
function makeCommandStage(name: string, command: string, overrides: Partial<Stage> = {}): Stage {
return {
name,
trigger: { on: ["manual"] },
executor: { kind: "command", command: "/bin/sh", args: ["-c", command] },
task: {},
...overrides,
};
}
describe("engine + command executor", () => {
it("runs a command stage to completion and persists its findings", async () => {
const registry = createPluginRegistry();
registry.register(makeAgentPlugin());
const store = createPipelineStore(storeRoot);
const engine = createPipelineEngine({
store,
registry,
agentExecutor: noopAgentExecutor(),
commandExecutor: createCommandExecutor(),
});
const finding = {
kind: "finding" as const,
filePath: "src/x.ts",
startLine: 1,
endLine: 1,
title: "lint-finding",
description: "demo",
category: "general",
severity: "info" as const,
confidence: 0.5,
};
const pipeline: Pipeline = {
id: asPipelineId("pl-cmd"),
name: "cmd",
stages: [
makeCommandStage(
"lint",
`printf '%s' ${JSON.stringify(JSON.stringify(finding))}`,
),
],
};
const runId = await engine.startRun({
pipeline,
projectId: "proj-a",
sessionId: "ses-1",
headSha: "sha-aaa",
});
const run = store.loadRun(runId);
expect(run?.stages["lint"]?.status).toBe("succeeded");
const stageRunId = run?.stages["lint"]?.stageRunId;
if (!stageRunId) throw new Error("no stageRunId");
const artifacts = store.listArtifacts(runId, stageRunId);
expect(artifacts).toHaveLength(1);
expect(artifacts[0]).toMatchObject({ kind: "finding", title: "lint-finding" });
});
it("refuses command stages on fork PRs unless allowFork: true", async () => {
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-fork"),
name: "fork",
stages: [makeCommandStage("scan", "echo should-not-run")],
};
const runId = await engine.startRun({
pipeline,
projectId: "proj-a",
sessionId: "ses-1",
headSha: "sha-fork",
isFromFork: true,
});
const run = store.loadRun(runId);
expect(run?.stages["scan"]?.status).toBe("failed");
expect(run?.stages["scan"]?.errorMessage).toBe(formatForkRefusalMessage("scan"));
// Run terminates as stalled on stage failure (per v1.1 reducer semantics).
expect(run?.loopState).toBe("stalled");
});
it("runs a command stage on a fork PR when allowFork: true", async () => {
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-fork-ok"),
name: "fork-ok",
stages: [makeCommandStage("scan", "true", { allowFork: true })],
};
const runId = await engine.startRun({
pipeline,
projectId: "proj-a",
sessionId: "ses-1",
headSha: "sha-fork",
isFromFork: true,
});
const run = store.loadRun(runId);
expect(run?.stages["scan"]?.status).toBe("succeeded");
});
});
describe("engine + builtin/router", () => {
it("delivers upstream findings to a target session via sendToSession", async () => {
const registry = createPluginRegistry();
registry.register(makeAgentPlugin());
const store = createPipelineStore(storeRoot);
const sendToSession = vi.fn(async (_id: string, _payload: string) => undefined);
const engine = createPipelineEngine({
store,
registry,
agentExecutor: noopAgentExecutor(),
commandExecutor: createCommandExecutor(),
builtinRouter: createBuiltinRouterExecutor(),
sendToSession,
});
const finding = {
kind: "finding" as const,
filePath: "src/x.ts",
startLine: 1,
endLine: 1,
title: "router-finding",
description: "demo",
category: "general",
severity: "warning" as const,
confidence: 0.9,
};
const pipeline: Pipeline = {
id: asPipelineId("pl-router"),
name: "router-pipe",
stages: [
makeCommandStage("lint", `printf '%s' ${JSON.stringify(JSON.stringify(finding))}`),
{
name: "deliver",
trigger: { on: ["manual"] },
executor: {
kind: "builtin/router",
fromStages: ["lint"],
target: { kind: "session", sessionId: "ses-target" },
},
task: {},
dependsOn: ["lint"],
},
],
};
const runId = await engine.startRun({
pipeline,
projectId: "proj-a",
sessionId: "ses-self",
headSha: "sha-aaa",
});
const run = store.loadRun(runId);
expect(run?.stages["lint"]?.status).toBe("succeeded");
expect(run?.stages["deliver"]?.status).toBe("succeeded");
expect(run?.loopState).toBe("done");
expect(sendToSession).toHaveBeenCalledTimes(1);
const [target, payload] = sendToSession.mock.calls[0];
expect(target).toBe("ses-target");
expect(payload).toContain("Pipeline routing: router-pipe → deliver");
expect(payload).toContain("router-finding");
});
it("resolves target.kind: 'self' to the run's session id", async () => {
const registry = createPluginRegistry();
registry.register(makeAgentPlugin());
const store = createPipelineStore(storeRoot);
const sendToSession = vi.fn(async () => undefined);
const engine = createPipelineEngine({
store,
registry,
agentExecutor: noopAgentExecutor(),
commandExecutor: createCommandExecutor(),
builtinRouter: createBuiltinRouterExecutor(),
sendToSession,
});
const pipeline: Pipeline = {
id: asPipelineId("pl-router-self"),
name: "router-self",
stages: [
makeCommandStage("lint", "true"),
{
name: "deliver",
trigger: { on: ["manual"] },
executor: {
kind: "builtin/router",
fromStages: ["lint"],
target: { kind: "self" },
},
task: {},
dependsOn: ["lint"],
},
],
};
await engine.startRun({
pipeline,
projectId: "proj-a",
sessionId: "ses-run-owner",
headSha: "sha-aaa",
});
expect(sendToSession).toHaveBeenCalledWith("ses-run-owner", expect.any(String));
});
});
describe("engine + builtin/compose", () => {
it("merges findings from two upstream stages into a single composite artifact", async () => {
const registry = createPluginRegistry();
registry.register(makeAgentPlugin());
const store = createPipelineStore(storeRoot);
const engine = createPipelineEngine({
store,
registry,
agentExecutor: noopAgentExecutor(),
commandExecutor: createCommandExecutor(),
builtinCompose: createBuiltinComposeExecutor(),
});
const lintFinding = {
kind: "finding" as const,
filePath: "a.ts",
startLine: 1,
endLine: 1,
title: "lint-A",
description: "d",
category: "c",
severity: "info" as const,
confidence: 0.4,
};
const scanFinding = {
kind: "finding" as const,
filePath: "b.ts",
startLine: 5,
endLine: 6,
title: "scan-A",
description: "d",
category: "c",
severity: "error" as const,
confidence: 1,
};
const pipeline: Pipeline = {
id: asPipelineId("pl-compose"),
name: "compose-pipe",
stages: [
makeCommandStage("lint", `printf '%s' ${JSON.stringify(JSON.stringify(lintFinding))}`),
makeCommandStage("scan", `printf '%s' ${JSON.stringify(JSON.stringify(scanFinding))}`),
{
name: "merge",
trigger: { on: ["manual"] },
executor: { kind: "builtin/compose", fromStages: ["lint", "scan"] },
task: {},
dependsOn: ["lint", "scan"],
},
],
};
const runId = await engine.startRun({
pipeline,
projectId: "proj-a",
sessionId: "ses-1",
headSha: "sha-aaa",
});
const run = store.loadRun(runId);
expect(run?.stages["merge"]?.status).toBe("succeeded");
expect(run?.loopState).toBe("done");
const mergeStageRunId = run?.stages["merge"]?.stageRunId;
if (!mergeStageRunId) throw new Error("missing stageRunId");
const composites = store.listArtifacts(runId, mergeStageRunId);
expect(composites).toHaveLength(1);
const composite = composites[0];
expect(composite.kind).toBe("json");
if (composite.kind !== "json") throw new Error("unreachable");
expect(composite.data).toMatchObject({
builtin: "compose",
sourceStages: ["lint", "scan"],
bundles: [
{ stage: "lint", count: 1 },
{ stage: "scan", count: 1 },
],
});
});
});
describe("engine + builtin executors — guardrails", () => {
it("fails a builtin/router stage when sendToSession is not configured", async () => {
const registry = createPluginRegistry();
registry.register(makeAgentPlugin());
const store = createPipelineStore(storeRoot);
const engine = createPipelineEngine({
store,
registry,
agentExecutor: noopAgentExecutor(),
commandExecutor: createCommandExecutor(),
builtinRouter: createBuiltinRouterExecutor(),
// sendToSession intentionally omitted
});
const pipeline: Pipeline = {
id: asPipelineId("pl-router-missing"),
name: "router-missing",
stages: [
makeCommandStage("dummy", "true"),
{
name: "deliver",
trigger: { on: ["manual"] },
executor: {
kind: "builtin/router",
fromStages: ["dummy"],
target: { kind: "self" },
},
task: {},
dependsOn: ["dummy"],
},
],
};
const runId = await engine.startRun({
pipeline,
projectId: "proj-a",
sessionId: "ses-1",
headSha: "sha-aaa",
});
const run = store.loadRun(runId);
expect(run?.stages["dummy"]?.status).toBe("succeeded");
expect(run?.stages["deliver"]?.status).toBe("failed");
expect(run?.stages["deliver"]?.errorMessage).toContain("sendToSession");
});
it("fails a command stage when commandExecutor is not configured", async () => {
const registry = createPluginRegistry();
registry.register(makeAgentPlugin());
const store = createPipelineStore(storeRoot);
const engine = createPipelineEngine({
store,
registry,
agentExecutor: noopAgentExecutor(),
// commandExecutor intentionally omitted
});
const pipeline: Pipeline = {
id: asPipelineId("pl-no-cmd"),
name: "no-cmd",
stages: [makeCommandStage("lint", "true")],
};
const runId = await engine.startRun({
pipeline,
projectId: "proj-a",
sessionId: "ses-1",
headSha: "sha-aaa",
});
const run = store.loadRun(runId);
expect(run?.stages["lint"]?.status).toBe("failed");
expect(run?.stages["lint"]?.errorMessage).toContain("command executor not configured");
});
});

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

@ -293,7 +293,7 @@ describe("pipeline engine — end-to-end", () => {
expect(executor.startCalls).toHaveLength(0);
});
it("synthesizes STAGE_FAILED for non-agent executor kinds (v0.2 only supports agent)", async () => {
it("fails a command stage with a clear message when commandExecutor is not configured", async () => {
const registry = withRegistry([makeAgentPlugin("codex", ["review"])]);
const store = createPipelineStore(storeRoot);
const executor = makeMockExecutor();
@ -317,7 +317,7 @@ describe("pipeline engine — end-to-end", () => {
const run = store.loadRun(runId)!;
expect(run.stages["lint"]?.status).toBe("failed");
expect(run.stages["lint"]?.errorMessage).toContain("not supported in v0.2");
expect(run.stages["lint"]?.errorMessage).toContain("command executor not configured");
expect(executor.startCalls).toHaveLength(0);
});

3
packages/core/src/__tests__/session-manager/claim-pr.test.ts
View File

@ -49,6 +49,7 @@ describe("claimPR", () => {
branch: "feat/existing-pr",
baseBranch: "main",
isDraft: false,
isFromFork: false,
}),
assignPRToCurrentUser: vi.fn().mockResolvedValue(undefined),
checkoutPR: vi.fn().mockResolvedValue(true),
@ -344,6 +345,7 @@ describe("claimPR", () => {
branch: "feat/first-pr",
baseBranch: "main",
isDraft: false,
isFromFork: false,
})
.mockResolvedValueOnce({
number: 99,
@ -354,6 +356,7 @@ describe("claimPR", () => {
branch: "feat/second-pr",
baseBranch: "main",
isDraft: false,
isFromFork: false,
}),
checkoutPR: vi.fn().mockResolvedValue(true),
});

1
packages/core/src/__tests__/test-utils.ts
View File

@ -129,6 +129,7 @@ export function makePR(overrides: Partial<PRInfo> = {}): PRInfo {
branch: "feat/test",
baseBranch: "main",
isDraft: false,
isFromFork: false,
...overrides,
};
}

72
packages/core/src/pipeline/config-schema.ts
View File

@ -44,9 +44,25 @@ const CommandExecutorSchema = z.object({
cwd: z.string().optional(),
});
const BuiltinRouterExecutorSchema = z.object({
kind: z.literal("builtin/router"),
fromStages: z.array(z.string().min(1)).min(1),
target: z.discriminatedUnion("kind", [
z.object({ kind: z.literal("session"), sessionId: z.string().min(1) }),
z.object({ kind: z.literal("self") }),
]),
});
const BuiltinComposeExecutorSchema = z.object({
kind: z.literal("builtin/compose"),
fromStages: z.array(z.string().min(1)).min(1),
});
const StageExecutorSchema = z.discriminatedUnion("kind", [
AgentExecutorSchema,
CommandExecutorSchema,
BuiltinRouterExecutorSchema,
BuiltinComposeExecutorSchema,
]);
const TaskSpecSchema = z.object({
@ -87,6 +103,7 @@ const StageSchema = z.object({
maxLoopRounds: z.number().int().positive().optional(),
dependsOn: z.array(z.string().min(1)).optional(),
routes: StageRoutesSchema.optional(),
allowFork: z.boolean().optional(),
});
/**
@ -191,21 +208,7 @@ export type PipelinesConfig = z.infer<typeof PipelinesConfigSchema>;
/** Convert a parsed YAML pipeline entry into a runtime Pipeline (branded id). */
export function configuredPipelineToRuntime(key: string, configured: ConfiguredPipeline): Pipeline {
const stages = configured.stages.map((stage): Stage => {
const executor: StageExecutor =
stage.executor.kind === "agent"
? {
kind: "agent",
plugin: stage.executor.plugin,
mode: stage.executor.mode as TaskMode,
...(stage.executor.config !== undefined ? { config: stage.executor.config } : {}),
}
: {
kind: "command",
command: stage.executor.command,
...(stage.executor.args !== undefined ? { args: stage.executor.args } : {}),
...(stage.executor.env !== undefined ? { env: stage.executor.env } : {}),
...(stage.executor.cwd !== undefined ? { cwd: stage.executor.cwd } : {}),
};
const executor: StageExecutor = toRuntimeExecutor(stage.executor);
const routes: StageRoutes | undefined = stage.routes
? {
@ -228,6 +231,7 @@ export function configuredPipelineToRuntime(key: string, configured: ConfiguredP
...(stage.maxLoopRounds !== undefined ? { maxLoopRounds: stage.maxLoopRounds } : {}),
...(stage.dependsOn !== undefined ? { dependsOn: [...stage.dependsOn] } : {}),
...(routes ? { routes } : {}),
...(stage.allowFork !== undefined ? { allowFork: stage.allowFork } : {}),
};
});
@ -240,3 +244,41 @@ export function configuredPipelineToRuntime(key: string, configured: ConfiguredP
: {}),
};
}
function toRuntimeExecutor(executor: ConfiguredPipeline["stages"][number]["executor"]): StageExecutor {
switch (executor.kind) {
case "agent":
return {
kind: "agent",
plugin: executor.plugin,
mode: executor.mode as TaskMode,
...(executor.config !== undefined ? { config: executor.config } : {}),
};
case "command":
return {
kind: "command",
command: executor.command,
...(executor.args !== undefined ? { args: executor.args } : {}),
...(executor.env !== undefined ? { env: executor.env } : {}),
...(executor.cwd !== undefined ? { cwd: executor.cwd } : {}),
};
case "builtin/router":
return {
kind: "builtin/router",
fromStages: [...executor.fromStages],
target:
executor.target.kind === "session"
? { kind: "session", sessionId: executor.target.sessionId }
: { kind: "self" },
};
case "builtin/compose":
return {
kind: "builtin/compose",
fromStages: [...executor.fromStages],
};
default: {
const exhaustive: never = executor;
throw new Error(`Unhandled executor kind: ${JSON.stringify(exhaustive)}`);
}
}
}

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

@ -39,9 +39,13 @@ import {
asStageRunId,
emptyEngineState,
isTerminalLoopState,
type Artifact,
type BuiltinTaskContext,
type EngineState,
type Pipeline,
type RunId,
type RunState,
type Stage,
type StageRunId,
type StageTriggerEvent,
} from "./types.js";
@ -51,11 +55,28 @@ import {
type RunningAgentStage,
type StartStageInput,
} from "./executors/agent.js";
import type { BuiltinExecutor } from "./executors/builtin-router.js";
import type { CommandStageExecutor, CommandStartInput } from "./executors/command.js";
export interface PipelineEngineDeps {
store: PipelineStore;
registry: PluginRegistry;
agentExecutor: AgentStageExecutor;
/**
* Required for `command` stages. When omitted, a `command` stage fails with
* a clear "command executor not configured" error rather than hanging.
*/
commandExecutor?: CommandStageExecutor;
/** Required for `builtin/router` stages. */
builtinRouter?: BuiltinExecutor;
/** Required for `builtin/compose` stages. */
builtinCompose?: BuiltinExecutor;
/**
* Callback used by `builtin/router` to deliver payloads to a target
* session. Typically `(id, msg) => sessionManager.send(asSessionId(id), msg)`.
* Without it, router stages fail with "sendToSession not configured".
*/
sendToSession?(sessionId: string, payload: string): Promise<void>;
/** Optional initial state (e.g. restored from disk on startup). Defaults to empty. */
initialState?: EngineState;
/** Override clock for tests. */
@ -72,6 +93,12 @@ export interface StartRunInput {
headSha: string;
/** Optional issue id forwarded into spawned sessions. */
issueId?: string;
/**
* True when the triggering PR is from a fork. Forwarded into the command
* executor so it can enforce the fork-safety opt-in (`Stage.allowFork`).
* Defaults to `false` for non-PR runs (manual triggers).
*/
isFromFork?: boolean;
}
export interface PipelineEngine {
@ -103,18 +130,30 @@ export interface PipelineEngine {
}
export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine {
const { store, registry, agentExecutor, now = Date.now } = deps;
const {
store,
registry,
agentExecutor,
commandExecutor,
builtinRouter,
builtinCompose,
sendToSession,
now = Date.now,
} = deps;
let state: EngineState = deps.initialState ?? emptyEngineState();
/** stageRunId → executor handle for stages we own. */
const inflight = new Map<StageRunId, RunningAgentStage>();
/**
* Side-table for projectId/issueId, keyed by RunId. The persisted RunState
* shape was locked by v0.1 and doesn't carry these, so the engine threads
* them out-of-band into START_STAGE inputs. Pruned by
* 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
* threads them out-of-band into START_STAGE inputs. Pruned by
* `pruneTerminatedRunMetadata` after every dispatch.
*/
const runMetadata = new Map<RunId, { projectId: string; issueId?: string }>();
const runMetadata = new Map<
RunId,
{ projectId: string; issueId?: string; isFromFork: boolean }
>();
/**
* Serialization lock for top-level dispatches. Each public dispatch chains
@ -190,21 +229,11 @@ export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine {
case "START_STAGE": {
const run = state.runs[effect.runId];
if (!run) break;
if (effect.stage.executor.kind !== "agent") {
// command/builtin executors are out of scope for v0.2 — synthesize
// a STAGE_FAILED so the run terminates cleanly instead of hanging.
await dispatchInline({
type: "STAGE_FAILED",
now: now(),
runId: effect.runId,
stageName: effect.stage.name,
errorMessage: `Executor kind "${effect.stage.executor.kind}" is not supported in v0.2 (agent executor only).`,
});
break;
}
// Mark the stage as running BEFORE starting the executor — failures
// during spawn translate to STAGE_FAILED, which requires running|pending.
// Mark the stage as running BEFORE handing off to an executor —
// failures during spawn translate to STAGE_FAILED, which requires
// running|pending. The reducer guards against double-START so this
// is safe even when the executor completes synchronously.
await dispatchInline({
type: "STAGE_STARTED",
now: now(),
@ -213,27 +242,20 @@ export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine {
});
const meta = runMetadata.get(run.runId);
const startInput: StartStageInput = {
pipelineName: run.pipelineName,
projectId: meta?.projectId ?? "",
runId: effect.runId,
stageRunId: effect.stageRunId,
stage: effect.stage,
loopRound: run.loopRounds,
...(meta?.issueId ? { issueId: meta.issueId } : {}),
};
try {
const handle = await agentExecutor.startStage(startInput);
inflight.set(effect.stageRunId, handle);
} catch (err) {
const kind = effect.stage.executor.kind;
if (kind === "agent") {
await runAgentStage(run, effect.runId, effect.stageRunId, effect.stage, meta);
} else if (kind === "command") {
await runCommandStage(run, effect.runId, effect.stage, meta);
} else if (kind === "builtin/router" || kind === "builtin/compose") {
await runBuiltinStage(run, effect.runId, effect.stageRunId, effect.stage);
} else {
await dispatchInline({
type: "STAGE_FAILED",
now: now(),
runId: effect.runId,
stageName: effect.stage.name,
errorMessage:
err instanceof Error ? err.message : `agent executor failed: ${String(err)}`,
errorMessage: `Executor kind "${kind}" is not supported.`,
});
}
break;
@ -259,6 +281,169 @@ export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine {
}
}
async function runAgentStage(
run: RunState,
runId: RunId,
stageRunId: StageRunId,
stage: Stage,
meta: { projectId: string; issueId?: string } | undefined,
): Promise<void> {
const startInput: StartStageInput = {
pipelineName: run.pipelineName,
projectId: meta?.projectId ?? "",
runId,
stageRunId,
stage,
loopRound: run.loopRounds,
...(meta?.issueId ? { issueId: meta.issueId } : {}),
};
try {
const handle = await agentExecutor.startStage(startInput);
inflight.set(stageRunId, handle);
} catch (err) {
await dispatchInline({
type: "STAGE_FAILED",
now: now(),
runId,
stageName: stage.name,
errorMessage:
err instanceof Error ? err.message : `agent executor failed: ${String(err)}`,
});
}
}
async function runCommandStage(
run: RunState,
runId: RunId,
stage: Stage,
meta: { projectId: string; issueId?: string; isFromFork: boolean } | undefined,
): Promise<void> {
if (!commandExecutor) {
await dispatchInline({
type: "STAGE_FAILED",
now: now(),
runId,
stageName: stage.name,
errorMessage: `command executor not configured for stage "${stage.name}"`,
});
return;
}
const stageRunId = run.stages[stage.name]?.stageRunId;
if (!stageRunId) {
await dispatchInline({
type: "STAGE_FAILED",
now: now(),
runId,
stageName: stage.name,
errorMessage: `command stage "${stage.name}" has no stageRunId in run state`,
});
return;
}
const input: CommandStartInput = {
pipelineName: run.pipelineName,
runId,
stageRunId,
stage,
loopRound: run.loopRounds,
isFromFork: meta?.isFromFork ?? false,
};
const outcome = await commandExecutor.run(input);
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,
});
}
}
async function runBuiltinStage(
run: RunState,
runId: RunId,
stageRunId: StageRunId,
stage: Stage,
): Promise<void> {
const isRouter = stage.executor.kind === "builtin/router";
const executor = isRouter ? builtinRouter : builtinCompose;
if (!executor) {
await dispatchInline({
type: "STAGE_FAILED",
now: now(),
runId,
stageName: stage.name,
errorMessage: `${stage.executor.kind} executor not configured for stage "${stage.name}"`,
});
return;
}
const ctx = createBuiltinContext(run, runId, stageRunId, stage.name);
const outcome = await executor.run({
runId,
stageRunId,
stage,
loopRound: run.loopRounds,
ctx,
});
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,
});
}
}
function createBuiltinContext(
run: RunState,
runId: RunId,
stageRunId: StageRunId,
stageName: string,
): BuiltinTaskContext {
return {
runId,
stageRunId,
stageName,
sessionId: run.sessionId,
pipelineName: run.pipelineName,
readSiblingArtifacts: async (upstreamStageName: string): Promise<Artifact[]> => {
const upstream = run.stages[upstreamStageName];
if (!upstream) return [];
return store.listArtifacts(runId, upstream.stageRunId);
},
sendToSession: async (targetSessionId: string, payload: string): Promise<void> => {
if (!sendToSession) {
throw new Error("sendToSession callback not configured on engine deps");
}
await sendToSession(targetSessionId, payload);
},
};
}
async function tick(): Promise<void> {
return withLock(async () => {
if (inflight.size === 0) return;
@ -307,12 +492,14 @@ export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine {
stageRunIds[stage.name] = asStageRunId(`sr-${randomUUID()}`);
}
// Stash projectId/issueId BEFORE dispatch so the START_STAGE effect — which
// fires synchronously inside the same dispatch — can read them. The
// persisted RunState shape was locked by v0.1, so we carry these out-of-band.
// Stash projectId/issueId/isFromFork BEFORE dispatch so the START_STAGE
// effect — which fires synchronously inside the same dispatch — can read
// them. The persisted RunState shape was locked by v0.1, so we carry
// these out-of-band.
runMetadata.set(runId, {
projectId: input.projectId,
issueId: input.issueId,
isFromFork: input.isFromFork ?? false,
});
await withLock(() =>

88
packages/core/src/pipeline/executors/agent.ts
View File

@ -33,6 +33,7 @@ import {
type Stage,
type StageRunId,
} from "../types.js";
import { parseFindingsJsonl } from "./findings-parser.js";
export interface AgentExecutorDeps {
sessionManager: SessionManager;
@ -235,92 +236,7 @@ function parseFindingsFile(path: string): ArtifactInput[] {
// engine by dumping its full reasoning trace into findings. Fine for v0.2
// since the engine and agent are co-located, but stream-and-cap once the
// engine moves out-of-process or runs untrusted plugins.
const body = readFileSync(path, "utf-8");
const out: ArtifactInput[] = [];
for (const [lineNo, raw] of body.split("\n").entries()) {
const trimmed = raw.trim();
if (!trimmed) continue;
let parsed: unknown;
try {
parsed = JSON.parse(trimmed);
} catch (err) {
throw new Error(`line ${lineNo + 1}: ${err instanceof Error ? err.message : String(err)}`, {
cause: err,
});
}
out.push(coerceArtifactInput(parsed, lineNo + 1));
}
return out;
}
const VALID_SEVERITIES = ["error", "warning", "info"] as const;
function coerceArtifactInput(value: unknown, lineNo: number): ArtifactInput {
if (!value || typeof value !== "object") {
throw new Error(`line ${lineNo}: expected object`);
}
const obj = value as Record<string, unknown>;
if (obj["kind"] === "finding") {
requireString(obj, "filePath", lineNo);
requireNumber(obj, "startLine", lineNo);
requireNumber(obj, "endLine", lineNo);
requireString(obj, "title", lineNo);
requireString(obj, "description", lineNo);
requireString(obj, "category", lineNo);
requireEnum(obj, "severity", VALID_SEVERITIES, lineNo);
requireNumberInRange(obj, "confidence", 0, 1, lineNo);
return obj as unknown as ArtifactInput;
}
if (obj["kind"] === "json") {
if (!obj["data"] || typeof obj["data"] !== "object") {
throw new Error(`line ${lineNo}: "json" artifact requires object \`data\``);
}
return obj as unknown as ArtifactInput;
}
throw new Error(`line ${lineNo}: unknown artifact kind=${JSON.stringify(obj["kind"])}`);
}
function requireString(obj: Record<string, unknown>, key: string, lineNo: number): void {
if (typeof obj[key] !== "string") {
throw new Error(`line ${lineNo}: missing string field "${key}"`);
}
}
function requireNumber(obj: Record<string, unknown>, key: string, lineNo: number): void {
if (typeof obj[key] !== "number") {
throw new Error(`line ${lineNo}: missing numeric field "${key}"`);
}
}
function requireNumberInRange(
obj: Record<string, unknown>,
key: string,
min: number,
max: number,
lineNo: number,
): void {
const value = obj[key];
if (typeof value !== "number" || value < min || value > max) {
throw new Error(
`line ${lineNo}: field "${key}" must be a number in [${min}, ${max}], got ${JSON.stringify(value)}`,
);
}
}
function requireEnum<T extends string>(
obj: Record<string, unknown>,
key: string,
allowed: readonly T[],
lineNo: number,
): void {
const value = obj[key];
if (typeof value !== "string" || !(allowed as readonly string[]).includes(value)) {
throw new Error(
`line ${lineNo}: field "${key}" must be one of ${allowed
.map((v) => `"${v}"`)
.join(", ")}, got ${JSON.stringify(value)}`,
);
}
return parseFindingsJsonl(readFileSync(path, "utf-8"));
}
async function safeKill(sm: SessionManager, sessionId: SessionId): Promise<void> {

79
packages/core/src/pipeline/executors/builtin-compose.ts Normal file
View File

@ -0,0 +1,79 @@
/**
* Builtin compose executor — `builtin/compose` stages.
*
* Bundles findings from multiple upstream sibling stages into a single
* composite artifact for a downstream stage to consume. The composite is
* emitted as a `json` artifact whose `data.findings` is the merged list,
* tagged with the originating stage name.
*
* Use case: a `route` stage downstream wants to deliver "everything the
* pipeline found this round" to one session. Composing avoids the
* O(stages) explosion of separate router calls and makes the round-trip
* auditable (the composite artifact is persisted in the pipeline store).
*
* Failure modes:
* - Wrong executor kind → returns failed.
* - `readSiblingArtifacts` throws → returns failed.
* - Zero upstream artifacts → still emits a composite artifact with
* `bundles[].count: 0` so downstream stages know the upstream ran.
*/
import type { ArtifactInput, BuiltinTaskContext, RunId, Stage, StageRunId } from "../types.js";
import type { BuiltinExecutor, BuiltinOutcome } from "./builtin-router.js";
export interface ComposeRunInput {
runId: RunId;
stageRunId: StageRunId;
stage: Stage;
loopRound?: number;
ctx: BuiltinTaskContext;
}
export function createBuiltinComposeExecutor(): BuiltinExecutor {
return {
async run(input): Promise<BuiltinOutcome> {
const stage = input.stage;
if (stage.executor.kind !== "builtin/compose") {
return {
status: "failed",
errorMessage: `builtin/compose executor cannot run stage "${stage.name}" with executor.kind=${stage.executor.kind}`,
};
}
const executor = stage.executor;
const ctx = input.ctx;
const bundles: Array<{ stage: string; artifacts: unknown[] }> = [];
try {
for (const upstream of executor.fromStages) {
const artifacts = await ctx.readSiblingArtifacts(upstream);
bundles.push({ stage: upstream, artifacts });
}
} catch (err) {
return {
status: "failed",
errorMessage: `builtin/compose failed to read sibling artifacts: ${
err instanceof Error ? err.message : String(err)
}`,
};
}
const composite: ArtifactInput = {
kind: "json",
data: {
builtin: "compose",
pipelineName: ctx.pipelineName,
sourceStages: [...executor.fromStages],
...(input.loopRound !== undefined ? { loopRound: input.loopRound } : {}),
bundles: bundles.map((b) => ({
stage: b.stage,
count: b.artifacts.length,
artifacts: b.artifacts,
})),
},
};
return { status: "completed", artifacts: [composite] };
},
};
}

136
packages/core/src/pipeline/executors/builtin-router.ts Normal file
View File

@ -0,0 +1,136 @@
/**
* Builtin router executor — `builtin/router` stages.
*
* Replaces the original spec's `SEND_TO_AGENT` reducer command. A router
* stage:
* 1. Reads findings emitted by each `fromStages` upstream sibling via
* `BuiltinTaskContext.readSiblingArtifacts`.
* 2. Formats them into a single payload (one block per source stage).
* 3. Delivers the payload to the target session via
* `BuiltinTaskContext.sendToSession`.
*
* The executor returns one `json` artifact recording what was delivered so
* downstream stages (and the dashboard) can audit the routing decision.
*
* Targets:
* - `{ kind: "self" }` resolves to the session this run is scoped to.
* - `{ kind: "session", sessionId }` routes to a literal session id.
*
* Failure modes:
* - All `fromStages` resolved zero findings → still delivers an "empty"
* payload; the target may want to know "the upstream stage ran and found
* nothing".
* - `sendToSession` throws → returns `{ status: "failed" }` with the
* underlying error message.
*/
import type {
Artifact,
ArtifactInput,
BuiltinTaskContext,
RunId,
Stage,
StageRunId,
} from "../types.js";
export interface BuiltinRunInput {
runId: RunId;
stageRunId: StageRunId;
stage: Stage;
/** Loop counter, surfaced in delivered payloads for traceability. */
loopRound?: number;
ctx: BuiltinTaskContext;
}
export type BuiltinOutcome =
| { status: "completed"; artifacts: ArtifactInput[] }
| { status: "failed"; errorMessage: string };
export interface BuiltinExecutor {
run(input: BuiltinRunInput): Promise<BuiltinOutcome>;
}
export function createBuiltinRouterExecutor(): BuiltinExecutor {
return {
async run(input) {
const stage = input.stage;
if (stage.executor.kind !== "builtin/router") {
return {
status: "failed",
errorMessage: `builtin/router executor cannot run stage "${stage.name}" with executor.kind=${stage.executor.kind}`,
};
}
const ctx = input.ctx;
const executor = stage.executor;
const targetSessionId =
executor.target.kind === "self" ? ctx.sessionId : executor.target.sessionId;
const bundles: Array<{ stage: string; count: number }> = [];
const payloadSections: string[] = [];
for (const upstream of executor.fromStages) {
const artifacts = await ctx.readSiblingArtifacts(upstream);
bundles.push({ stage: upstream, count: artifacts.length });
payloadSections.push(formatStageSection(upstream, artifacts));
}
const payload = formatRouterPayload({
pipelineName: ctx.pipelineName,
stageName: stage.name,
loopRound: input.loopRound,
sections: payloadSections,
});
try {
await ctx.sendToSession(targetSessionId, payload);
} catch (err) {
return {
status: "failed",
errorMessage: `builtin/router failed to deliver to session "${targetSessionId}": ${
err instanceof Error ? err.message : String(err)
}`,
};
}
const auditArtifact: ArtifactInput = {
kind: "json",
data: {
builtin: "router",
targetSessionId,
bundles,
},
};
return { status: "completed", artifacts: [auditArtifact] };
},
};
}
function formatStageSection(
stageName: string,
artifacts: ReadonlyArray<Artifact>,
): string {
const header = `## From stage: ${stageName} (${artifacts.length} artifact${
artifacts.length === 1 ? "" : "s"
})`;
if (artifacts.length === 0) return `${header}\n_no findings_`;
const body = artifacts.map((a) => " - " + JSON.stringify(a)).join("\n");
return `${header}\n${body}`;
}
interface RouterPayload {
pipelineName: string;
stageName: string;
loopRound?: number;
sections: string[];
}
function formatRouterPayload(p: RouterPayload): string {
const lines = [
`# Pipeline routing: ${p.pipelineName} → ${p.stageName}`,
...(p.loopRound !== undefined ? [`Loop round: ${p.loopRound}`] : []),
"",
...p.sections,
];
return lines.join("\n");
}

272
packages/core/src/pipeline/executors/command.ts Normal file
View File

@ -0,0 +1,272 @@
/**
* Command executor — shell-based pipeline stages.
*
* A `command` stage is a script. The engine spawns it as a child process,
* waits for it to exit, and parses its stdout as findings. Stages are NOT
* talk-to-able (per locked decision 10): there is no AO session, no
* dashboard chat, no terminal attach. Use the agent executor when you need
* an interactive collaborator.
*
* Contract:
* - The command is invoked via `spawn(command, args, { cwd, env })`.
* - `cwd` is resolved against the run's workspace root (defaults to the root
* itself when the stage doesn't set one).
* - The command must exit `0` to be considered successful. Non-zero exit
* codes — and unexpected I/O errors — surface as `StageOutcome.failed`.
* - The stage's stdout is parsed as JSONL ArtifactInput records, the same
* format as the agent executor's findings file. Whitespace-only stdout
* yields zero artifacts; an empty stdout is treated identically.
* - The stage's stderr is captured into the error message on failure but
* never parsed as findings.
*
* Fork-PR safety: if the run's triggering PR is from a fork (the engine
* threads `isFromFork: true` into the start input), the executor refuses to
* run unless the stage opts in with `Stage.allowFork: true`. The refusal is
* 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.
*/
import { spawn } from "node:child_process";
import { join } from "node:path";
import {
type ArtifactInput,
type RunId,
type Stage,
type StageRunId,
} from "../types.js";
import { coerceArtifactInput, parseFindingsJsonl } from "./findings-parser.js";
/**
* Inputs the engine passes when starting a command stage. Mirrors the agent
* executor's StartStageInput but adds `isFromFork` — the PR fork bit is the
* only piece of SCM state the command executor needs.
*/
export interface CommandStartInput {
pipelineName: string;
runId: RunId;
stageRunId: StageRunId;
stage: Stage;
/**
* Root the command runs in. Stages that set `executor.cwd` resolve against
* this root. When unset, the executor uses `process.cwd()` so unit tests
* can call the executor without threading a workspace through.
*/
workspaceRoot?: string;
/**
* True when the triggering PR is from a fork. The executor refuses to run
* unless the stage sets `allowFork: true`. Defaults to `false` for
* non-PR runs (manual triggers, internal pipelines).
*/
isFromFork?: boolean;
/** Loop counter, surfaced as `AO_LOOP_ROUND` in the child env. */
loopRound?: number;
}
export type CommandOutcome =
| { status: "completed"; artifacts: ArtifactInput[] }
| { status: "failed"; errorMessage: string; refused?: boolean };
export interface CommandStageExecutor {
run(input: CommandStartInput): Promise<CommandOutcome>;
}
/** Default millisecond cap on a single command stage. Engine override TBD. */
export const DEFAULT_COMMAND_TIMEOUT_MS = 10 * 60 * 1000;
/** Default cap on stdout bytes; protects the engine from misbehaving scripts. */
export const DEFAULT_COMMAND_STDOUT_CAP_BYTES = 4 * 1024 * 1024;
/** Format the fork-refusal error message — exposed so engine logs match. */
export function formatForkRefusalMessage(stageName: string): string {
return (
`command stage "${stageName}" refused to run on a fork PR — ` +
`set stage.allowFork=true to opt in (only safe for trusted scripts).`
);
}
export interface CommandExecutorDeps {
/**
* Logger hook for refusals. Engine wires this to the observation effect
* stream so the refusal appears in pipeline logs. Default: no-op.
*/
onRefuse?(stage: Stage, message: string): void;
/** Override clock for tests. */
now?(): number;
/** Process spawner — defaults to node:child_process.spawn. Override for tests. */
spawnFn?: typeof spawn;
}
export function createCommandExecutor(deps: CommandExecutorDeps = {}): CommandStageExecutor {
const onRefuse = deps.onRefuse ?? (() => undefined);
const spawnFn = deps.spawnFn ?? spawn;
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}`,
});
}
if (input.isFromFork && stage.allowFork !== true) {
const message = formatForkRefusalMessage(stage.name);
onRefuse(stage, message);
return Promise.resolve({ status: "failed", errorMessage: message, refused: true });
}
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;
return new Promise<CommandOutcome>((resolve) => {
let child;
try {
child = spawnFn(executor.command, executor.args ?? [], {
cwd,
env,
stdio: ["ignore", "pipe", "pipe"],
});
} catch (err) {
resolve({
status: "failed",
errorMessage: `failed to spawn command "${executor.command}": ${
err instanceof Error ? err.message : String(err)
}`,
});
return;
}
const stdoutChunks: Buffer[] = [];
const stderrChunks: Buffer[] = [];
let stdoutBytes = 0;
let truncated = false;
let settled = false;
const settle = (outcome: CommandOutcome) => {
if (settled) return;
settled = true;
resolve(outcome);
};
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) => {
stderrChunks.push(chunk);
});
child.on("error", (err) => {
settle({
status: "failed",
errorMessage: `command "${executor.command}" failed: ${err.message}`,
});
});
child.on("close", (code, signal) => {
const stderr = Buffer.concat(stderrChunks).toString("utf-8").trim();
if (truncated) {
settle({
status: "failed",
errorMessage: `command "${executor.command}" stdout exceeded ${stdoutCap} bytes`,
});
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 });
});
});
},
};
}
function buildChildEnv(
input: CommandStartInput,
overrides: Record<string, string> | undefined,
): NodeJS.ProcessEnv {
const base: NodeJS.ProcessEnv = { ...process.env };
base["AO_PIPELINE_NAME"] = input.pipelineName;
base["AO_PIPELINE_RUN_ID"] = String(input.runId);
base["AO_PIPELINE_STAGE_NAME"] = input.stage.name;
base["AO_PIPELINE_STAGE_RUN_ID"] = String(input.stageRunId);
if (input.loopRound !== undefined) {
base["AO_PIPELINE_LOOP_ROUND"] = String(input.loopRound);
}
if (overrides) {
for (const [k, v] of Object.entries(overrides)) base[k] = v;
}
return base;
}
/**
* Parse stdout into ArtifactInput records.
*
* Stdout may be either:
* - JSONL (one JSON object per line) — the same shape produced by agent
* stages' findings file. Recommended for streaming output.
* - A single JSON array of ArtifactInput records.
* - Empty / whitespace-only — yields zero artifacts.
*
* Anything else (a bare JSON object, partial line, comma-separated values)
* throws — operators get a precise line number rather than silent data loss.
*/
function parseStdoutFindings(stdout: string): ArtifactInput[] {
const trimmed = stdout.trim();
if (!trimmed) return [];
if (trimmed.startsWith("[")) {
let parsed: unknown;
try {
parsed = JSON.parse(trimmed);
} catch (err) {
throw new Error(
`stdout JSON array failed to parse: ${err instanceof Error ? err.message : String(err)}`,
{ cause: err },
);
}
if (!Array.isArray(parsed)) {
throw new Error("stdout JSON did not produce an array");
}
return parsed.map((entry, idx) => coerceArtifactInput(entry, idx + 1));
}
return parseFindingsJsonl(stdout);
}

108
packages/core/src/pipeline/executors/findings-parser.ts Normal file
View File

@ -0,0 +1,108 @@
/**
* Shared findings parser for the agent and command executors.
*
* Both executors collect ArtifactInput records produced by their stage —
* agents drop them in a JSONL file, commands print them to stdout. The
* validation rules (kind discrimination, required fields, severity enum,
* confidence range) are identical, so they live here.
*/
import type { ArtifactInput } from "../types.js";
const VALID_SEVERITIES = ["error", "warning", "info"] as const;
/**
* Parse a JSONL body into ArtifactInput records.
*
* Empty / whitespace-only lines are skipped. The first bad line throws with
* its 1-based line number so the caller can echo it back to operators.
*/
export function parseFindingsJsonl(body: string): ArtifactInput[] {
const out: ArtifactInput[] = [];
for (const [lineNo, raw] of body.split("\n").entries()) {
const trimmed = raw.trim();
if (!trimmed) continue;
let parsed: unknown;
try {
parsed = JSON.parse(trimmed);
} catch (err) {
throw new Error(`line ${lineNo + 1}: ${err instanceof Error ? err.message : String(err)}`, {
cause: err,
});
}
out.push(coerceArtifactInput(parsed, lineNo + 1));
}
return out;
}
/**
* Coerce one parsed JSON value into an ArtifactInput, validating required
* fields by `kind`. Throws on unknown kinds or missing/out-of-range fields.
*/
export function coerceArtifactInput(value: unknown, lineNo: number): ArtifactInput {
if (!value || typeof value !== "object") {
throw new Error(`line ${lineNo}: expected object`);
}
const obj = value as Record<string, unknown>;
if (obj["kind"] === "finding") {
requireString(obj, "filePath", lineNo);
requireNumber(obj, "startLine", lineNo);
requireNumber(obj, "endLine", lineNo);
requireString(obj, "title", lineNo);
requireString(obj, "description", lineNo);
requireString(obj, "category", lineNo);
requireEnum(obj, "severity", VALID_SEVERITIES, lineNo);
requireNumberInRange(obj, "confidence", 0, 1, lineNo);
return obj as unknown as ArtifactInput;
}
if (obj["kind"] === "json") {
if (!obj["data"] || typeof obj["data"] !== "object") {
throw new Error(`line ${lineNo}: "json" artifact requires object \`data\``);
}
return obj as unknown as ArtifactInput;
}
throw new Error(`line ${lineNo}: unknown artifact kind=${JSON.stringify(obj["kind"])}`);
}
function requireString(obj: Record<string, unknown>, key: string, lineNo: number): void {
if (typeof obj[key] !== "string") {
throw new Error(`line ${lineNo}: missing string field "${key}"`);
}
}
function requireNumber(obj: Record<string, unknown>, key: string, lineNo: number): void {
if (typeof obj[key] !== "number") {
throw new Error(`line ${lineNo}: missing numeric field "${key}"`);
}
}
function requireNumberInRange(
obj: Record<string, unknown>,
key: string,
min: number,
max: number,
lineNo: number,
): void {
const value = obj[key];
if (typeof value !== "number" || value < min || value > max) {
throw new Error(
`line ${lineNo}: field "${key}" must be a number in [${min}, ${max}], got ${JSON.stringify(value)}`,
);
}
}
function requireEnum<T extends string>(
obj: Record<string, unknown>,
key: string,
allowed: readonly T[],
lineNo: number,
): void {
const value = obj[key];
if (typeof value !== "string" || !(allowed as readonly string[]).includes(value)) {
throw new Error(
`line ${lineNo}: field "${key}" must be one of ${allowed
.map((v) => `"${v}"`)
.join(", ")}, got ${JSON.stringify(value)}`,
);
}
}

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

@ -8,3 +8,25 @@ export {
type StageOutcome,
type StartStageInput,
} from "./agent.js";
export {
createCommandExecutor,
formatForkRefusalMessage,
DEFAULT_COMMAND_TIMEOUT_MS,
DEFAULT_COMMAND_STDOUT_CAP_BYTES,
type CommandExecutorDeps,
type CommandStageExecutor,
type CommandStartInput,
type CommandOutcome,
} from "./command.js";
export { parseFindingsJsonl, coerceArtifactInput } from "./findings-parser.js";
export {
createBuiltinRouterExecutor,
type BuiltinExecutor,
type BuiltinRunInput,
type BuiltinOutcome,
} from "./builtin-router.js";
export { createBuiltinComposeExecutor, type ComposeRunInput } from "./builtin-compose.js";

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

@ -39,6 +39,22 @@ export {
type RunningAgentStage,
type StageOutcome,
type StartStageInput,
createCommandExecutor,
formatForkRefusalMessage,
DEFAULT_COMMAND_TIMEOUT_MS,
DEFAULT_COMMAND_STDOUT_CAP_BYTES,
type CommandExecutorDeps,
type CommandStageExecutor,
type CommandStartInput,
type CommandOutcome,
parseFindingsJsonl,
coerceArtifactInput,
createBuiltinRouterExecutor,
createBuiltinComposeExecutor,
type BuiltinExecutor,
type BuiltinRunInput,
type BuiltinOutcome,
type ComposeRunInput,
} from "./executors/index.js";
export {

91
packages/core/src/pipeline/types.ts
View File

@ -64,7 +64,47 @@ export interface CommandExecutor {
cwd?: string;
}
export type StageExecutor = AgentExecutor | CommandExecutor;
/**
* Routes findings from upstream stages into an existing AO session via
* {@link BuiltinTaskContext.sendToSession}. Replaces the original spec's
* `SEND_TO_AGENT` reducer command — the routing decision is now a stage with
* its own state, retries, and DAG position.
*/
export interface BuiltinRouterExecutor {
kind: "builtin/router";
/**
* The upstream stage names whose findings should be delivered. Must be a
* subset of this stage's `dependsOn` so the scheduler has already
* finalized them when the router runs.
*/
fromStages: string[];
/**
* Target session resolution. Either a literal sessionId or a sentinel
* keyword the engine resolves at run time. v1.2 only supports `"self"`
* (the session this pipeline run is scoped to) — additional resolvers
* land alongside cross-session orchestration.
*/
target: { kind: "session"; sessionId: string } | { kind: "self" };
}
/**
* Bundles findings from multiple upstream stages into a single composite
* findings artifact for a downstream stage to consume. The composite is
* emitted as a `json` artifact whose `data.findings` is the merged list,
* tagged with the originating stage name.
*/
export interface BuiltinComposeExecutor {
kind: "builtin/compose";
/**
* The upstream stage names whose findings should be merged. Must be a
* subset of this stage's `dependsOn`.
*/
fromStages: string[];
}
export type BuiltinExecutor = BuiltinRouterExecutor | BuiltinComposeExecutor;
export type StageExecutor = AgentExecutor | CommandExecutor | BuiltinExecutor;
export interface TaskSpec {
/** Prompt text injected into the spawned agent session, or main script body for command. */
@ -139,6 +179,14 @@ export interface Stage {
* "all `dependsOn` stages must have succeeded".
*/
routes?: StageRoutes;
/**
* Opt-in fork-PR safety override for `command` stages. The command executor
* refuses to run a stage whose triggering PR is from a fork unless this is
* set to `true`. Agent and builtin stages ignore the flag — agent stages
* run in their own sandboxed sessions, and builtins are pure functions over
* findings. Default `false` (refuse on forks).
*/
allowFork?: boolean;
}
export interface Pipeline {
@ -322,3 +370,44 @@ export function emptyEngineState(): EngineState {
historySummaries: {},
};
}
// ============================================================================
// Builtin task context
// ============================================================================
/**
* Capabilities surface a `builtin/*` executor sees at run time. The engine
* constructs a fresh context per stage invocation — the implementation is the
* only place that touches the pipeline store and session manager, so builtin
* executors stay testable as pure functions over their context.
*
* v1.2 covers the three capabilities the spec calls out:
* - read findings from sibling stages (router + compose)
* - send a payload to a target session (router)
* - write artifacts back into the pipeline store (compose)
*
* The context is intentionally narrow: builtins must not need access to the
* full SessionManager or PipelineStore. If a builtin needs a capability not
* exposed here, extend this interface rather than passing the underlying
* dependency through.
*/
export interface BuiltinTaskContext {
/** Identity of the stage currently executing. */
runId: RunId;
stageRunId: StageRunId;
stageName: string;
/** Pipeline run scope, for routing and downstream lookups. */
sessionId: string;
pipelineName: string;
/**
* Return artifacts emitted by an upstream sibling stage in the same run.
* Returns `[]` when the sibling has no artifacts or has not run.
*/
readSiblingArtifacts(stageName: string): Promise<Artifact[]>;
/**
* Deliver a payload to a target session. Implementations route through
* `SessionManager.send()`. The payload is the literal message body the
* target session receives.
*/
sendToSession(targetSessionId: string, payload: string): Promise<void>;
}

7
packages/core/src/types.ts
View File

@ -947,6 +947,13 @@ export interface PRInfo {
branch: string;
baseBranch: string;
isDraft: boolean;
/**
* True when the PR head is in a different repository than the base
* (i.e. opened from a fork). Used by the pipeline command executor to
* refuse running untrusted scripts on fork PRs unless the stage opts in
* via `Stage.allowFork: true`.
*/
isFromFork: boolean;
}
export type PRState = "open" | "merged" | "closed";

3
packages/core/src/utils/session-from-metadata.ts
View File

@ -83,6 +83,9 @@ export function sessionFromMetadata(
branch: meta["branch"] ?? "",
baseBranch: "",
isDraft: prIsDraft,
// Metadata reconstruction can't know whether the PR is from a
// fork; fork-aware behavior is gated on a fresh SCM probe.
isFromFork: false,
};
})()
: null,

8
packages/plugins/scm-github/src/index.ts
View File

@ -112,6 +112,7 @@ function prInfoFromView(
headRefName: string;
baseRefName: string;
isDraft: boolean;
isCrossRepository?: boolean;
},
projectRepo: string,
): PRInfo {
@ -126,6 +127,7 @@ function prInfoFromView(
branch: data.headRefName,
baseBranch: data.baseRefName,
isDraft: data.isDraft,
isFromFork: data.isCrossRepository ?? false,
};
}
@ -701,7 +703,7 @@ function createGitHubSCM(): SCM {
"--head",
session.branch,
"--json",
"number,url,title,headRefName,baseRefName,isDraft",
"number,url,title,headRefName,baseRefName,isDraft,isCrossRepository",
"--limit",
"1",
]);
@ -713,6 +715,7 @@ function createGitHubSCM(): SCM {
headRefName: string;
baseRefName: string;
isDraft: boolean;
isCrossRepository?: boolean;
}> = JSON.parse(raw);
if (prs.length === 0) return null;
@ -742,7 +745,7 @@ function createGitHubSCM(): SCM {
"--repo",
repo,
"--json",
"number,url,title,headRefName,baseRefName,isDraft",
"number,url,title,headRefName,baseRefName,isDraft,isCrossRepository",
]);
const data: {
@ -752,6 +755,7 @@ function createGitHubSCM(): SCM {
headRefName: string;
baseRefName: string;
isDraft: boolean;
isCrossRepository?: boolean;
} = JSON.parse(raw);
return prInfoFromView(data, repo);

2
packages/plugins/scm-github/test/index.test.ts
View File

@ -30,6 +30,7 @@ const pr: PRInfo = {
branch: "feat/my-feature",
baseBranch: "main",
isDraft: false,
isFromFork: false,
};
const project: ProjectConfig = {
@ -413,6 +414,7 @@ describe("scm-github plugin", () => {
branch: "feat/my-feature",
baseBranch: "main",
isDraft: false,
isFromFork: false,
});
});

6
packages/plugins/scm-gitlab/src/index.ts
View File

@ -467,6 +467,8 @@ function createGitLabSCM(config?: Record<string, unknown>): SCM {
source_branch: string;
target_branch: string;
draft: boolean;
source_project_id?: number;
target_project_id?: number;
}>
>(raw, `detectPR for branch "${session.branch}"`);
@ -482,6 +484,10 @@ function createGitLabSCM(config?: Record<string, unknown>): SCM {
branch: mr.source_branch,
baseBranch: mr.target_branch,
isDraft: mr.draft ?? false,
isFromFork:
mr.source_project_id !== undefined &&
mr.target_project_id !== undefined &&
mr.source_project_id !== mr.target_project_id,
};
} catch (err) {
console.warn(`detectPR: failed for branch "${session.branch}": ${(err as Error).message}`);

2
packages/plugins/scm-gitlab/test/index.test.ts
View File

@ -28,6 +28,7 @@ const pr: PRInfo = {
branch: "feat/my-feature",
baseBranch: "main",
isDraft: false,
isFromFork: false,
};
const project: ProjectConfig = {
@ -341,6 +342,7 @@ describe("scm-gitlab plugin", () => {
branch: "feat/my-feature",
baseBranch: "main",
isDraft: false,
isFromFork: false,
});
});

2
packages/web/src/lib/scm-webhooks.test.ts
View File

@ -111,6 +111,7 @@ describe("findAffectedSessions", () => {
branch: "feat/one",
baseBranch: "main",
isDraft: false,
isFromFork: false,
},
workspacePath: null,
runtimeHandle: null,
@ -141,6 +142,7 @@ describe("findAffectedSessions", () => {
branch: "feat/one",
baseBranch: "main",
isDraft: false,
isFromFork: false,
},
workspacePath: null,
runtimeHandle: null,