diff --git a/packages/core/src/__tests__/pipeline-builtin-compose.test.ts b/packages/core/src/__tests__/pipeline-builtin-compose.test.ts
new file mode 100644
index 000000000..771d3afd8
--- /dev/null
+++ b/packages/core/src/__tests__/pipeline-builtin-compose.test.ts
@@ -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");
+  });
+});
diff --git a/packages/core/src/__tests__/pipeline-builtin-router.test.ts b/packages/core/src/__tests__/pipeline-builtin-router.test.ts
new file mode 100644
index 000000000..de9b5aaa6
--- /dev/null
+++ b/packages/core/src/__tests__/pipeline-builtin-router.test.ts
@@ -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");
+  });
+});
diff --git a/packages/core/src/__tests__/pipeline-command-executor.test.ts b/packages/core/src/__tests__/pipeline-command-executor.test.ts
new file mode 100644
index 000000000..b16914193
--- /dev/null
+++ b/packages/core/src/__tests__/pipeline-command-executor.test.ts
@@ -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" } });
+  });
+});
diff --git a/packages/core/src/__tests__/pipeline-engine-executors.test.ts b/packages/core/src/__tests__/pipeline-engine-executors.test.ts
new file mode 100644
index 000000000..504bd56e3
--- /dev/null
+++ b/packages/core/src/__tests__/pipeline-engine-executors.test.ts
@@ -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");
+  });
+});
diff --git a/packages/core/src/__tests__/pipeline-engine.test.ts b/packages/core/src/__tests__/pipeline-engine.test.ts
index 0f4fdd7f0..ec26b86bf 100644
--- a/packages/core/src/__tests__/pipeline-engine.test.ts
+++ b/packages/core/src/__tests__/pipeline-engine.test.ts
@@ -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);
   });
 
diff --git a/packages/core/src/__tests__/session-manager/claim-pr.test.ts b/packages/core/src/__tests__/session-manager/claim-pr.test.ts
index 3625a0c1e..38d2a4978 100644
--- a/packages/core/src/__tests__/session-manager/claim-pr.test.ts
+++ b/packages/core/src/__tests__/session-manager/claim-pr.test.ts
@@ -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),
     });
diff --git a/packages/core/src/__tests__/test-utils.ts b/packages/core/src/__tests__/test-utils.ts
index e7a51e9df..286a90af6 100644
--- a/packages/core/src/__tests__/test-utils.ts
+++ b/packages/core/src/__tests__/test-utils.ts
@@ -129,6 +129,7 @@ export function makePR(overrides: Partial<PRInfo> = {}): PRInfo {
     branch: "feat/test",
     baseBranch: "main",
     isDraft: false,
+    isFromFork: false,
     ...overrides,
   };
 }
diff --git a/packages/core/src/pipeline/config-schema.ts b/packages/core/src/pipeline/config-schema.ts
index 0f09c1331..80461f56c 100644
--- a/packages/core/src/pipeline/config-schema.ts
+++ b/packages/core/src/pipeline/config-schema.ts
@@ -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)}`);
+    }
+  }
+}
diff --git a/packages/core/src/pipeline/engine.ts b/packages/core/src/pipeline/engine.ts
index 416fe97c4..c3e812437 100644
--- a/packages/core/src/pipeline/engine.ts
+++ b/packages/core/src/pipeline/engine.ts
@@ -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(() =>
diff --git a/packages/core/src/pipeline/executors/agent.ts b/packages/core/src/pipeline/executors/agent.ts
index 1c47bdc64..ec0414404 100644
--- a/packages/core/src/pipeline/executors/agent.ts
+++ b/packages/core/src/pipeline/executors/agent.ts
@@ -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> {
diff --git a/packages/core/src/pipeline/executors/builtin-compose.ts b/packages/core/src/pipeline/executors/builtin-compose.ts
new file mode 100644
index 000000000..cc36d44bf
--- /dev/null
+++ b/packages/core/src/pipeline/executors/builtin-compose.ts
@@ -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] };
+    },
+  };
+}
diff --git a/packages/core/src/pipeline/executors/builtin-router.ts b/packages/core/src/pipeline/executors/builtin-router.ts
new file mode 100644
index 000000000..4fbec451f
--- /dev/null
+++ b/packages/core/src/pipeline/executors/builtin-router.ts
@@ -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");
+}
diff --git a/packages/core/src/pipeline/executors/command.ts b/packages/core/src/pipeline/executors/command.ts
new file mode 100644
index 000000000..267f72365
--- /dev/null
+++ b/packages/core/src/pipeline/executors/command.ts
@@ -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);
+}
diff --git a/packages/core/src/pipeline/executors/findings-parser.ts b/packages/core/src/pipeline/executors/findings-parser.ts
new file mode 100644
index 000000000..a554ae118
--- /dev/null
+++ b/packages/core/src/pipeline/executors/findings-parser.ts
@@ -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)}`,
+    );
+  }
+}
diff --git a/packages/core/src/pipeline/executors/index.ts b/packages/core/src/pipeline/executors/index.ts
index 8ab6f0b45..6effefe1c 100644
--- a/packages/core/src/pipeline/executors/index.ts
+++ b/packages/core/src/pipeline/executors/index.ts
@@ -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";
diff --git a/packages/core/src/pipeline/index.ts b/packages/core/src/pipeline/index.ts
index 3876da485..7b76ed2c5 100644
--- a/packages/core/src/pipeline/index.ts
+++ b/packages/core/src/pipeline/index.ts
@@ -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 {
diff --git a/packages/core/src/pipeline/types.ts b/packages/core/src/pipeline/types.ts
index 4858712c5..f743491b1 100644
--- a/packages/core/src/pipeline/types.ts
+++ b/packages/core/src/pipeline/types.ts
@@ -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>;
+}
diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts
index d4d31415c..a74ef5b14 100644
--- a/packages/core/src/types.ts
+++ b/packages/core/src/types.ts
@@ -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";
diff --git a/packages/core/src/utils/session-from-metadata.ts b/packages/core/src/utils/session-from-metadata.ts
index aeda52d79..d08c9308d 100644
--- a/packages/core/src/utils/session-from-metadata.ts
+++ b/packages/core/src/utils/session-from-metadata.ts
@@ -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,
diff --git a/packages/plugins/scm-github/src/index.ts b/packages/plugins/scm-github/src/index.ts
index 8d065db01..fb712feae 100644
--- a/packages/plugins/scm-github/src/index.ts
+++ b/packages/plugins/scm-github/src/index.ts
@@ -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);
diff --git a/packages/plugins/scm-github/test/index.test.ts b/packages/plugins/scm-github/test/index.test.ts
index 501ff4e09..19f1a5927 100644
--- a/packages/plugins/scm-github/test/index.test.ts
+++ b/packages/plugins/scm-github/test/index.test.ts
@@ -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,
       });
     });
 
diff --git a/packages/plugins/scm-gitlab/src/index.ts b/packages/plugins/scm-gitlab/src/index.ts
index df9511244..016e65396 100644
--- a/packages/plugins/scm-gitlab/src/index.ts
+++ b/packages/plugins/scm-gitlab/src/index.ts
@@ -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}`);
diff --git a/packages/plugins/scm-gitlab/test/index.test.ts b/packages/plugins/scm-gitlab/test/index.test.ts
index 9dad4d0eb..994c88ff8 100644
--- a/packages/plugins/scm-gitlab/test/index.test.ts
+++ b/packages/plugins/scm-gitlab/test/index.test.ts
@@ -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,
       });
     });
 
diff --git a/packages/web/src/lib/scm-webhooks.test.ts b/packages/web/src/lib/scm-webhooks.test.ts
index c01e42516..e8382fa61 100644
--- a/packages/web/src/lib/scm-webhooks.test.ts
+++ b/packages/web/src/lib/scm-webhooks.test.ts
@@ -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,