feat(pipeline): v1.3 — workspace classes, predicate DSL, exitPredicates
Adds three pieces that round out v1 (issue #1632, sub-task of #1349): ## Typed predicate DSL (pipeline/predicate.ts) - Leaves: all_pass, no_open_findings, finding_count_below - Boolean ops: and, or, not - Recursive Zod schema with nesting depth cap (MAX_PREDICATE_DEPTH=16) - Pure evaluator over (stages, artifactsByStage) context - Legacy v1.1 StageRoutePredicate shapes still parse — bridged via fromLegacyRoutePredicate so existing pipelines need no rewrites - collectReferencedStages walks both legacy + DSL forms (used by dag.ts for upstream-ref discovery and by config-schema for ref validation) ## Pipeline.exitPredicates (reducer.ts, dag.ts) - Replaces hardcoded "allTerminal → done" path in the reducer - New helper evaluateRunExitOutcome returns succeeded / failed_exhausted / failed_can_retry based on Pipeline.maxLoopRounds - Loop state mapping: * predicates true (or unset) → done (loop_succeeded) * predicates false, rounds out → stalled (loop_failed) * predicates false, rounds remain → awaiting_context (re-trigger) - Emits pipeline.loop.succeeded / pipeline.loop.failed / pipeline.loop.continuing observations alongside run.terminated - RunState gains optional runArtifacts buffer so the reducer can evaluate finding-based predicates without round-tripping the store ## Workspace classes (types.ts, config-schema.ts, stage-prompt.ts, engine.ts) - Stage.workspaceClass: "independent" (default) | "read-siblings" - WorkspaceGuard at config load rejects orphan read-siblings stages (no dependsOn, no route refs, no prior stage in declaration order) - Engine threads transitive upstream artifacts to read-siblings stages - buildStagePrompt surfaces them as an Upstream Artifacts JSON block ## CONFIG_CHANGED structural vs tuning (pipeline/config-diff.ts) - classifyConfigChange returns "none" | "tuning" | "structural" - Structural: stage list changes, dependsOn / routes / executor edits, pipeline rename → must terminate the active run - Tuning: predicate thresholds, retries, budgets, exitPredicates, maxLoopRounds, workspaceClass → engine can hot-reload - Short-circuits on first structural diff for cheap classification ## Tests (57 new vitests) - pipeline-predicate.test.ts (25): schema + evaluator + legacy bridge - pipeline-workspace.test.ts (10): WorkspaceGuard + prompt injection - pipeline-exit-predicates.test.ts (5): reducer integration - pipeline-config-diff.test.ts (17): structural vs tuning classification
This commit is contained in:
Harsh Batheja
parent
9144c0ec06
commit
e599509adc
|
|
@ -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/);
|
||||
});
|
||||
});
|
||||
|
|
@ -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();
|
||||
});
|
||||
});
|
||||
|
|
@ -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"]);
|
||||
});
|
||||
});
|
||||
|
|
@ -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");
|
||||
});
|
||||
});
|
||||
|
|
@ -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);
|
||||
}
|
||||
|
|
@ -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) };
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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";
|
||||
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
@ -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,
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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`;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue