diff --git a/packages/core/src/__tests__/pipeline-config-diff.test.ts b/packages/core/src/__tests__/pipeline-config-diff.test.ts
new file mode 100644
index 000000000..0baeecbcc
--- /dev/null
+++ b/packages/core/src/__tests__/pipeline-config-diff.test.ts
@@ -0,0 +1,182 @@
+/**
+ * v1.3 — CONFIG_CHANGED classification.
+ *
+ * Verifies the structural vs tuning split: drivers can pick `tuning` changes
+ * up live, but `structural` changes must terminate the active run via
+ * CONFIG_CHANGED. The reducer's existing CONFIG_CHANGED handler is unchanged
+ * — it terminates regardless. The classifier decides whether to dispatch.
+ */
+
+import { describe, expect, it } from "vitest";
+
+import {
+  asPipelineId,
+  classifyConfigChange,
+  type Pipeline,
+  type Predicate,
+  type Stage,
+} from "../pipeline/index.js";
+
+function stage(name: string, overrides: Partial<Stage> = {}): Stage {
+  return {
+    name,
+    trigger: { on: ["pr.opened"] },
+    executor: { kind: "agent", plugin: "codex", mode: "review" },
+    task: { prompt: `run ${name}` },
+    ...overrides,
+  };
+}
+
+function pipeline(stages: Stage[], overrides: Partial<Pipeline> = {}): Pipeline {
+  return {
+    id: asPipelineId("pl-1"),
+    name: "default",
+    stages,
+    maxConcurrentStages: 1,
+    ...overrides,
+  };
+}
+
+describe("classifyConfigChange — none", () => {
+  it("returns none when nothing changed", () => {
+    const p = pipeline([stage("review")]);
+    expect(classifyConfigChange(p, p)).toEqual({ kind: "none", reasons: [] });
+  });
+
+  it("deep-equals nested objects (executor.config)", () => {
+    const a = pipeline([
+      stage("review", {
+        executor: {
+          kind: "agent",
+          plugin: "codex",
+          mode: "review",
+          config: { tone: "strict" },
+        },
+      }),
+    ]);
+    const b = pipeline([
+      stage("review", {
+        executor: {
+          kind: "agent",
+          plugin: "codex",
+          mode: "review",
+          config: { tone: "strict" },
+        },
+      }),
+    ]);
+    expect(classifyConfigChange(a, b).kind).toBe("none");
+  });
+});
+
+describe("classifyConfigChange — structural", () => {
+  it("renamed stage is structural", () => {
+    expect(classifyConfigChange(pipeline([stage("review")]), pipeline([stage("audit")])).kind).toBe(
+      "structural",
+    );
+  });
+
+  it("added stage is structural", () => {
+    const before = pipeline([stage("review")]);
+    const after = pipeline([stage("review"), stage("fix")]);
+    expect(classifyConfigChange(before, after).kind).toBe("structural");
+  });
+
+  it("removed stage is structural", () => {
+    const before = pipeline([stage("review"), stage("fix")]);
+    const after = pipeline([stage("review")]);
+    expect(classifyConfigChange(before, after).kind).toBe("structural");
+  });
+
+  it("reordered stages is structural (DAG slot priority depends on order)", () => {
+    const before = pipeline([stage("review"), stage("fix")]);
+    const after = pipeline([stage("fix"), stage("review")]);
+    expect(classifyConfigChange(before, after).kind).toBe("structural");
+  });
+
+  it("changed dependsOn is structural", () => {
+    const before = pipeline([stage("review"), stage("fix", { dependsOn: ["review"] })]);
+    const after = pipeline([stage("review"), stage("fix")]);
+    expect(classifyConfigChange(before, after).kind).toBe("structural");
+  });
+
+  it("changed executor plugin is structural", () => {
+    const before = pipeline([
+      stage("review", { executor: { kind: "agent", plugin: "codex", mode: "review" } }),
+    ]);
+    const after = pipeline([
+      stage("review", { executor: { kind: "agent", plugin: "aider", mode: "review" } }),
+    ]);
+    expect(classifyConfigChange(before, after).kind).toBe("structural");
+  });
+
+  it("changed routes is structural", () => {
+    const before = pipeline([stage("review"), stage("fix")]);
+    const after = pipeline([
+      stage("review"),
+      stage("fix", {
+        routes: { when: { kind: "all_pass", stages: ["review"] } as Predicate },
+      }),
+    ]);
+    expect(classifyConfigChange(before, after).kind).toBe("structural");
+  });
+
+  it("pipeline rename is structural", () => {
+    const before = pipeline([stage("review")], { name: "ci" });
+    const after = pipeline([stage("review")], { name: "ci-v2" });
+    expect(classifyConfigChange(before, after).kind).toBe("structural");
+  });
+});
+
+describe("classifyConfigChange — tuning", () => {
+  it("retries change is tuning", () => {
+    const before = pipeline([stage("review")]);
+    const after = pipeline([stage("review", { retries: 3 })]);
+    expect(classifyConfigChange(before, after).kind).toBe("tuning");
+  });
+
+  it("budget tweak is tuning", () => {
+    const before = pipeline([stage("review", { budget: { maxUsd: 1 } })]);
+    const after = pipeline([stage("review", { budget: { maxUsd: 5 } })]);
+    expect(classifyConfigChange(before, after).kind).toBe("tuning");
+  });
+
+  it("exitPredicates change is tuning (threshold edits)", () => {
+    const before = pipeline([stage("review")], {
+      exitPredicates: [{ kind: "finding_count_below", n: 3 }],
+    });
+    const after = pipeline([stage("review")], {
+      exitPredicates: [{ kind: "finding_count_below", n: 5 }],
+    });
+    expect(classifyConfigChange(before, after).kind).toBe("tuning");
+  });
+
+  it("maxLoopRounds bump is tuning", () => {
+    const before = pipeline([stage("review")], { maxLoopRounds: 3 });
+    const after = pipeline([stage("review")], { maxLoopRounds: 5 });
+    expect(classifyConfigChange(before, after).kind).toBe("tuning");
+  });
+
+  it("maxConcurrentStages bump is tuning", () => {
+    const before = pipeline([stage("review")], { maxConcurrentStages: 1 });
+    const after = pipeline([stage("review")], { maxConcurrentStages: 4 });
+    expect(classifyConfigChange(before, after).kind).toBe("tuning");
+  });
+
+  it("workspaceClass annotation is tuning", () => {
+    const before = pipeline([stage("review"), stage("fix")]);
+    const after = pipeline([stage("review"), stage("fix", { workspaceClass: "read-siblings" })]);
+    expect(classifyConfigChange(before, after).kind).toBe("tuning");
+  });
+});
+
+describe("classifyConfigChange — promotion semantics", () => {
+  it("any structural change wins over concurrent tuning changes", () => {
+    const before = pipeline([stage("review", { retries: 2 })]);
+    const after = pipeline([stage("audit", { retries: 5 })]);
+    const result = classifyConfigChange(before, after);
+    expect(result.kind).toBe("structural");
+    // The dominant reason is the structural one — the tuning reasons may or
+    // may not appear depending on early-exit; only assert the structural one.
+    expect(result.reasons[0]).toMatch(/stage list changed/);
+  });
+});
diff --git a/packages/core/src/__tests__/pipeline-exit-predicates.test.ts b/packages/core/src/__tests__/pipeline-exit-predicates.test.ts
new file mode 100644
index 000000000..bd0e7f396
--- /dev/null
+++ b/packages/core/src/__tests__/pipeline-exit-predicates.test.ts
@@ -0,0 +1,249 @@
+/**
+ * v1.3 — Pipeline.exitPredicates integration with the reducer.
+ *
+ * Verifies that a fully-terminal run reaches:
+ *  - `loopState: done`              when exitPredicates evaluate true
+ *  - `loopState: stalled`           when exitPredicates evaluate false and
+ *                                    `maxLoopRounds` is exhausted (or unset)
+ *  - `loopState: awaiting_context`  when exitPredicates evaluate false but
+ *                                    rounds remain (loop awaits next trigger)
+ *
+ * Also verifies the legacy path: when `exitPredicates` is unset, behavior
+ * matches v1.1 (every terminal run → `done`).
+ */
+
+import { describe, expect, it } from "vitest";
+
+import {
+  asPipelineId,
+  asRunId,
+  asStageRunId,
+  emptyEngineState,
+  reduce,
+  type Pipeline,
+  type PipelineEvent,
+  type Predicate,
+  type Stage,
+} from "../pipeline/index.js";
+
+const NOW = 1_700_000_000_000;
+
+function makeStage(name: string, overrides: Partial<Stage> = {}): Stage {
+  return {
+    name,
+    trigger: { on: ["pr.opened"] },
+    executor: { kind: "agent", plugin: "codex", mode: "review" },
+    task: { prompt: `run ${name}` },
+    ...overrides,
+  };
+}
+
+function makePipeline(
+  stages: Stage[],
+  exitPredicates?: Predicate[],
+  maxLoopRounds?: number,
+): Pipeline {
+  return {
+    id: asPipelineId("pl-1"),
+    name: "default",
+    stages,
+    maxConcurrentStages: 1,
+    ...(exitPredicates !== undefined ? { exitPredicates } : {}),
+    ...(maxLoopRounds !== undefined ? { maxLoopRounds } : {}),
+  };
+}
+
+describe("Pipeline.exitPredicates — terminal loop state mapping", () => {
+  it("loop state = done when predicates are unset (v1.1 default)", () => {
+    const pipeline = makePipeline([makeStage("review")]);
+    const runId = asRunId("run-1");
+    const trigger: PipelineEvent = {
+      type: "TRIGGER_FIRED",
+      now: NOW,
+      trigger: "pr.opened",
+      sessionId: "ses-1",
+      pipeline,
+      headSha: "sha-aaa",
+      runId,
+      stageRunIds: { review: asStageRunId("sr-1") },
+    };
+    let { state } = reduce(emptyEngineState(), trigger);
+    state = reduce(state, {
+      type: "STAGE_STARTED",
+      now: NOW + 1,
+      runId,
+      stageName: "review",
+    }).state;
+    state = reduce(state, {
+      type: "STAGE_COMPLETED",
+      now: NOW + 2,
+      runId,
+      stageName: "review",
+      artifacts: [],
+    }).state;
+    expect(state.runs[runId].loopState).toBe("done");
+    expect(state.runs[runId].terminationReason).toBe("completed");
+  });
+
+  it("loop state = done when exitPredicates evaluate true", () => {
+    const pipeline = makePipeline(
+      [makeStage("review")],
+      [{ kind: "all_pass", stages: ["review"] }, { kind: "no_open_findings" }],
+    );
+    const runId = asRunId("run-1");
+    let { state } = reduce(emptyEngineState(), {
+      type: "TRIGGER_FIRED",
+      now: NOW,
+      trigger: "pr.opened",
+      sessionId: "ses-1",
+      pipeline,
+      headSha: "sha-aaa",
+      runId,
+      stageRunIds: { review: asStageRunId("sr-1") },
+    });
+    state = reduce(state, {
+      type: "STAGE_STARTED",
+      now: NOW + 1,
+      runId,
+      stageName: "review",
+    }).state;
+    state = reduce(state, {
+      type: "STAGE_COMPLETED",
+      now: NOW + 2,
+      runId,
+      stageName: "review",
+      artifacts: [],
+    }).state;
+    expect(state.runs[runId].loopState).toBe("done");
+  });
+
+  it("loop state = stalled when exitPredicates evaluate false and rounds exhausted", () => {
+    const pipeline = makePipeline(
+      [makeStage("review")],
+      // Open finding produced → no_open_findings fails.
+      [{ kind: "no_open_findings" }],
+    );
+    const runId = asRunId("run-1");
+    let { state } = reduce(emptyEngineState(), {
+      type: "TRIGGER_FIRED",
+      now: NOW,
+      trigger: "pr.opened",
+      sessionId: "ses-1",
+      pipeline,
+      headSha: "sha-aaa",
+      runId,
+      stageRunIds: { review: asStageRunId("sr-1") },
+    });
+    state = reduce(state, {
+      type: "STAGE_STARTED",
+      now: NOW + 1,
+      runId,
+      stageName: "review",
+    }).state;
+    const result = reduce(state, {
+      type: "STAGE_COMPLETED",
+      now: NOW + 2,
+      runId,
+      stageName: "review",
+      artifacts: [
+        {
+          kind: "finding",
+          filePath: "x.ts",
+          startLine: 1,
+          endLine: 2,
+          title: "t",
+          description: "d",
+          category: "general",
+          severity: "warning",
+          confidence: 0.9,
+        },
+      ],
+    });
+    expect(result.state.runs[runId].loopState).toBe("stalled");
+    // pipeline.loop.failed observation is emitted in addition to
+    // pipeline.run.terminated.
+    const loopFailed = result.effects.find(
+      (e) => e.type === "EMIT_OBSERVATION" && e.event.name === "pipeline.loop.failed",
+    );
+    expect(loopFailed).toBeDefined();
+  });
+
+  it("loop state = awaiting_context when predicates fail but rounds remain", () => {
+    const pipeline = makePipeline(
+      [makeStage("review")],
+      [{ kind: "no_open_findings" }],
+      // 5 rounds allowed, current loopRounds will be 1 — plenty of headroom.
+      5,
+    );
+    const runId = asRunId("run-1");
+    let { state } = reduce(emptyEngineState(), {
+      type: "TRIGGER_FIRED",
+      now: NOW,
+      trigger: "pr.opened",
+      sessionId: "ses-1",
+      pipeline,
+      headSha: "sha-aaa",
+      runId,
+      stageRunIds: { review: asStageRunId("sr-1") },
+    });
+    state = reduce(state, {
+      type: "STAGE_STARTED",
+      now: NOW + 1,
+      runId,
+      stageName: "review",
+    }).state;
+    state = reduce(state, {
+      type: "STAGE_COMPLETED",
+      now: NOW + 2,
+      runId,
+      stageName: "review",
+      artifacts: [
+        {
+          kind: "finding",
+          filePath: "x.ts",
+          startLine: 1,
+          endLine: 2,
+          title: "t",
+          description: "d",
+          category: "general",
+          severity: "warning",
+          confidence: 0.9,
+        },
+      ],
+    }).state;
+    expect(state.runs[runId].loopState).toBe("awaiting_context");
+    expect(state.runs[runId].terminationReason).toBe("completed");
+  });
+
+  it("emits pipeline.loop.succeeded observation on done", () => {
+    const pipeline = makePipeline([makeStage("review")], [{ kind: "no_open_findings" }]);
+    const runId = asRunId("run-1");
+    let { state } = reduce(emptyEngineState(), {
+      type: "TRIGGER_FIRED",
+      now: NOW,
+      trigger: "pr.opened",
+      sessionId: "ses-1",
+      pipeline,
+      headSha: "sha-aaa",
+      runId,
+      stageRunIds: { review: asStageRunId("sr-1") },
+    });
+    state = reduce(state, {
+      type: "STAGE_STARTED",
+      now: NOW + 1,
+      runId,
+      stageName: "review",
+    }).state;
+    const result = reduce(state, {
+      type: "STAGE_COMPLETED",
+      now: NOW + 2,
+      runId,
+      stageName: "review",
+      artifacts: [],
+    });
+    const succeeded = result.effects.find(
+      (e) => e.type === "EMIT_OBSERVATION" && e.event.name === "pipeline.loop.succeeded",
+    );
+    expect(succeeded).toBeDefined();
+  });
+});
diff --git a/packages/core/src/__tests__/pipeline-predicate.test.ts b/packages/core/src/__tests__/pipeline-predicate.test.ts
new file mode 100644
index 000000000..db900e348
--- /dev/null
+++ b/packages/core/src/__tests__/pipeline-predicate.test.ts
@@ -0,0 +1,353 @@
+/**
+ * v1.3 typed predicate DSL — parser (Zod) + evaluator coverage.
+ *
+ * Verifies:
+ *  - PredicateSchema accepts every leaf + boolean combinator
+ *  - PredicateSchema rejects malformed forms with line-precise errors
+ *  - evaluatePredicate honors leaf semantics over stages + artifacts
+ *  - Legacy `StageRoutePredicate` shapes still parse and translate via
+ *    fromLegacyRoutePredicate
+ *  - collectReferencedStages walks the tree (transitive across and/or/not)
+ */
+
+import { describe, expect, it } from "vitest";
+
+import {
+  asArtifactId,
+  asRunId,
+  asStageRunId,
+  collectReferencedStages,
+  evaluateExitPredicates,
+  evaluatePredicate,
+  fromLegacyRoutePredicate,
+  isLegacyRoutePredicate,
+  normalizeRoutePredicate,
+  predicateDepth,
+  PredicateSchema,
+  type Artifact,
+  type Predicate,
+  type PredicateContext,
+  type StageState,
+} from "../pipeline/index.js";
+
+const NOW_ISO = new Date(1_700_000_000_000).toISOString();
+
+function makeStageState(name: string, overrides: Partial<StageState> = {}): StageState {
+  return {
+    stageRunId: asStageRunId(`sr-${name}`),
+    status: "succeeded",
+    attempt: 1,
+    artifacts: [],
+    ...overrides,
+  };
+}
+
+function makeFinding(
+  stageRunId: string,
+  index: number,
+  overrides: Partial<Artifact> = {},
+): Artifact {
+  return {
+    artifactId: asArtifactId(`${stageRunId}-${index}`),
+    pipelineRunId: asRunId("run-1"),
+    stageRunId: asStageRunId(stageRunId),
+    stageName: "review",
+    kind: "finding",
+    filePath: "src/x.ts",
+    startLine: 1,
+    endLine: 2,
+    title: "t",
+    description: "d",
+    category: "general",
+    severity: "warning",
+    confidence: 0.9,
+    status: "open",
+    createdAt: NOW_ISO,
+    ...overrides,
+  } as Artifact;
+}
+
+function ctx(
+  stages: Record<string, StageState>,
+  artifactsByStage: Record<string, Artifact[]> = {},
+): PredicateContext {
+  return {
+    stages,
+    artifactsByStage,
+    allStageNames: Object.keys(stages),
+  };
+}
+
+describe("PredicateSchema — parser", () => {
+  it("accepts every leaf kind", () => {
+    expect(PredicateSchema.safeParse({ kind: "all_pass", stages: ["a", "b"] }).success).toBe(true);
+    expect(PredicateSchema.safeParse({ kind: "no_open_findings" }).success).toBe(true);
+    expect(PredicateSchema.safeParse({ kind: "finding_count_below", n: 3 }).success).toBe(true);
+  });
+
+  it("accepts boolean combinators with nested leaves", () => {
+    const pred: Predicate = {
+      kind: "and",
+      predicates: [
+        { kind: "all_pass", stages: ["a"] },
+        {
+          kind: "or",
+          predicates: [
+            { kind: "no_open_findings" },
+            { kind: "not", predicate: { kind: "finding_count_below", n: 5 } },
+          ],
+        },
+      ],
+    };
+    expect(PredicateSchema.safeParse(pred).success).toBe(true);
+  });
+
+  it("rejects unknown kinds with a discriminator error", () => {
+    const result = PredicateSchema.safeParse({ kind: "nope" });
+    expect(result.success).toBe(false);
+  });
+
+  it("rejects empty boolean combinators", () => {
+    expect(PredicateSchema.safeParse({ kind: "and", predicates: [] }).success).toBe(false);
+    expect(PredicateSchema.safeParse({ kind: "or", predicates: [] }).success).toBe(false);
+  });
+
+  it("rejects finding_count_below with a negative n", () => {
+    expect(PredicateSchema.safeParse({ kind: "finding_count_below", n: -1 }).success).toBe(false);
+  });
+
+  it("predicateDepth counts boolean nesting only", () => {
+    expect(predicateDepth({ kind: "all_pass", stages: ["a"] })).toBe(0);
+    expect(
+      predicateDepth({
+        kind: "and",
+        predicates: [
+          { kind: "all_pass", stages: ["a"] },
+          { kind: "all_pass", stages: ["b"] },
+        ],
+      }),
+    ).toBe(1);
+    expect(
+      predicateDepth({
+        kind: "not",
+        predicate: {
+          kind: "and",
+          predicates: [{ kind: "all_pass", stages: ["a"] }],
+        },
+      }),
+    ).toBe(2);
+  });
+});
+
+describe("evaluatePredicate — leaves", () => {
+  it("all_pass uses stage scope when set, else every stage", () => {
+    const stages = {
+      a: makeStageState("a", { status: "succeeded" }),
+      b: makeStageState("b", { status: "failed" }),
+    };
+    expect(evaluatePredicate({ kind: "all_pass", stages: ["a"] }, ctx(stages))).toBe(true);
+    expect(evaluatePredicate({ kind: "all_pass", stages: ["b"] }, ctx(stages))).toBe(false);
+    // No scope → all stages — fails because b is failed.
+    expect(evaluatePredicate({ kind: "all_pass" }, ctx(stages))).toBe(false);
+  });
+
+  it("all_pass prefers verdict over status when verdict is set", () => {
+    const stages = {
+      a: makeStageState("a", { status: "succeeded", verdict: "fail" }),
+    };
+    expect(evaluatePredicate({ kind: "all_pass", stages: ["a"] }, ctx(stages))).toBe(false);
+  });
+
+  it("all_pass treats unknown stages as not-passing", () => {
+    const stages = { a: makeStageState("a") };
+    expect(evaluatePredicate({ kind: "all_pass", stages: ["ghost"] }, ctx(stages))).toBe(false);
+  });
+
+  it("no_open_findings counts only open finding artifacts", () => {
+    const stages = { review: makeStageState("review") };
+    const open = makeFinding("review", 0, { status: "open" });
+    const resolved = makeFinding("review", 1, { status: "resolved" });
+    expect(evaluatePredicate({ kind: "no_open_findings" }, ctx(stages, { review: [open] }))).toBe(
+      false,
+    );
+    expect(
+      evaluatePredicate({ kind: "no_open_findings" }, ctx(stages, { review: [resolved] })),
+    ).toBe(true);
+  });
+
+  it("finding_count_below is strictly less than n", () => {
+    const stages = { review: makeStageState("review") };
+    const findings = [makeFinding("review", 0), makeFinding("review", 1), makeFinding("review", 2)];
+    expect(
+      evaluatePredicate({ kind: "finding_count_below", n: 3 }, ctx(stages, { review: findings })),
+    ).toBe(false); // count = 3, not < 3
+    expect(
+      evaluatePredicate({ kind: "finding_count_below", n: 4 }, ctx(stages, { review: findings })),
+    ).toBe(true);
+  });
+});
+
+describe("evaluatePredicate — booleans", () => {
+  const stages = {
+    a: makeStageState("a", { status: "succeeded" }),
+    b: makeStageState("b", { status: "failed" }),
+  };
+
+  it("and is true only when every child is true", () => {
+    expect(
+      evaluatePredicate(
+        {
+          kind: "and",
+          predicates: [
+            { kind: "all_pass", stages: ["a"] },
+            { kind: "all_pass", stages: ["a"] },
+          ],
+        },
+        ctx(stages),
+      ),
+    ).toBe(true);
+    expect(
+      evaluatePredicate(
+        {
+          kind: "and",
+          predicates: [
+            { kind: "all_pass", stages: ["a"] },
+            { kind: "all_pass", stages: ["b"] },
+          ],
+        },
+        ctx(stages),
+      ),
+    ).toBe(false);
+  });
+
+  it("or is true when at least one child is true", () => {
+    expect(
+      evaluatePredicate(
+        {
+          kind: "or",
+          predicates: [
+            { kind: "all_pass", stages: ["b"] },
+            { kind: "all_pass", stages: ["a"] },
+          ],
+        },
+        ctx(stages),
+      ),
+    ).toBe(true);
+  });
+
+  it("not inverts its child", () => {
+    expect(
+      evaluatePredicate(
+        { kind: "not", predicate: { kind: "all_pass", stages: ["a"] } },
+        ctx(stages),
+      ),
+    ).toBe(false);
+    expect(
+      evaluatePredicate(
+        { kind: "not", predicate: { kind: "all_pass", stages: ["b"] } },
+        ctx(stages),
+      ),
+    ).toBe(true);
+  });
+});
+
+describe("evaluateExitPredicates", () => {
+  const stages = { a: makeStageState("a") };
+
+  it("returns true for empty / undefined predicate lists (v1.1 default)", () => {
+    expect(evaluateExitPredicates(undefined, ctx(stages))).toBe(true);
+    expect(evaluateExitPredicates([], ctx(stages))).toBe(true);
+  });
+
+  it("AND-combines multiple predicates", () => {
+    expect(
+      evaluateExitPredicates(
+        [{ kind: "all_pass", stages: ["a"] }, { kind: "no_open_findings" }],
+        ctx(stages),
+      ),
+    ).toBe(true);
+    expect(
+      evaluateExitPredicates(
+        [
+          { kind: "all_pass", stages: ["a"] },
+          { kind: "finding_count_below", n: 0 },
+        ],
+        ctx(stages, { a: [makeFinding("a", 0)] }),
+      ),
+    ).toBe(false);
+  });
+});
+
+describe("Legacy route predicate bridge", () => {
+  it("isLegacyRoutePredicate detects v1.1 shapes only", () => {
+    expect(isLegacyRoutePredicate({ kind: "allSucceeded", stages: ["a"] })).toBe(true);
+    expect(isLegacyRoutePredicate({ kind: "anySucceeded", stages: ["a"] })).toBe(true);
+    expect(isLegacyRoutePredicate({ kind: "anyFailed", stages: ["a"] })).toBe(true);
+    expect(isLegacyRoutePredicate({ kind: "all_pass", stages: ["a"] })).toBe(false);
+  });
+
+  it("fromLegacyRoutePredicate maps allSucceeded → all_pass", () => {
+    expect(fromLegacyRoutePredicate({ kind: "allSucceeded", stages: ["a", "b"] })).toEqual({
+      kind: "all_pass",
+      stages: ["a", "b"],
+    });
+  });
+
+  it("fromLegacyRoutePredicate maps anySucceeded → or-of-single-stage", () => {
+    const out = fromLegacyRoutePredicate({ kind: "anySucceeded", stages: ["a", "b"] });
+    expect(out.kind).toBe("or");
+    if (out.kind !== "or") return;
+    expect(out.predicates).toEqual([
+      { kind: "all_pass", stages: ["a"] },
+      { kind: "all_pass", stages: ["b"] },
+    ]);
+  });
+
+  it("fromLegacyRoutePredicate maps anyFailed → not(all_pass)", () => {
+    expect(fromLegacyRoutePredicate({ kind: "anyFailed", stages: ["a"] })).toEqual({
+      kind: "not",
+      predicate: { kind: "all_pass", stages: ["a"] },
+    });
+  });
+
+  it("normalizeRoutePredicate passes new DSL through unchanged", () => {
+    const dsl: Predicate = { kind: "all_pass", stages: ["a"] };
+    expect(normalizeRoutePredicate(dsl)).toBe(dsl);
+  });
+
+  it("anyFailed semantics: not(all_pass) fires when any stage didn't succeed", () => {
+    const stages = {
+      a: makeStageState("a", { status: "succeeded" }),
+      b: makeStageState("b", { status: "failed" }),
+    };
+    const pred = fromLegacyRoutePredicate({ kind: "anyFailed", stages: ["a", "b"] });
+    expect(evaluatePredicate(pred, ctx(stages))).toBe(true);
+  });
+});
+
+describe("collectReferencedStages", () => {
+  it("walks transitively across and/or/not", () => {
+    const refs = collectReferencedStages({
+      kind: "and",
+      predicates: [
+        { kind: "all_pass", stages: ["a"] },
+        {
+          kind: "or",
+          predicates: [
+            { kind: "no_open_findings", stages: ["b"] },
+            { kind: "not", predicate: { kind: "finding_count_below", n: 1, stages: ["c"] } },
+          ],
+        },
+      ],
+    });
+    expect([...refs].sort()).toEqual(["a", "b", "c"]);
+  });
+
+  it("returns empty for leaves without an explicit scope", () => {
+    expect(collectReferencedStages({ kind: "no_open_findings" })).toEqual([]);
+  });
+
+  it("normalizes legacy route predicates before walking", () => {
+    expect(collectReferencedStages({ kind: "allSucceeded", stages: ["a"] })).toEqual(["a"]);
+  });
+});
diff --git a/packages/core/src/__tests__/pipeline-workspace.test.ts b/packages/core/src/__tests__/pipeline-workspace.test.ts
new file mode 100644
index 000000000..72770c254
--- /dev/null
+++ b/packages/core/src/__tests__/pipeline-workspace.test.ts
@@ -0,0 +1,161 @@
+/**
+ * v1.3 — workspace class config validation + read-siblings prompt wiring.
+ *
+ * Covers WorkspaceGuard at config load: a stage that declares
+ * `workspaceClass: "read-siblings"` must have an upstream stage to read
+ * from. Also verifies the runtime prompt surfaces sibling artifacts when
+ * the class is set.
+ */
+
+import { describe, expect, it } from "vitest";
+
+import {
+  asArtifactId,
+  asRunId,
+  asStageRunId,
+  buildStagePrompt,
+  ConfiguredPipelineSchema,
+  type Artifact,
+  type Stage,
+} from "../pipeline/index.js";
+
+function makeStageInput(name: string, overrides: Record<string, unknown> = {}): unknown {
+  return {
+    name,
+    trigger: { on: ["pr.opened"] },
+    executor: { kind: "agent", plugin: "codex", mode: "review" },
+    task: { prompt: `run ${name}` },
+    ...overrides,
+  };
+}
+
+describe("WorkspaceGuard — read-siblings", () => {
+  it("rejects a single stage declaring read-siblings (orphan)", () => {
+    const result = ConfiguredPipelineSchema.safeParse({
+      stages: [makeStageInput("only", { workspaceClass: "read-siblings" })],
+    });
+    expect(result.success).toBe(false);
+    if (result.success) return;
+    const messages = result.error.issues.map((i) => i.message).join("\n");
+    expect(messages).toContain('Stage "only"');
+    expect(messages).toContain("read-siblings");
+    expect(messages).toContain("no upstream stages");
+  });
+
+  it("rejects the first-declared stage when it claims read-siblings", () => {
+    const result = ConfiguredPipelineSchema.safeParse({
+      stages: [makeStageInput("a", { workspaceClass: "read-siblings" }), makeStageInput("b")],
+    });
+    expect(result.success).toBe(false);
+    if (result.success) return;
+    const messages = result.error.issues.map((i) => i.message).join("\n");
+    expect(messages).toContain('Stage "a"');
+  });
+
+  it("accepts read-siblings when dependsOn names an upstream stage", () => {
+    const result = ConfiguredPipelineSchema.safeParse({
+      stages: [
+        makeStageInput("review"),
+        makeStageInput("fix", { workspaceClass: "read-siblings", dependsOn: ["review"] }),
+      ],
+    });
+    expect(result.success).toBe(true);
+  });
+
+  it("accepts read-siblings when a prior stage is declared (implicit upstream)", () => {
+    const result = ConfiguredPipelineSchema.safeParse({
+      stages: [
+        makeStageInput("review"),
+        makeStageInput("fix", { workspaceClass: "read-siblings" }),
+      ],
+    });
+    expect(result.success).toBe(true);
+  });
+
+  it("accepts read-siblings when routes reference an upstream stage", () => {
+    const result = ConfiguredPipelineSchema.safeParse({
+      stages: [
+        makeStageInput("review"),
+        makeStageInput("fix", {
+          workspaceClass: "read-siblings",
+          routes: { when: { kind: "all_pass", stages: ["review"] } },
+        }),
+      ],
+    });
+    expect(result.success).toBe(true);
+  });
+
+  it("accepts independent (default) on a single-stage pipeline", () => {
+    const result = ConfiguredPipelineSchema.safeParse({
+      stages: [makeStageInput("only")],
+    });
+    expect(result.success).toBe(true);
+  });
+
+  it("accepts an explicit independent declaration on a first stage", () => {
+    const result = ConfiguredPipelineSchema.safeParse({
+      stages: [makeStageInput("only", { workspaceClass: "independent" })],
+    });
+    expect(result.success).toBe(true);
+  });
+});
+
+describe("buildStagePrompt — read-siblings artifact injection", () => {
+  function makeArtifact(stageName: string): Artifact {
+    return {
+      artifactId: asArtifactId(`art-${stageName}`),
+      pipelineRunId: asRunId("run-1"),
+      stageRunId: asStageRunId(`sr-${stageName}`),
+      stageName,
+      kind: "finding",
+      filePath: "src/x.ts",
+      startLine: 1,
+      endLine: 2,
+      title: "t",
+      description: "d",
+      category: "general",
+      severity: "warning",
+      confidence: 0.9,
+      status: "open",
+      createdAt: new Date().toISOString(),
+    } as Artifact;
+  }
+
+  const baseStage: Stage = {
+    name: "fix",
+    trigger: { on: ["pr.opened"] },
+    executor: { kind: "agent", plugin: "codex", mode: "code" },
+    task: { prompt: "Apply the suggested fixes." },
+    dependsOn: ["review"],
+    workspaceClass: "read-siblings",
+  };
+
+  it("emits an Upstream Artifacts block when artifacts exist", () => {
+    const prompt = buildStagePrompt({
+      pipelineName: "demo",
+      stage: baseStage,
+      siblingArtifacts: { review: [makeArtifact("review")] },
+    });
+    expect(prompt).toContain("## Upstream Artifacts");
+    expect(prompt).toContain('"stageName": "review"');
+  });
+
+  it("omits the section when siblingArtifacts is empty", () => {
+    const prompt = buildStagePrompt({
+      pipelineName: "demo",
+      stage: baseStage,
+      siblingArtifacts: {},
+    });
+    expect(prompt).not.toContain("## Upstream Artifacts");
+  });
+
+  it("omits the section when workspaceClass is independent (default)", () => {
+    const stage: Stage = { ...baseStage, workspaceClass: "independent" };
+    const prompt = buildStagePrompt({
+      pipelineName: "demo",
+      stage,
+      siblingArtifacts: { review: [makeArtifact("review")] },
+    });
+    expect(prompt).not.toContain("## Upstream Artifacts");
+  });
+});
diff --git a/packages/core/src/pipeline/config-diff.ts b/packages/core/src/pipeline/config-diff.ts
new file mode 100644
index 000000000..15b597606
--- /dev/null
+++ b/packages/core/src/pipeline/config-diff.ts
@@ -0,0 +1,149 @@
+/**
+ * Pipeline config-change classification — distinguishes "structural" changes
+ * that invalidate an in-flight run from "tuning" changes the engine can pick
+ * up live.
+ *
+ * v1.3 spec (issue #1632):
+ *   structural — adds, removes, renames, or reshapes stages; alters dependsOn;
+ *                changes a stage's executor (kind / plugin / mode / command).
+ *   tuning      — predicate-threshold changes, retry counts, timeout / budget
+ *                values, exit-predicate body, maxLoopRounds, workspaceClass
+ *                annotations. The engine can apply these without aborting the
+ *                current run.
+ *
+ * Used by the driver (CLI / config-watcher) to decide whether to emit a
+ * `CONFIG_CHANGED` event (terminating the run) or to hot-reload the
+ * pipeline snapshot mid-run. Pure: deterministic, no I/O.
+ */
+
+import type { Pipeline, Stage } from "./types.js";
+
+export type ConfigChangeKind = "none" | "tuning" | "structural";
+
+export interface ConfigChangeClassification {
+  kind: ConfigChangeKind;
+  /** Stable, human-readable reasons. First entry is the dominant reason. */
+  reasons: string[];
+}
+
+/**
+ * Compare two pipeline configurations and classify the delta. The contract:
+ *   - `none`        → snapshots are deeply equivalent
+ *   - `tuning`      → only tuning-class fields changed (engine may hot-reload)
+ *   - `structural`  → at least one structural field changed (run must abort)
+ *
+ * Classification short-circuits on the first structural diff — once we know
+ * a change is structural, the engine has to terminate the run regardless of
+ * other diffs, so we don't waste cycles enumerating them all. The `reasons`
+ * array still records all detected differences for diagnostics.
+ */
+export function classifyConfigChange(prev: Pipeline, next: Pipeline): ConfigChangeClassification {
+  const reasons: string[] = [];
+  const acc: { kind: ConfigChangeKind } = { kind: "none" };
+
+  const bumpTo = (level: Exclude<ConfigChangeKind, "none">, reason: string): void => {
+    reasons.push(reason);
+    if (acc.kind === "structural") return;
+    if (level === "structural" || acc.kind === "none") acc.kind = level;
+  };
+
+  if (prev.name !== next.name) bumpTo("structural", `pipeline name changed`);
+  if ((prev.maxConcurrentStages ?? 1) !== (next.maxConcurrentStages ?? 1)) {
+    bumpTo("tuning", `maxConcurrentStages changed`);
+  }
+  if ((prev.maxLoopRounds ?? null) !== (next.maxLoopRounds ?? null)) {
+    bumpTo("tuning", `maxLoopRounds changed`);
+  }
+  if (!deepEqual(prev.exitPredicates ?? [], next.exitPredicates ?? [])) {
+    bumpTo("tuning", `exitPredicates changed`);
+  }
+
+  // Stage-list shape: any rename, add, remove, or reorder is structural — the
+  // DAG depends on declaration order for slot allocation, and stage names are
+  // the join key for runtime state.
+  const prevNames = prev.stages.map((s) => s.name);
+  const nextNames = next.stages.map((s) => s.name);
+  if (!arraysEqual(prevNames, nextNames)) {
+    bumpTo("structural", `stage list changed (${prevNames.join(",")} → ${nextNames.join(",")})`);
+    return { kind: acc.kind, reasons };
+  }
+
+  for (let i = 0; i < prev.stages.length; i++) {
+    const before = prev.stages[i];
+    const after = next.stages[i];
+    classifyStage(before, after, bumpTo);
+    if (acc.kind === "structural") return { kind: acc.kind, reasons };
+  }
+
+  return { kind: acc.kind, reasons };
+}
+
+function classifyStage(
+  before: Stage,
+  after: Stage,
+  bump: (level: Exclude<ConfigChangeKind, "none">, reason: string) => void,
+): void {
+  // Executor is structural: changing the plugin or kind invalidates the
+  // running subprocess. Mode/command changes mean the agent would do
+  // different work, so they're also structural.
+  if (!deepEqual(before.executor, after.executor)) {
+    bump("structural", `stage "${after.name}" executor changed`);
+  }
+  if (!arraysEqual(before.dependsOn ?? [], after.dependsOn ?? [])) {
+    bump("structural", `stage "${after.name}" dependsOn changed`);
+  }
+  // Trigger events are structural — they decide whether a stage even runs.
+  if (!arraysEqual([...before.trigger.on], [...after.trigger.on])) {
+    bump("structural", `stage "${after.name}" trigger.on changed`);
+  }
+  if (!deepEqual(before.routes ?? null, after.routes ?? null)) {
+    // Routes are part of the DAG shape — predicate thresholds aren't, but the
+    // route topology itself is. We don't try to distinguish "same DAG, looser
+    // predicate" from "different DAG" here; any route change is structural.
+    // Tuning-only predicate edits should live in `exitPredicates`.
+    bump("structural", `stage "${after.name}" routes changed`);
+  }
+
+  // Tuning-class fields below.
+  if (!deepEqual(before.task ?? {}, after.task ?? {})) {
+    bump("tuning", `stage "${after.name}" task changed`);
+  }
+  if (!deepEqual(before.policy ?? {}, after.policy ?? {})) {
+    bump("tuning", `stage "${after.name}" policy changed`);
+  }
+  if (!deepEqual(before.budget ?? {}, after.budget ?? {})) {
+    bump("tuning", `stage "${after.name}" budget changed`);
+  }
+  if ((before.timeoutMs ?? null) !== (after.timeoutMs ?? null)) {
+    bump("tuning", `stage "${after.name}" timeoutMs changed`);
+  }
+  if ((before.retries ?? null) !== (after.retries ?? null)) {
+    bump("tuning", `stage "${after.name}" retries changed`);
+  }
+  if ((before.maxLoopRounds ?? null) !== (after.maxLoopRounds ?? null)) {
+    bump("tuning", `stage "${after.name}" maxLoopRounds changed`);
+  }
+  if ((before.workspaceClass ?? "independent") !== (after.workspaceClass ?? "independent")) {
+    // Workspace class affects future stage invocations only — current
+    // invocation is unaffected. Tuning.
+    bump("tuning", `stage "${after.name}" workspaceClass changed`);
+  }
+}
+
+function arraysEqual<T>(a: readonly T[], b: readonly T[]): boolean {
+  if (a.length !== b.length) return false;
+  for (let i = 0; i < a.length; i++) if (a[i] !== b[i]) return false;
+  return true;
+}
+
+/**
+ * Structural deep-equal for plain-data records. Pipelines/Stages don't
+ * contain Maps, Sets, Dates, or class instances — JSON.stringify is enough
+ * and avoids hand-rolling a recursive comparator that would risk missing
+ * an edge case. Output order isn't deterministic across JS engines for
+ * objects with non-string keys, but Pipeline values are all string-keyed.
+ */
+function deepEqual(a: unknown, b: unknown): boolean {
+  if (a === b) return true;
+  return JSON.stringify(a) === JSON.stringify(b);
+}
diff --git a/packages/core/src/pipeline/config-schema.ts b/packages/core/src/pipeline/config-schema.ts
index 0f09c1331..f23c790b0 100644
--- a/packages/core/src/pipeline/config-schema.ts
+++ b/packages/core/src/pipeline/config-schema.ts
@@ -13,14 +13,22 @@
 import { z } from "zod";
 
 import { findFirstStageCycle } from "./dag.js";
+import {
+  collectReferencedStages,
+  MAX_PREDICATE_DEPTH,
+  PredicateSchema,
+  predicateDepth,
+} from "./predicate.js";
 import {
   asPipelineId,
   type Pipeline,
+  type Predicate,
   type Stage,
   type StageExecutor,
   type StageRoutePredicate,
   type StageRoutes,
   type TaskMode,
+  type WorkspaceClass,
 } from "./types.js";
 
 const TaskModeSchema = z.enum(["review", "code", "answer"]);
@@ -65,6 +73,11 @@ const StageBudgetSchema = z.object({
   maxDurationMs: z.number().int().nonnegative().optional(),
 });
 
+/**
+ * v1.1's hardcoded route predicates. Kept alongside the v1.3 typed DSL so
+ * existing pipeline configs continue to parse without rewrites — the union
+ * below accepts either shape.
+ */
 const StageRoutePredicateSchema = z.discriminatedUnion("kind", [
   z.object({ kind: z.literal("allSucceeded"), stages: z.array(z.string().min(1)).min(1) }),
   z.object({ kind: z.literal("anySucceeded"), stages: z.array(z.string().min(1)).min(1) }),
@@ -72,9 +85,13 @@ const StageRoutePredicateSchema = z.discriminatedUnion("kind", [
 ]);
 
 const StageRoutesSchema = z.object({
-  when: StageRoutePredicateSchema,
+  when: z.union([StageRoutePredicateSchema, PredicateSchema]),
 });
 
+const WorkspaceClassSchema = z.enum(["independent", "read-siblings"]);
+
+const LEGACY_ROUTE_KINDS = new Set<string>(["allSucceeded", "anySucceeded", "anyFailed"]);
+
 const StageSchema = z.object({
   name: z.string().min(1),
   trigger: StageTriggerSchema,
@@ -87,6 +104,7 @@ const StageSchema = z.object({
   maxLoopRounds: z.number().int().positive().optional(),
   dependsOn: z.array(z.string().min(1)).optional(),
   routes: StageRoutesSchema.optional(),
+  workspaceClass: WorkspaceClassSchema.optional(),
 });
 
 /**
@@ -108,6 +126,13 @@ export const ConfiguredPipelineSchema = z
     name: z.string().min(1).optional(),
     stages: z.array(StageSchema).min(1),
     maxConcurrentStages: z.number().int().positive().optional(),
+    /**
+     * v1.3 — pipeline-level run-completion conditions (AND-combined).
+     * Empty / unset preserves v1.1 behavior (every terminal run is "done").
+     */
+    exitPredicates: z.array(PredicateSchema).optional(),
+    /** v1.3 — pipeline-level loop cap. See `Pipeline.maxLoopRounds`. */
+    maxLoopRounds: z.number().int().positive().optional(),
   })
   .superRefine((pipeline, ctx) => {
     const stageNames = new Set(pipeline.stages.map((s) => s.name));
@@ -128,6 +153,10 @@ export const ConfiguredPipelineSchema = z
     }
 
     // dependsOn / routes references must point to known stage names.
+    // WorkspaceGuard: `read-siblings` requires at least one upstream stage
+    // (dependsOn, route reference, or a prior-declared stage). Reject orphans
+    // at config load so runtime never has to handle a "no siblings to read"
+    // case.
     for (let i = 0; i < pipeline.stages.length; i++) {
       const stage = pipeline.stages[i];
       for (const dep of stage.dependsOn ?? []) {
@@ -148,22 +177,78 @@ export const ConfiguredPipelineSchema = z
       }
       const routes = stage.routes;
       if (routes) {
-        for (const ref of routes.when.stages) {
+        // Cast: superRefine receives the parsed-but-not-narrowed input.
+        // `collectReferencedStages` accepts either shape.
+        const refs = collectReferencedStages(routes.when as StageRoutePredicate | Predicate);
+        for (const ref of refs) {
           if (!stageNames.has(ref)) {
             ctx.addIssue({
               code: z.ZodIssueCode.custom,
-              path: ["stages", i, "routes", "when", "stages"],
+              path: ["stages", i, "routes", "when"],
               message: `Stage "${stage.name}" routes references unknown stage "${ref}".`,
             });
           }
           if (ref === stage.name) {
             ctx.addIssue({
               code: z.ZodIssueCode.custom,
-              path: ["stages", i, "routes", "when", "stages"],
+              path: ["stages", i, "routes", "when"],
               message: `Stage "${stage.name}" cannot route to itself.`,
             });
           }
         }
+        // Bound predicate nesting so malformed configs can't OOM the evaluator.
+        for (const wh of [routes.when as StageRoutePredicate | Predicate]) {
+          // Legacy shapes are flat (depth 0); only DSL predicates need depth
+          // checks. Detect by `kind` to avoid `predicateDepth` blowing up on
+          // legacy shapes it doesn't know about.
+          if (!LEGACY_ROUTE_KINDS.has(wh.kind)) {
+            const depth = predicateDepth(wh as Predicate);
+            if (depth > MAX_PREDICATE_DEPTH) {
+              ctx.addIssue({
+                code: z.ZodIssueCode.custom,
+                path: ["stages", i, "routes", "when"],
+                message: `Stage "${stage.name}" route predicate is nested ${depth} levels deep; the maximum is ${MAX_PREDICATE_DEPTH}.`,
+              });
+            }
+          }
+        }
+      }
+
+      if (stage.workspaceClass === "read-siblings") {
+        const hasDependsOn = (stage.dependsOn?.length ?? 0) > 0;
+        const hasRouteRefs = stage.routes
+          ? collectReferencedStages(stage.routes.when as StageRoutePredicate | Predicate).length > 0
+          : false;
+        const hasPriorStage = i > 0;
+        if (!hasDependsOn && !hasRouteRefs && !hasPriorStage) {
+          ctx.addIssue({
+            code: z.ZodIssueCode.custom,
+            path: ["stages", i, "workspaceClass"],
+            message: `Stage "${stage.name}" declares workspaceClass="read-siblings" but has no upstream stages; declare dependsOn or place the stage after the stages it should read.`,
+          });
+        }
+      }
+    }
+
+    // Validate pipeline-level exitPredicates references and depth.
+    for (let p = 0; p < (pipeline.exitPredicates?.length ?? 0); p++) {
+      const pred = pipeline.exitPredicates![p];
+      const depth = predicateDepth(pred);
+      if (depth > MAX_PREDICATE_DEPTH) {
+        ctx.addIssue({
+          code: z.ZodIssueCode.custom,
+          path: ["exitPredicates", p],
+          message: `exitPredicates[${p}] is nested ${depth} levels deep; the maximum is ${MAX_PREDICATE_DEPTH}.`,
+        });
+      }
+      for (const ref of collectReferencedStages(pred)) {
+        if (!stageNames.has(ref)) {
+          ctx.addIssue({
+            code: z.ZodIssueCode.custom,
+            path: ["exitPredicates", p],
+            message: `exitPredicates[${p}] references unknown stage "${ref}".`,
+          });
+        }
       }
     }
 
@@ -208,12 +293,7 @@ export function configuredPipelineToRuntime(key: string, configured: ConfiguredP
           };
 
     const routes: StageRoutes | undefined = stage.routes
-      ? {
-          when: {
-            kind: stage.routes.when.kind,
-            stages: [...stage.routes.when.stages],
-          } as StageRoutePredicate,
-        }
+      ? { when: cloneRouteWhen(stage.routes.when as StageRoutePredicate | Predicate) }
       : undefined;
 
     return {
@@ -228,6 +308,9 @@ export function configuredPipelineToRuntime(key: string, configured: ConfiguredP
       ...(stage.maxLoopRounds !== undefined ? { maxLoopRounds: stage.maxLoopRounds } : {}),
       ...(stage.dependsOn !== undefined ? { dependsOn: [...stage.dependsOn] } : {}),
       ...(routes ? { routes } : {}),
+      ...(stage.workspaceClass !== undefined
+        ? { workspaceClass: stage.workspaceClass as WorkspaceClass }
+        : {}),
     };
   });
 
@@ -238,5 +321,40 @@ export function configuredPipelineToRuntime(key: string, configured: ConfiguredP
     ...(configured.maxConcurrentStages !== undefined
       ? { maxConcurrentStages: configured.maxConcurrentStages }
       : {}),
+    ...(configured.exitPredicates !== undefined
+      ? { exitPredicates: configured.exitPredicates.map(clonePredicate) }
+      : {}),
+    ...(configured.maxLoopRounds !== undefined ? { maxLoopRounds: configured.maxLoopRounds } : {}),
   };
 }
+
+/**
+ * Deep clone a route's `when` so the runtime Pipeline is fully detached from
+ * the Zod-parsed object. Legacy shapes stay legacy; DSL shapes stay DSL —
+ * the runtime evaluator handles both.
+ */
+function cloneRouteWhen(when: StageRoutePredicate | Predicate): StageRoutePredicate | Predicate {
+  if (LEGACY_ROUTE_KINDS.has(when.kind)) {
+    const legacy = when as StageRoutePredicate;
+    return { kind: legacy.kind, stages: [...legacy.stages] } as StageRoutePredicate;
+  }
+  return clonePredicate(when as Predicate);
+}
+
+function clonePredicate(p: Predicate): Predicate {
+  switch (p.kind) {
+    case "all_pass":
+    case "no_open_findings":
+      return p.stages !== undefined ? { kind: p.kind, stages: [...p.stages] } : { kind: p.kind };
+    case "finding_count_below":
+      return p.stages !== undefined
+        ? { kind: "finding_count_below", n: p.n, stages: [...p.stages] }
+        : { kind: "finding_count_below", n: p.n };
+    case "and":
+      return { kind: "and", predicates: p.predicates.map(clonePredicate) };
+    case "or":
+      return { kind: "or", predicates: p.predicates.map(clonePredicate) };
+    case "not":
+      return { kind: "not", predicate: clonePredicate(p.predicate) };
+  }
+}
diff --git a/packages/core/src/pipeline/dag.ts b/packages/core/src/pipeline/dag.ts
index 13606f2d5..6a0efbaa3 100644
--- a/packages/core/src/pipeline/dag.ts
+++ b/packages/core/src/pipeline/dag.ts
@@ -17,11 +17,19 @@
  */
 
 import type { PipelineEffect } from "./events.js";
+import {
+  collectReferencedStages,
+  evaluatePredicate as evalDslPredicate,
+  evaluateExitPredicates as evalExit,
+  normalizeRoutePredicate,
+  type PredicateContext,
+} from "./predicate.js";
 import { iso, patchRun } from "./reducer-helpers.js";
 import {
+  type Artifact,
   type RunState,
   type Stage,
-  type StageRoutePredicate,
+  type StageRoutes,
   type StageState,
   isTerminalStageStatus,
 } from "./types.js";
@@ -59,7 +67,7 @@ export function scheduleAfterChange(run: RunState, now: number): ScheduleResult
       const state = current.stages[stageDef.name];
       if (state.status !== "pending") continue;
       if (!areDepsSatisfiedForStart(stageDef, current.stages)) continue;
-      if (!evaluateRoutes(stageDef, current.stages)) continue;
+      if (stageDef.routes && !evaluateRoutes(stageDef.routes, current)) continue;
       startEffects.push({
         type: "START_STAGE",
         runId: current.runId,
@@ -98,7 +106,7 @@ function applyEligibleSkips(run: RunState, now: number): { run: RunState; newlyS
       if (!arePreconditionsTerminal(stageDef, current.stages)) continue;
 
       const shouldSkip = stageDef.routes
-        ? !evaluatePredicate(stageDef.routes.when, current.stages)
+        ? !evaluateRoutes(stageDef.routes, current)
         : !areAllDepsSucceeded(stageDef, current.stages);
 
       if (shouldSkip) {
@@ -119,7 +127,7 @@ function applyEligibleSkips(run: RunState, now: number): { run: RunState; newlyS
 function arePreconditionsTerminal(stage: Stage, stages: Record<string, StageState>): boolean {
   if (!areDepsTerminal(stage, stages)) return false;
   if (stage.routes) {
-    for (const ref of stage.routes.when.stages) {
+    for (const ref of collectReferencedStages(stage.routes.when)) {
       const refState = stages[ref];
       if (!refState || !isTerminalStageStatus(refState.status)) return false;
     }
@@ -154,29 +162,53 @@ function areDepsSatisfiedForStart(stage: Stage, stages: Record<string, StageStat
   return areAllDepsSucceeded(stage, stages);
 }
 
-function evaluateRoutes(stage: Stage, stages: Record<string, StageState>): boolean {
-  if (!stage.routes) return true;
-  return evaluatePredicate(stage.routes.when, stages);
+function evaluateRoutes(routes: StageRoutes, run: RunState): boolean {
+  const predicate = normalizeRoutePredicate(routes.when);
+  return evalDslPredicate(predicate, predicateContextForRun(run));
 }
 
 /**
- * Pure predicate evaluator over the runtime stage states. Stages referenced
- * here are guaranteed by config validation to exist in the pipeline; missing
- * entries are treated as non-terminal (i.e. `false` for everything except
- * `anyFailed` short-circuits) so an unevaluable predicate never green-lights.
+ * Build the evaluator context off the live run state. Routes only reference
+ * stage statuses today, but `Pipeline.exitPredicates` reuses the same context
+ * to count open findings; threading `runArtifacts` through both paths keeps
+ * a single evaluator implementation rather than two near-duplicates.
  */
-function evaluatePredicate(
-  predicate: StageRoutePredicate,
-  stages: Record<string, StageState>,
-): boolean {
-  switch (predicate.kind) {
-    case "allSucceeded":
-      return predicate.stages.every((name) => stages[name]?.status === "succeeded");
-    case "anySucceeded":
-      return predicate.stages.some((name) => stages[name]?.status === "succeeded");
-    case "anyFailed":
-      return predicate.stages.some((name) => stages[name]?.status === "failed");
+export function predicateContextForRun(run: RunState): PredicateContext {
+  const artifactsByStage: Record<string, Artifact[]> = run.runArtifacts ?? {};
+  return {
+    stages: run.stages,
+    artifactsByStage,
+    allStageNames: run.pipelineConfigSnapshot.stages.map((s) => s.name),
+  };
+}
+
+export type RunExitOutcome =
+  | { kind: "succeeded" }
+  | { kind: "failed_exhausted" }
+  | { kind: "failed_can_retry" };
+
+/**
+ * Evaluate a run's `exitPredicates` once every stage has reached a terminal
+ * status. Decision table (all stages terminal, predicates fully evaluated):
+ *
+ *   exitPredicates true  (or unset)       → `succeeded`
+ *   exitPredicates false, rounds exhausted → `failed_exhausted` (loop_failed)
+ *   exitPredicates false, rounds remain    → `failed_can_retry` (loop awaits)
+ *
+ * Rounds are "exhausted" when `Pipeline.maxLoopRounds` is unset (no retry
+ * budget configured) OR when `loopRounds >= maxLoopRounds`. The reducer maps
+ * these outcomes onto `LoopStateName` — see `reducer.ts`.
+ */
+export function evaluateRunExitOutcome(run: RunState): RunExitOutcome {
+  const pipeline = run.pipelineConfigSnapshot;
+  const ctx = predicateContextForRun(run);
+  const passed = evalExit(pipeline.exitPredicates, ctx);
+  if (passed) return { kind: "succeeded" };
+  const maxRounds = pipeline.maxLoopRounds;
+  if (maxRounds === undefined || run.loopRounds >= maxRounds) {
+    return { kind: "failed_exhausted" };
   }
+  return { kind: "failed_can_retry" };
 }
 
 /**
@@ -198,15 +230,13 @@ export function findFirstStageCycle(
   stages: ReadonlyArray<{
     name: string;
     dependsOn?: string[];
-    routes?: { when: { stages: string[] } };
+    routes?: StageRoutes;
   }>,
 ): string[] | null {
   const adjacency = new Map<string, string[]>();
   for (const stage of stages) {
-    const edges = new Set<string>([
-      ...(stage.dependsOn ?? []),
-      ...(stage.routes?.when.stages ?? []),
-    ]);
+    const routeRefs = stage.routes ? collectReferencedStages(stage.routes.when) : [];
+    const edges = new Set<string>([...(stage.dependsOn ?? []), ...routeRefs]);
     adjacency.set(stage.name, [...edges]);
   }
   const WHITE = 0;
diff --git a/packages/core/src/pipeline/engine.ts b/packages/core/src/pipeline/engine.ts
index 416fe97c4..cf9a3789b 100644
--- a/packages/core/src/pipeline/engine.ts
+++ b/packages/core/src/pipeline/engine.ts
@@ -39,9 +39,12 @@ import {
   asStageRunId,
   emptyEngineState,
   isTerminalLoopState,
+  type Artifact,
   type EngineState,
   type Pipeline,
   type RunId,
+  type RunState,
+  type Stage,
   type StageRunId,
   type StageTriggerEvent,
 } from "./types.js";
@@ -213,6 +216,10 @@ export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine {
         });
 
         const meta = runMetadata.get(run.runId);
+        const siblingArtifacts =
+          effect.stage.workspaceClass === "read-siblings"
+            ? collectSiblingArtifacts(run, effect.stage)
+            : undefined;
         const startInput: StartStageInput = {
           pipelineName: run.pipelineName,
           projectId: meta?.projectId ?? "",
@@ -221,6 +228,7 @@ export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine {
           stage: effect.stage,
           loopRound: run.loopRounds,
           ...(meta?.issueId ? { issueId: meta.issueId } : {}),
+          ...(siblingArtifacts !== undefined ? { siblingArtifacts } : {}),
         };
 
         try {
@@ -347,3 +355,33 @@ export function createPipelineEngine(deps: PipelineEngineDeps): PipelineEngine {
     cancelRun,
   };
 }
+
+/**
+ * Collect artifacts from upstream stages a `read-siblings` stage should see.
+ * Upstream = the transitive closure of `dependsOn`. Routes references aren't
+ * included here because they're activation gates, not data dependencies — a
+ * stage that reacts to a sibling's *failure* doesn't necessarily want to read
+ * the sibling's artifacts.
+ *
+ * Returns an object keyed by upstream stage name. Stages that produced no
+ * artifacts (yet) are omitted to keep the prompt block compact.
+ */
+function collectSiblingArtifacts(run: RunState, stage: Stage): Record<string, Artifact[]> {
+  const buffer = run.runArtifacts ?? {};
+  const visited = new Set<string>();
+  const queue: string[] = [...(stage.dependsOn ?? [])];
+  const stagesByName = new Map<string, Stage>();
+  for (const s of run.pipelineConfigSnapshot.stages) stagesByName.set(s.name, s);
+
+  const out: Record<string, Artifact[]> = {};
+  while (queue.length > 0) {
+    const name = queue.shift()!;
+    if (visited.has(name)) continue;
+    visited.add(name);
+    const artifacts = buffer[name];
+    if (artifacts && artifacts.length > 0) out[name] = artifacts;
+    const upstream = stagesByName.get(name)?.dependsOn ?? [];
+    for (const next of upstream) if (!visited.has(next)) queue.push(next);
+  }
+  return out;
+}
diff --git a/packages/core/src/pipeline/executors/agent.ts b/packages/core/src/pipeline/executors/agent.ts
index 1c47bdc64..9d638f279 100644
--- a/packages/core/src/pipeline/executors/agent.ts
+++ b/packages/core/src/pipeline/executors/agent.ts
@@ -28,6 +28,7 @@ import { type SessionId, type SessionManager } from "../../types.js";
 import { buildStagePrompt } from "../stage-prompt.js";
 import {
   PIPELINE_FINDINGS_FILENAME,
+  type Artifact,
   type ArtifactInput,
   type RunId,
   type Stage,
@@ -51,6 +52,13 @@ export interface StartStageInput {
   issueId?: string;
   /** Loop counter from the engine. Surfaced in the prompt only. */
   loopRound?: number;
+  /**
+   * Artifacts from upstream sibling stages, keyed by stage name. Only used
+   * when `stage.workspaceClass === "read-siblings"`. The engine collects
+   * these from `RunState.runArtifacts` for stages this one transitively
+   * depends on; `undefined` is the default (independent semantics).
+   */
+  siblingArtifacts?: Record<string, Artifact[]>;
 }
 
 /**
@@ -118,6 +126,9 @@ export function createAgentExecutor(deps: AgentExecutorDeps): AgentStageExecutor
       pipelineName: input.pipelineName,
       stage: input.stage,
       loopRound: input.loopRound,
+      ...(input.siblingArtifacts !== undefined
+        ? { siblingArtifacts: input.siblingArtifacts }
+        : {}),
     });
 
     let session;
diff --git a/packages/core/src/pipeline/index.ts b/packages/core/src/pipeline/index.ts
index 3876da485..e6e9e806b 100644
--- a/packages/core/src/pipeline/index.ts
+++ b/packages/core/src/pipeline/index.ts
@@ -26,7 +26,33 @@ export {
   validatePipelineDag,
 } from "./validation.js";
 
-export { findFirstStageCycle, scheduleAfterChange, type ScheduleResult } from "./dag.js";
+export {
+  evaluateRunExitOutcome,
+  findFirstStageCycle,
+  predicateContextForRun,
+  scheduleAfterChange,
+  type RunExitOutcome,
+  type ScheduleResult,
+} from "./dag.js";
+
+export {
+  PredicateSchema,
+  MAX_PREDICATE_DEPTH,
+  collectReferencedStages,
+  evaluateExitPredicates,
+  evaluatePredicate,
+  fromLegacyRoutePredicate,
+  isLegacyRoutePredicate,
+  normalizeRoutePredicate,
+  predicateDepth,
+  type PredicateContext,
+} from "./predicate.js";
+
+export {
+  classifyConfigChange,
+  type ConfigChangeClassification,
+  type ConfigChangeKind,
+} from "./config-diff.js";
 
 export { buildStagePrompt, type StagePromptInput } from "./stage-prompt.js";
 
diff --git a/packages/core/src/pipeline/predicate.ts b/packages/core/src/pipeline/predicate.ts
new file mode 100644
index 000000000..55c3e7451
--- /dev/null
+++ b/packages/core/src/pipeline/predicate.ts
@@ -0,0 +1,237 @@
+/**
+ * Typed predicate DSL — parser (Zod) + pure evaluator.
+ *
+ * Used in two places:
+ *  - `Stage.routes.when` — gates a stage's activation on prior stage state
+ *    (new form alongside v1.1's hardcoded `StageRoutePredicate`).
+ *  - `Pipeline.exitPredicates` — decides whether a fully-terminal run is
+ *    `loop_succeeded` (done) or `loop_failed` (stalled).
+ *
+ * Leaves:
+ *  - `all_pass`            — every referenced stage's terminal status is
+ *                            `succeeded` (or `verdict === "pass"` when set).
+ *                            Empty scope is vacuously true; the caller is
+ *                            responsible for choosing a sensible default scope.
+ *  - `no_open_findings`    — zero finding artifacts in scope have
+ *                            `status === "open"`.
+ *  - `finding_count_below` — strictly fewer than `n` open findings in scope.
+ *
+ * Boolean ops (`and`/`or`/`not`) compose them; nesting depth is bounded only
+ * by Zod's recursion guard so deeply pathological configs fail at parse time.
+ *
+ * Pure: no I/O, no clock reads, no allocation that escapes the call.
+ */
+
+import { z } from "zod";
+
+import type {
+  Artifact,
+  FindingArtifactInput,
+  Predicate,
+  StageRoutePredicate,
+  StageState,
+} from "./types.js";
+
+// ============================================================================
+// Zod schema
+// ============================================================================
+
+const StagesScope = z.array(z.string().min(1)).optional();
+
+/**
+ * Recursive Zod schema for `Predicate`. `z.ZodType<Predicate>` short-circuits
+ * the type inference Zod can't perform across the lazy reference; without it
+ * the inferred output of `z.lazy(() => PredicateSchema)` becomes `any` and we
+ * lose the precise discriminator narrowing in callers.
+ */
+export const PredicateSchema: z.ZodType<Predicate> = z.lazy(() =>
+  z.discriminatedUnion("kind", [
+    z.object({ kind: z.literal("all_pass"), stages: StagesScope }),
+    z.object({ kind: z.literal("no_open_findings"), stages: StagesScope }),
+    z.object({
+      kind: z.literal("finding_count_below"),
+      n: z.number().int().nonnegative(),
+      stages: StagesScope,
+    }),
+    z.object({ kind: z.literal("and"), predicates: z.array(PredicateSchema).min(1) }),
+    z.object({ kind: z.literal("or"), predicates: z.array(PredicateSchema).min(1) }),
+    z.object({ kind: z.literal("not"), predicate: PredicateSchema }),
+  ]),
+);
+
+// ============================================================================
+// Evaluator
+// ============================================================================
+
+export interface PredicateContext {
+  /** Live stage states keyed by stage name. */
+  stages: Record<string, StageState>;
+  /** Artifacts produced during the run, keyed by stage name. */
+  artifactsByStage: Record<string, Artifact[]>;
+  /**
+   * All stage names declared by the pipeline, in declaration order. Used when
+   * a leaf omits its `stages` scope (default = "every stage in the run").
+   */
+  allStageNames: string[];
+}
+
+/**
+ * Evaluate a typed `Predicate` against runtime state.
+ *
+ * Stage references that don't appear in `context.stages` are treated as
+ * non-terminal (i.e. `all_pass` returns false). The reducer guarantees every
+ * stage in `pipelineConfigSnapshot` has an entry in `RunState.stages`, so this
+ * fallback only ever fires when a predicate references a stage outside the
+ * pipeline — which the schema rejects at config load.
+ */
+export function evaluatePredicate(predicate: Predicate, context: PredicateContext): boolean {
+  switch (predicate.kind) {
+    case "all_pass": {
+      const scope = predicate.stages ?? context.allStageNames;
+      if (scope.length === 0) return true;
+      return scope.every((name) => isStagePassing(context.stages[name]));
+    }
+    case "no_open_findings": {
+      return collectOpenFindings(predicate.stages, context).length === 0;
+    }
+    case "finding_count_below": {
+      return collectOpenFindings(predicate.stages, context).length < predicate.n;
+    }
+    case "and":
+      return predicate.predicates.every((p) => evaluatePredicate(p, context));
+    case "or":
+      return predicate.predicates.some((p) => evaluatePredicate(p, context));
+    case "not":
+      return !evaluatePredicate(predicate.predicate, context);
+  }
+}
+
+function isStagePassing(stage: StageState | undefined): boolean {
+  if (!stage) return false;
+  if (stage.verdict !== undefined) return stage.verdict === "pass";
+  return stage.status === "succeeded";
+}
+
+function collectOpenFindings(
+  stagesScope: string[] | undefined,
+  context: PredicateContext,
+): FindingArtifactInput[] {
+  const scope = stagesScope ?? context.allStageNames;
+  const out: FindingArtifactInput[] = [];
+  for (const name of scope) {
+    const artifacts = context.artifactsByStage[name];
+    if (!artifacts) continue;
+    for (const a of artifacts) {
+      if (a.kind !== "finding") continue;
+      if (a.status !== "open") continue;
+      out.push(a);
+    }
+  }
+  return out;
+}
+
+// ============================================================================
+// Bridge: legacy v1.1 routes → new DSL
+// ============================================================================
+
+/**
+ * Convert a v1.1 `StageRoutePredicate` into an equivalent `Predicate`. The
+ * scheduler keeps both shapes accepted so existing pipelines don't need a
+ * rewrite, but evaluation goes through a single code path.
+ *
+ *  - `allSucceeded` → `all_pass`
+ *  - `anySucceeded` → `or(all_pass([s]))`
+ *  - `anyFailed`    → `not(all_pass)` over the same scope. v1.1's reducer
+ *    treats "any non-success terminal status" as failure for routing
+ *    purposes, which is exactly what `not all_pass` captures.
+ */
+export function fromLegacyRoutePredicate(legacy: StageRoutePredicate): Predicate {
+  switch (legacy.kind) {
+    case "allSucceeded":
+      return { kind: "all_pass", stages: [...legacy.stages] };
+    case "anySucceeded":
+      return {
+        kind: "or",
+        predicates: legacy.stages.map((s) => ({ kind: "all_pass", stages: [s] })),
+      };
+    case "anyFailed":
+      return { kind: "not", predicate: { kind: "all_pass", stages: [...legacy.stages] } };
+  }
+}
+
+const LEGACY_KINDS = new Set(["allSucceeded", "anySucceeded", "anyFailed"]);
+
+/** Type-guard: is this an old StageRoutePredicate rather than the typed DSL? */
+export function isLegacyRoutePredicate(
+  value: StageRoutePredicate | Predicate,
+): value is StageRoutePredicate {
+  return LEGACY_KINDS.has(value.kind);
+}
+
+/** Normalize either shape into the typed DSL for unified evaluation. */
+export function normalizeRoutePredicate(value: StageRoutePredicate | Predicate): Predicate {
+  return isLegacyRoutePredicate(value) ? fromLegacyRoutePredicate(value) : value;
+}
+
+/**
+ * Collect every stage name a predicate references (transitive — boolean
+ * combinators descend). Used by `dag.ts` to figure out which upstream stages
+ * a stage's routes wait on, and by validation to verify references resolve.
+ */
+export function collectReferencedStages(value: StageRoutePredicate | Predicate): string[] {
+  const predicate = normalizeRoutePredicate(value);
+  const refs = new Set<string>();
+  walk(predicate);
+  return [...refs];
+
+  function walk(p: Predicate): void {
+    switch (p.kind) {
+      case "all_pass":
+      case "no_open_findings":
+      case "finding_count_below":
+        for (const s of p.stages ?? []) refs.add(s);
+        return;
+      case "and":
+      case "or":
+        for (const child of p.predicates) walk(child);
+        return;
+      case "not":
+        walk(p.predicate);
+        return;
+    }
+  }
+}
+
+/**
+ * Combine multiple pipeline exit predicates into a single decision. Returns
+ * `true` when the array is empty (callers default to "no predicate → success"
+ * to preserve v1.1 behavior).
+ */
+export function evaluateExitPredicates(
+  predicates: Predicate[] | undefined,
+  context: PredicateContext,
+): boolean {
+  if (!predicates || predicates.length === 0) return true;
+  return predicates.every((p) => evaluatePredicate(p, context));
+}
+
+/**
+ * Predicate-tree depth (defense-in-depth against deeply-nested config).
+ * `0` for a leaf, `1 + max(child depths)` for combinators.
+ */
+export function predicateDepth(predicate: Predicate): number {
+  switch (predicate.kind) {
+    case "all_pass":
+    case "no_open_findings":
+    case "finding_count_below":
+      return 0;
+    case "and":
+    case "or":
+      return 1 + Math.max(...predicate.predicates.map(predicateDepth));
+    case "not":
+      return 1 + predicateDepth(predicate.predicate);
+  }
+}
+
+/** Maximum predicate nesting depth accepted at config load. */
+export const MAX_PREDICATE_DEPTH = 16;
diff --git a/packages/core/src/pipeline/reducer.ts b/packages/core/src/pipeline/reducer.ts
index 604d45651..ffa76e88b 100644
--- a/packages/core/src/pipeline/reducer.ts
+++ b/packages/core/src/pipeline/reducer.ts
@@ -12,7 +12,7 @@
  * reducer-helpers.ts.
  */
 
-import { scheduleAfterChange } from "./dag.js";
+import { evaluateRunExitOutcome, scheduleAfterChange } from "./dag.js";
 import type { PipelineEffect, PipelineEvent, ReducerResult } from "./events.js";
 import {
   deriveLoopStateFromRun,
@@ -123,6 +123,7 @@ function reduceTriggerFired(state: EngineState, event: TriggerFiredEvent): Reduc
       runs: { ...state.runs, [runId]: runState },
       currentRunByLoop: { ...state.currentRunByLoop, [key]: runId },
     };
+    const { reason, loopState: finalLoopState, observation } = terminalLoopForRun(runState);
     const preceding: PipelineEffect[] = [
       {
         type: "EMIT_OBSERVATION",
@@ -132,8 +133,9 @@ function reduceTriggerFired(state: EngineState, event: TriggerFiredEvent): Reduc
         },
       },
       ...skipObservations(runState.runId, sched.newlySkipped, runState),
+      ...(observation ? [observation] : []),
     ];
-    return terminateRunFromState(stateWithRun, runState, "completed", now, "done", preceding);
+    return terminateRunFromState(stateWithRun, runState, reason, now, finalLoopState, preceding);
   }
 
   const nextState: EngineState = {
@@ -271,7 +273,26 @@ function reduceStageCompleted(state: EngineState, event: StageCompletedEvent): R
     artifacts: [...stage.artifacts, ...newArtifacts.map((a) => a.artifactId)],
   };
 
-  return finalizeStageCompletion(state, run, stageName, updatedStage, newArtifacts, "success", now);
+  // Mirror materialized artifacts into the run-level buffer so
+  // `exitPredicates` can count/filter findings without round-tripping
+  // through the store. The buffer is overwritten per stage rather than
+  // appended: a stage's artifacts represent its current attempt, not the
+  // historical union (resume/retry semantics handle that elsewhere).
+  const priorArtifacts = run.runArtifacts ?? {};
+  const runWithArtifacts: RunState = {
+    ...run,
+    runArtifacts: { ...priorArtifacts, [stageName]: newArtifacts },
+  };
+
+  return finalizeStageCompletion(
+    state,
+    runWithArtifacts,
+    stageName,
+    updatedStage,
+    newArtifacts,
+    "success",
+    now,
+  );
 }
 
 interface StageFailedEvent {
@@ -417,10 +438,7 @@ function reduceRunResumed(state: EngineState, event: RunResumedEvent): ReducerRe
   for (const name of failedStageNames) {
     const fresh = stageRunIds[name];
     if (!fresh) {
-      return invalidTransition(
-        state,
-        `RUN_RESUMED missing stageRunId for failed stage "${name}"`,
-      );
+      return invalidTransition(state, `RUN_RESUMED missing stageRunId for failed stage "${name}"`);
     }
     const prior = run.stages[name];
     const cap = stageRetriesByName.get(name);
@@ -611,17 +629,20 @@ function finalizeStageCompletion(
   // Success path: the DAG scheduler may cascade-skip downstream stages whose
   // routes are now unsatisfied, and may immediately schedule the next batch
   // of parallel-eligible stages. Cascade-driven terminality is checked AFTER
-  // the cascade — only this can carry the run to `done` in one reducer step.
+  // the cascade — only this can carry the run to a terminal loop state in
+  // one reducer step.
   const sched = scheduleAfterChange(updatedRun, now);
   effects.push(...skipObservations(run.runId, sched.newlySkipped, sched.run));
 
   if (sched.allTerminal) {
+    const { reason, loopState: finalLoopState, observation } = terminalLoopForRun(sched.run);
+    if (observation) effects.push(observation);
     return terminateRunFromState(
       replaceRun(state, sched.run),
       sched.run,
-      "completed",
+      reason,
       now,
-      "done",
+      finalLoopState,
       effects,
     );
   }
@@ -631,3 +652,69 @@ function finalizeStageCompletion(
 
   return { state: replaceRun(state, sched.run), effects };
 }
+
+/**
+ * Decide how a fully-terminal run exits, applying `Pipeline.exitPredicates`.
+ * Returns the `terminateRunFromState` arguments + an optional observation so
+ * downstream consumers can distinguish loop_succeeded / loop_failed without
+ * inferring from `loopState` alone.
+ */
+function terminalLoopForRun(run: RunState): {
+  reason: RunTerminationReason;
+  loopState: LoopStateName;
+  observation?: PipelineEffect;
+} {
+  const outcome = evaluateRunExitOutcome(run);
+  if (outcome.kind === "succeeded") {
+    return {
+      reason: "completed",
+      loopState: "done",
+      observation: {
+        type: "EMIT_OBSERVATION",
+        event: {
+          name: "pipeline.loop.succeeded",
+          data: { runId: run.runId, pipelineName: run.pipelineName, loopRounds: run.loopRounds },
+        },
+      },
+    };
+  }
+  if (outcome.kind === "failed_exhausted") {
+    return {
+      reason: "completed",
+      loopState: "stalled",
+      observation: {
+        type: "EMIT_OBSERVATION",
+        event: {
+          name: "pipeline.loop.failed",
+          data: {
+            runId: run.runId,
+            pipelineName: run.pipelineName,
+            loopRounds: run.loopRounds,
+            reason: "exit_predicates_unsatisfied",
+          },
+        },
+      },
+    };
+  }
+  // failed_can_retry: the run is done but the loop awaits the next trigger.
+  // We still mark the run terminal (no more stages can fire) — the loop key
+  // gets freed by `terminateRunFromState` so the next TRIGGER_FIRED creates
+  // a fresh run. Using `awaiting_context` here advertises the intent on the
+  // run's final record without affecting cleanup.
+  return {
+    reason: "completed",
+    loopState: "awaiting_context",
+    observation: {
+      type: "EMIT_OBSERVATION",
+      event: {
+        name: "pipeline.loop.continuing",
+        data: {
+          runId: run.runId,
+          pipelineName: run.pipelineName,
+          loopRounds: run.loopRounds,
+          maxLoopRounds: run.pipelineConfigSnapshot.maxLoopRounds,
+        },
+      },
+    },
+  };
+}
diff --git a/packages/core/src/pipeline/stage-prompt.ts b/packages/core/src/pipeline/stage-prompt.ts
index 5467f53e6..b51f5a2a7 100644
--- a/packages/core/src/pipeline/stage-prompt.ts
+++ b/packages/core/src/pipeline/stage-prompt.ts
@@ -10,13 +10,22 @@
  * spawn-time `prompt` field. Keep it terse — agents read it once.
  */
 
-import { PIPELINE_FINDINGS_FILENAME, type Stage, type TaskMode } from "./types.js";
+import { PIPELINE_FINDINGS_FILENAME, type Artifact, type Stage, type TaskMode } from "./types.js";
 
 export interface StagePromptInput {
   pipelineName: string;
   stage: Stage;
   /** Loop counter from the engine — included so prompts surface progress. */
   loopRound?: number;
+  /**
+   * Artifacts from upstream sibling stages, keyed by stage name. Only consulted
+   * when `stage.workspaceClass === "read-siblings"`. Empty / unset = no sibling
+   * artifacts are surfaced (the default `independent` semantics).
+   *
+   * The executor is responsible for collecting these from the run's artifact
+   * store and threading them through; the prompt builder only formats them.
+   */
+  siblingArtifacts?: Record<string, Artifact[]>;
 }
 
 /**
@@ -26,7 +35,7 @@ export interface StagePromptInput {
  * agent doesn't need to know the absolute path.
  */
 export function buildStagePrompt(input: StagePromptInput): string {
-  const { pipelineName, stage, loopRound } = input;
+  const { pipelineName, stage, loopRound, siblingArtifacts } = input;
   const mode = stage.executor.kind === "agent" ? stage.executor.mode : null;
   const lines: string[] = [];
 
@@ -53,6 +62,15 @@ export function buildStagePrompt(input: StagePromptInput): string {
     lines.push("```");
   }
 
+  if (stage.workspaceClass === "read-siblings") {
+    const block = formatSiblingArtifactsBlock(siblingArtifacts);
+    if (block) {
+      lines.push(``);
+      lines.push(`## Upstream Artifacts`);
+      lines.push(block);
+    }
+  }
+
   lines.push(``);
   lines.push(`## Reporting Findings`);
   lines.push(formatFindingsInstructions(mode));
@@ -60,6 +78,25 @@ export function buildStagePrompt(input: StagePromptInput): string {
   return lines.join("\n");
 }
 
+/**
+ * Render upstream artifacts as a single JSON block. The executor harvests
+ * them from the artifact store; the prompt just exposes them so the agent
+ * can react to prior findings without rummaging through the workspace.
+ *
+ * Returns `null` when there's nothing to surface — caller omits the section
+ * entirely in that case.
+ */
+function formatSiblingArtifactsBlock(
+  siblingArtifacts: Record<string, Artifact[]> | undefined,
+): string | null {
+  if (!siblingArtifacts) return null;
+  const entries = Object.entries(siblingArtifacts).filter(([, arr]) => arr.length > 0);
+  if (entries.length === 0) return null;
+  const flat: Record<string, Artifact[]> = {};
+  for (const [name, arr] of entries) flat[name] = arr;
+  return `\`\`\`json\n${JSON.stringify(flat, null, 2)}\n\`\`\``;
+}
+
 function formatFindingsInstructions(mode: TaskMode | null): string {
   const path = `.ao/${PIPELINE_FINDINGS_FILENAME}`;
   const tmpPath = `${path}.tmp`;
diff --git a/packages/core/src/pipeline/types.ts b/packages/core/src/pipeline/types.ts
index 4858712c5..10a05d23b 100644
--- a/packages/core/src/pipeline/types.ts
+++ b/packages/core/src/pipeline/types.ts
@@ -87,9 +87,10 @@ export interface StageBudget {
 }
 
 /**
- * Hardcoded predicate forms for v1.1. The typed predicate DSL (and richer
- * `exitPredicates`) lands in v1.3 — this union is intentionally minimal so the
- * scheduler can ship without committing to a DSL surface yet.
+ * Hardcoded route predicates from v1.1. v1.3 introduces the typed `Predicate`
+ * DSL ({@link Predicate}); routes accept either shape so existing pipelines
+ * keep working without a rewrite. Internally `dag.ts` normalizes both into the
+ * same evaluator.
  *
  * All forms reference stage names within the same pipeline. Validation at
  * config load rejects unknown names; the scheduler trusts the input here.
@@ -109,11 +110,53 @@ export type StageRoutePredicate =
   | { kind: "anySucceeded"; stages: string[] }
   | { kind: "anyFailed"; stages: string[] };
 
+/**
+ * v1.3 typed predicate DSL used by `Stage.routes.when` (new form) and
+ * `Pipeline.exitPredicates`. JSON-schema validated at config load.
+ *
+ * Leaf semantics:
+ *  - `all_pass`             — every referenced stage's terminal status is `succeeded`
+ *                             (or, when verdict is set, verdict === "pass").
+ *  - `no_open_findings`     — no finding artifacts in scope have `status === "open"`.
+ *  - `finding_count_below`  — fewer than `n` open findings exist in scope.
+ *
+ * The optional `stages` field scopes the leaf to a subset of stages. When
+ * omitted, the leaf evaluates over every stage in the run (the natural default
+ * for `Pipeline.exitPredicates`). Routes always set `stages` since they're
+ * gating an individual stage on prior siblings.
+ */
+export type Predicate =
+  | { kind: "all_pass"; stages?: string[] }
+  | { kind: "no_open_findings"; stages?: string[] }
+  | { kind: "finding_count_below"; n: number; stages?: string[] }
+  | { kind: "and"; predicates: Predicate[] }
+  | { kind: "or"; predicates: Predicate[] }
+  | { kind: "not"; predicate: Predicate };
+
 export interface StageRoutes {
   /** Evaluated once every referenced upstream stage reaches a terminal state. */
-  when: StageRoutePredicate;
+  when: StageRoutePredicate | Predicate;
 }
 
+/**
+ * Workspace isolation class — see issue #1632.
+ *
+ *  - `independent`   (default): fresh session + fresh worktree; the stage sees
+ *                               nothing from prior sibling stages.
+ *  - `read-siblings`         : fresh session + fresh worktree, but the stage's
+ *                               prompt is augmented with read-only access to
+ *                               artifacts produced by upstream stages. Useful for
+ *                               "fix" stages that consume "review" findings.
+ *
+ * The original v0.1 spec had `shared-ro` / `isolated-rw` classes designed for a
+ * model where stages shared a workspace. Fresh-session-per-stage made that
+ * obsolete; the surviving distinction is just whether sibling artifacts are
+ * surfaced to the agent.
+ *
+ * Enforced by WorkspaceGuard at config load — see {@link ConfiguredPipelineSchema}.
+ */
+export type WorkspaceClass = "independent" | "read-siblings";
+
 export interface Stage {
   name: string;
   trigger: StageTrigger;
@@ -139,6 +182,12 @@ export interface Stage {
    * "all `dependsOn` stages must have succeeded".
    */
   routes?: StageRoutes;
+  /**
+   * v1.3 — workspace isolation class. Defaults to `independent`.
+   * `read-siblings` requires at least one upstream stage (`dependsOn` or any
+   * preceding stage); WorkspaceGuard rejects orphans at config load.
+   */
+  workspaceClass?: WorkspaceClass;
 }
 
 export interface Pipeline {
@@ -147,6 +196,26 @@ export interface Pipeline {
   stages: Stage[];
   /** Default 1 in v0; engine enforces serial execution when unset. */
   maxConcurrentStages?: number;
+  /**
+   * v1.3 — run-completion conditions. Evaluated when every stage in the run
+   * reaches a terminal status. When unset, defaults to the v1.1 behavior
+   * (success ⇔ allTerminal regardless of artifacts).
+   *
+   * Multiple predicates are AND-combined. The result selects the terminal
+   * loop state:
+   *   - true  → loop state `done`     (loop_succeeded)
+   *   - false → loop state `stalled`  (loop_failed) once `maxLoopRounds` is
+   *             reached; otherwise the run terminates as `awaiting_context`
+   *             so the next trigger can attempt another round.
+   */
+  exitPredicates?: Predicate[];
+  /**
+   * v1.3 — pipeline-level loop cap. When set, `loopRounds >= maxLoopRounds`
+   * combined with falsy `exitPredicates` produces the `loop_failed` terminal.
+   * Per-stage `maxLoopRounds` is unrelated — that one caps retries within a
+   * single run; this caps re-trigger rounds across runs.
+   */
+  maxLoopRounds?: number;
 }
 
 // ============================================================================
@@ -270,6 +339,14 @@ export interface RunState {
   loopRounds: number;
   /** Keyed by stage name. v0 has at most one entry per stage. */
   stages: Record<string, StageState>;
+  /**
+   * v1.3 — artifacts materialized by the reducer during this run, keyed by
+   * stage name (declaration order preserved by Object semantics is irrelevant
+   * — predicates iterate explicitly). Required for `exitPredicates` to
+   * count/filter findings without round-tripping through the store. Optional
+   * so v1.1 RunStates loaded from disk still parse.
+   */
+  runArtifacts?: Record<string, Artifact[]>;
   createdAt: string;
   updatedAt: string;
 }