From 0cd3b9fec5bd862ae4a9bd470271fc53c9960ed7 Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Wed, 30 Sep 2026 13:16:38 +0300 Subject: [PATCH 1/2] feat: support conditional edges in composed workflows --- .../skills/forge-workflow-authoring/SKILL.md | 3 +- .../references/workflow-format.md | 17 +- docs/reference/declarative-workflows.md | 33 +- src/forge/workflow/declarative/cli.py | 42 ++- src/forge/workflow/declarative/compiler.py | 107 ++++-- src/forge/workflow/declarative/composition.py | 32 +- src/forge/workflow/declarative/manifest.py | 37 +- src/forge/workflow/declarative/models.py | 54 ++- src/forge/workflow/declarative/predicates.py | 82 +++++ .../workflow/declarative/router_contracts.py | 335 ++++++++++++++++++ tests/unit/workflow/test_conditional_edges.py | 216 +++++++++++ 11 files changed, 911 insertions(+), 47 deletions(-) create mode 100644 src/forge/workflow/declarative/predicates.py create mode 100644 src/forge/workflow/declarative/router_contracts.py create mode 100644 tests/unit/workflow/test_conditional_edges.py diff --git a/.agents/skills/forge-workflow-authoring/SKILL.md b/.agents/skills/forge-workflow-authoring/SKILL.md index bfeeb1a7..ae41e7be 100644 --- a/.agents/skills/forge-workflow-authoring/SKILL.md +++ b/.agents/skills/forge-workflow-authoring/SKILL.md @@ -12,7 +12,7 @@ Help the user express a Forge process as readable YAML. Treat canonical JSON as 1. Read [references/workflow-format.md](references/workflow-format.md). 2. For a small local example, copy [assets/workflow.yaml](assets/workflow.yaml). For a publishable workflow, start from the closest full built-in definition and convert it to YAML if needed; publication requires the state profile's mandatory gates. 3. Establish the intended stages, decisions, loops, human pauses, and external commands before editing fields. -4. Run `forge workflow catalog STATE` and use only the nodes and routers it reports. Never invent catalog names. +4. Run `forge workflow catalog STATE` and use only the nodes, routers, and conditional facts it reports. Never invent catalog names. For reusable steps, inspect the built-in `github_pr_review` subworkflow or an active project definition. Bind every `@exit/name` in `spec.includes` and keep node names unique. 5. Keep the definition flow-only. Do not add step kinds, station contracts, effect capabilities, required or mandatory policies, extension declarations, observation policies, or external-entry flags. Forge derives and enforces those concerns from its trusted catalog and publication policy. @@ -50,6 +50,7 @@ Before publication, verify: - workflow transitions resolve to existing steps or `__end__`; subworkflow exits use `@exit/name` and every caller binds them; +- conditional cases use typed catalog facts, an explicit `otherwise`, and a trusted router on protected gates; - expected human and CI pause points remain present; - cycles cross an approved pause boundary; - exceptional commands such as PR rebasing are absent from graph topology; diff --git a/.agents/skills/forge-workflow-authoring/references/workflow-format.md b/.agents/skills/forge-workflow-authoring/references/workflow-format.md index b3ae25f8..63333989 100644 --- a/.agents/skills/forge-workflow-authoring/references/workflow-format.md +++ b/.agents/skills/forge-workflow-authoring/references/workflow-format.md @@ -16,7 +16,7 @@ complete workflow with `returnTo` for normal completion. Use `source: builtin` o project. Nodes remain canonical and cannot collide. New runs pin the fully expanded graph; active runs keep their existing pin. -Start at `entry`. Follow `next` for a fixed transition. At a step with `route`, follow every target in `branches`. A branch key is a possible router result; its value is the next step. `__end__` stops the current invocation and is not itself a declared step. In an included full workflow, fixed `next: __end__` returns to the caller's `returnTo` target; routed `__end__` still pauses the invocation. +Start at `entry`. Follow `next` for a fixed transition. At a step with `route`, follow every target in `branches`. A branch key is a possible router result; its value is the next step. A step may instead use ordered `cases` with predicates over catalog facts and a required `otherwise` target. `__end__` stops the current invocation and is not itself a declared step. In an included full workflow, fixed `next: __end__` returns to the caller's `returnTo` target; routed `__end__` still pauses the invocation. Use `forge workflow render FILE` instead of tracing a large definition manually. @@ -43,6 +43,21 @@ prd_approval_gate: Every possible static router result must be represented in `branches`. +A conditional transition uses the first matching case, then `otherwise`: + +```yaml +ci_evaluator: + cases: + - when: {fact: ci.status, op: in, value: [failed, blocked, no_prs]} + next: escalate_blocked + - when: {fact: ci.status, op: equals, value: fixing} + next: attempt_ci_fix + otherwise: human_review_gate +``` + +Inspect available facts with `forge workflow catalog STATE`. Supported predicates are `equals`, +`in`, `isNull`, and bounded `all`, `any`, and `not` groups. Protected gates require a trusted router. + Dynamic fan-out uses `dynamicRoute: true` and an explicit `maxConcurrency`. Forge derives the router's permitted destinations from the trusted catalog; inspect them with `forge workflow catalog STATE`. A join uses `join: all` or `join: any`. Copy the applicable shape from a validated built-in definition instead of reconstructing advanced routing from memory. ## Separation of concerns diff --git a/docs/reference/declarative-workflows.md b/docs/reference/declarative-workflows.md index 78e1270e..c7c54a82 100644 --- a/docs/reference/declarative-workflows.md +++ b/docs/reference/declarative-workflows.md @@ -81,7 +81,32 @@ forge workflow render src/forge/workflow/declarative/definitions/feature.json -- Authors may use YAML, as in the example above; publishing converts it to canonical JSON. In either format, the fields that describe the process are `spec.entry` and `spec.steps`. Each step declares -either a fixed `next` step or a named `route` with possible `branches`. +a fixed `next`, a named `route` with `branches`, or ordered conditional `cases` with `otherwise`. + +### Conditional edges + +Use conditional edges when a composed process needs its own destinations based on Forge-owned +facts. The first matching case wins; `otherwise` is required and may target `__end__`. Conditions +are pure and do not execute authored code. + +```yaml +ci_evaluator: + cases: + - when: {fact: ci.status, op: in, value: [failed, blocked, no_prs]} + next: escalate_blocked + - when: {fact: ci.status, op: equals, value: fixing} + next: attempt_ci_fix + otherwise: human_review_gate +``` + +Atomic predicates use `fact`, `op`, and, except for `isNull`, `value`. Supported operators are +`equals`, `in`, and `isNull`. Combine predicates with `all`, `any`, or `not`; groups are limited to +three nesting levels and eight members per group. `in` accepts 1–32 typed values. Run +`forge workflow catalog STATE` to inspect the version 1 fact names, types, and enumerated values. +Unknown facts and wrong value types fail validation. Missing string facts project as null; +missing boolean facts project as false. Existing static routers remain available, and protected +approval and review gates continue to use trusted routers. Conditional steps implicitly use fact +contract version 1; authors may write `factVersion: "1"` explicitly. ### Reusable definitions @@ -155,8 +180,8 @@ than asking users to edit canonical JSON directly. are also Jira label suffixes; subworkflows are reusable dependencies and cannot be selected. - `metadata.revision` must increase whenever content changes. - `spec.state` is `feature`, `bug`, or `task_takeover` and controls the available node catalog. -- Each step name is a canonical, registered Forge node. A step has either `next` or `route` with a - complete branch map. Use `__end__` to stop the current invocation. +- Each step name is a canonical, registered Forge node. A step has either `next`, `route` with a + complete branch map, or `cases` with `otherwise`. Use `__end__` to stop the current invocation. - Node kind, station contract, effect authority, mandatory policies, observation handling, and precondition contracts are owned by the trusted state-profile catalog. They are not workflow authoring fields. Older pinned definitions containing this metadata remain readable. @@ -226,7 +251,7 @@ forge workflow show-history MYPROJ prd-only Project authors can configure these flow-level choices: - the built-in state profile (`feature`, `bug`, or `task_takeover`); -- registered steps, fixed edges, router branches, joins, dynamic fan-out, retry +- registered steps, fixed edges, router branches, conditional cases over catalog facts, joins, dynamic fan-out, retry bounds, and concurrency; and - includes of built-in and same-project active subworkflows or workflows, with explicit exits or return targets; and diff --git a/src/forge/workflow/declarative/cli.py b/src/forge/workflow/declarative/cli.py index 498bbc39..91cd8a2b 100644 --- a/src/forge/workflow/declarative/cli.py +++ b/src/forge/workflow/declarative/cli.py @@ -69,6 +69,19 @@ async def cmd_workflow(args: Any) -> int: lines.append(f' {name}["{name}"]') if step.next: lines.append(f" {name} --> {step.next.replace('@exit/', 'exit_')}") + elif step.cases: + for index, case in enumerate(step.cases): + label = json.dumps( + case.when.model_dump(by_alias=True, exclude_none=True), + sort_keys=True, + ) + label = label.replace('"', """) + lines.append( + f" {name} -->|case {index + 1}: {label}| {case.next.replace('@exit/', 'exit_')}" + ) + lines.append( + f" {name} -->|otherwise| {step.otherwise.replace('@exit/', 'exit_')}" + ) else: for outcome, target in sorted(step.branches.items()): lines.append( @@ -111,6 +124,12 @@ async def cmd_workflow(args: Any) -> int: if action == "catalog": from forge.workflow.declarative.catalog import get_state_profile + from forge.workflow.declarative.predicates import ( + FACT_CONTRACT_VERSION, + FACTS, + PROFILE_FACTS, + ) + from forge.workflow.declarative.router_contracts import ROUTER_CONTRACTS profile = get_state_profile(args.state) catalog = { @@ -143,10 +162,31 @@ async def cmd_workflow(args: Any) -> int: {"dynamicTargets": sorted(profile.dynamic_router_targets[name])} if name in profile.dynamic_router_targets else {} - ) + ), + "outcomesByStep": { + step_name: sorted(outcomes) + for step_name, (router_name, outcomes) in sorted( + ROUTER_CONTRACTS[args.state].items() + ) + if router_name == name + }, } for name in sorted(profile.routers) }, + "conditionalFacts": { + "version": FACT_CONTRACT_VERSION, + "facts": { + name: { + "type": FACTS[name].value_type.__name__, + **( + {"values": sorted(FACTS[name].values)} + if FACTS[name].values is not None + else {} + ), + } + for name in sorted(PROFILE_FACTS[args.state]) + }, + }, "pauseNodes": sorted(profile.pause_nodes), "mandatoryPolicies": sorted(profile.mandatory_policies), "observationPolicies": sorted(profile.observation_policy_targets), diff --git a/src/forge/workflow/declarative/compiler.py b/src/forge/workflow/declarative/compiler.py index f274c2bb..a514d2fe 100644 --- a/src/forge/workflow/declarative/compiler.py +++ b/src/forge/workflow/declarative/compiler.py @@ -16,6 +16,8 @@ ) from forge.workflow.declarative.catalog import get_state_profile from forge.workflow.declarative.models import MAX_TRANSITIONS, WorkflowDefinition +from forge.workflow.declarative.predicates import evaluate_predicate, validate_predicate +from forge.workflow.declarative.router_contracts import validate_router_outcomes from forge.workflow.preconditions import NodeContract, project_capabilities, with_preconditions @@ -105,10 +107,22 @@ def validate(self) -> None: targets = ( [step.next] if step.next + else [*(case.next for case in step.cases), step.otherwise] + if step.cases else list(self.dynamic_targets(step)) if step.dynamic_route else list(step.branches.values()) ) + if step.cases: + if node_name in self.profile.mandatory_nodes or node_name.endswith("_gate"): + raise WorkflowValidationError( + f"protected gate '{node_name}' requires a trusted router" + ) + for case in step.cases: + try: + validate_predicate(case.when, spec.state) + except ValueError as exc: + raise WorkflowValidationError(f"step '{node_name}' {exc}") from exc for target in targets: if target == "__end__": has_terminal = True @@ -224,44 +238,19 @@ def validate_for_publication(self) -> None: f"publication omits mandatory gate '{sorted(missing_nodes)[0]}'" ) self.validate() - self._validate_golden_route_contracts() - - def _validate_golden_route_contracts(self) -> None: - """Keep custom routing within the reviewed golden-path outcome contract.""" - from forge.workflow.declarative.builtins import ( - builtin_bug_definition, - builtin_feature_definition, - builtin_task_takeover_definition, - ) + self._validate_router_contracts() - factories = { - "feature": builtin_feature_definition, - "bug": builtin_bug_definition, - "task_takeover": builtin_task_takeover_definition, - } - golden = factories[self.definition.spec.state]() + def _validate_router_contracts(self) -> None: + """Validate outcomes against reviewed router contracts, independent of topology.""" for node_name, step in self.definition.spec.steps.items(): - expected = golden.spec.steps.get(node_name) - if expected is None or not expected.route or step.route != expected.route: + if not step.route or step.dynamic_route: continue - expected_outcomes = ( - set(self.dynamic_targets(expected)) - if expected.dynamic_route - else set(expected.branches) - ) - declared_outcomes = ( - set(self.dynamic_targets(step)) if step.dynamic_route else set(step.branches) - ) - missing = expected_outcomes - declared_outcomes - if missing: - raise WorkflowValidationError( - f"step '{node_name}' omits router outcome '{sorted(missing)[0]}'" - ) - extra = declared_outcomes - expected_outcomes - if extra: - raise WorkflowValidationError( - f"step '{node_name}' adds unregistered router outcome '{sorted(extra)[0]}'" + try: + validate_router_outcomes( + self.definition.spec.state, node_name, step.route, set(step.branches) ) + except ValueError as exc: + raise WorkflowValidationError(str(exc)) from exc def build_graph(self) -> StateGraph[Any]: self.validate() @@ -277,6 +266,8 @@ def build_graph(self) -> StateGraph[Any]: contract=self.profile.contracts.get(node_name), retry_bound=step.retry_bound, allowed_effects=self.effective_effects(node_name), + cases=step.cases, + otherwise=step.otherwise, ), ) graph.set_entry_point("_forge_entry") @@ -296,6 +287,17 @@ def build_graph(self) -> StateGraph[Any]: ) continue + if step.cases: + targets = [*(case.next for case in step.cases), step.otherwise] + branches = {target: END if target == "__end__" else target for target in targets} + branches.setdefault("__end__", END) + graph.add_conditional_edges( + node_name, + self._conditional_route(step.cases, step.otherwise), + branches, + ) + continue + assert step.route is not None if step.dynamic_route: graph.add_conditional_edges( @@ -346,6 +348,8 @@ def _guarded_node( contract: NodeContract | None = None, retry_bound: int | None = None, allowed_effects: tuple[str, ...] = (), + cases: tuple[Any, ...] = (), + otherwise: str | None = None, ) -> Callable[..., Awaitable[dict[str, Any]]]: guarded_func = with_preconditions(func, contract, node_name=node_name) @@ -385,7 +389,16 @@ async def run(state: dict[str, Any]) -> dict[str, Any]: (result.get("last_error"), result.get("is_paused"), result.get("is_blocked")) ): result = {**result, "current_node": "complete", "is_paused": False} - target = str(result.get("current_node") or node_name) + selected = None + if cases and otherwise is not None: + selected = DeclarativeWorkflowCompiler._select_case( + cases, otherwise, {**state, **result} + ) + target = ( + selected[1] + if selected is not None + else str(result.get("current_node") or node_name) + ) occurred_at = str(result.get("updated_at") or state.get("updated_at") or "") transition = { "transition_id": stable_identity( @@ -401,6 +414,8 @@ async def run(state: dict[str, Any]) -> dict[str, Any]: "target": target, "occurred_at": occurred_at, } + if selected is not None: + transition["case"] = selected[0] history = list(state.get("transition_history") or []) return { **result, @@ -420,6 +435,28 @@ def route(state: dict[str, Any]) -> str: return route + @staticmethod + def _select_case( + cases: tuple[Any, ...], otherwise: str, state: dict[str, Any] + ) -> tuple[str, str]: + if state.get("is_blocked"): + return "blocked", "__end__" + for index, case in enumerate(cases): + if evaluate_predicate(case.when, state): + return str(index), case.next + return "otherwise", otherwise + + @staticmethod + def _conditional_route( + cases: tuple[Any, ...], otherwise: str | None + ) -> Callable[[dict[str, Any]], str]: + assert otherwise is not None + + def route(state: dict[str, Any]) -> str: + return DeclarativeWorkflowCompiler._select_case(cases, otherwise, state)[1] + + return route + @staticmethod def _guarded_router(func: Callable[..., Any], outcomes: set[str]) -> Callable[..., Any]: async def route(state: dict[str, Any]) -> str: diff --git a/src/forge/workflow/declarative/composition.py b/src/forge/workflow/declarative/composition.py index fb6c00f8..ca9ec399 100644 --- a/src/forge/workflow/declarative/composition.py +++ b/src/forge/workflow/declarative/composition.py @@ -28,8 +28,25 @@ def validate_subworkflow(definition: WorkflowDefinition) -> None: raise ValueError(f"node '{name}' is not registered for state '{state}'") if step.route and step.route not in profile.routers: raise ValueError(f"router '{step.route}' is not registered for state '{state}'") + if step.route and not step.dynamic_route: + from forge.workflow.declarative.router_contracts import validate_router_outcomes + + validate_router_outcomes(state, name, step.route, set(step.branches)) + if step.cases: + from forge.workflow.declarative.predicates import validate_predicate + + if name in profile.mandatory_nodes or name.endswith("_gate"): + raise ValueError(f"protected gate '{name}' requires a trusted router") + for case in step.cases: + validate_predicate(case.when, state) profile.effect_policies[name].resolve(step.allowed_effects) - targets = [step.next] if step.next else step.branches.values() + targets = ( + [step.next] + if step.next + else [*(case.next for case in step.cases), step.otherwise] + if step.cases + else step.branches.values() + ) for target in targets: if ( target != "__end__" @@ -89,7 +106,13 @@ async def expand( declared_exits = { target[6:] for step in child_steps.values() - for target in ([step.next] if step.next else step.branches.values()) + for target in ( + [step.next] + if step.next + else [*(case.next for case in step.cases), step.otherwise] + if step.cases + else step.branches.values() + ) if target and target.startswith("@exit/") } if set(include.exits) != declared_exits or include.return_to is not None: @@ -153,5 +176,10 @@ def _replace_targets(step: WorkflowStep, replacements: dict[str, str]) -> Workfl outcome: replacements.get(target, target) for outcome, target in step.branches.items() }, + "cases": tuple( + case.model_copy(update={"next": replacements.get(case.next, case.next)}) + for case in step.cases + ), + "otherwise": replacements.get(step.otherwise, step.otherwise), } ) diff --git a/src/forge/workflow/declarative/manifest.py b/src/forge/workflow/declarative/manifest.py index f4203e9f..20c9e9d3 100644 --- a/src/forge/workflow/declarative/manifest.py +++ b/src/forge/workflow/declarative/manifest.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json from collections.abc import Iterable, Mapping from enum import StrEnum from typing import Any @@ -23,6 +24,7 @@ class ProcessTransition(DomainModel): source: str target: str outcome: str | None = None + condition: dict[str, Any] | None = None class ProcessNode(DomainModel): @@ -245,6 +247,19 @@ def build_process_manifest(definition: WorkflowDefinition) -> ProcessManifest: ProcessTransition(source=name, target=target, outcome="dynamic") for target in compiler.dynamic_targets(step) ) + elif step.cases: + transitions.extend( + ProcessTransition( + source=name, + target=case.next, + outcome=f"case {index + 1}", + condition=case.when.model_dump(by_alias=True, exclude_none=True), + ) + for index, case in enumerate(step.cases) + ) + transitions.append( + ProcessTransition(source=name, target=step.otherwise, outcome="otherwise") + ) else: transitions.extend( ProcessTransition(source=name, target=target, outcome=outcome) @@ -280,7 +295,11 @@ def render_mermaid(manifest: ProcessManifest) -> str: manifest.transitions, key=lambda edge: (edge.source, edge.target, edge.outcome or ""), ): - label = f"|{transition.outcome}|" if transition.outcome else "" + if transition.condition is not None: + description = json.dumps(transition.condition, sort_keys=True, separators=(",", ":")) + label = f"|{transition.outcome}: {description.replace(chr(34), '"')}|" + else: + label = f"|{transition.outcome}|" if transition.outcome else "" lines.append(f" {transition.source} -->{label} {transition.target}") return "\n".join(lines) @@ -318,6 +337,12 @@ def step_signature(definition: WorkflowDefinition, name: str, step: Any) -> tupl step.next, step.route, tuple(sorted(step.branches.items())), + tuple( + (case.when.model_dump_json(by_alias=True, exclude_none=True), case.next) + for case in step.cases + ), + step.otherwise, + step.fact_version or ("1" if step.cases else None), step.dynamic_route, tuple(sorted(dynamic_targets)), effective_effects(definition, name), @@ -349,6 +374,16 @@ def transitions( (name, target, "dynamic") for target in profile.dynamic_router_targets.get(step.route, frozenset()) } + elif step.cases: + edges = { + ( + name, + case.next, + f"case:{index}:{case.when.model_dump_json(by_alias=True, exclude_none=True)}", + ) + for index, case in enumerate(step.cases) + } + edges.add((name, step.otherwise, "otherwise")) else: edges = {(name, target, outcome) for outcome, target in step.branches.items()} result[name] = frozenset(edges) diff --git a/src/forge/workflow/declarative/models.py b/src/forge/workflow/declarative/models.py index 417f4046..1c19a469 100644 --- a/src/forge/workflow/declarative/models.py +++ b/src/forge/workflow/declarative/models.py @@ -36,10 +36,51 @@ def validate_name(cls, value: str) -> str: return value +class WorkflowPredicate(StrictModel): + fact: str | None = None + op: Literal["equals", "in", "isNull"] | None = None + value: Any = None + all_of: tuple[WorkflowPredicate, ...] = Field( + default=(), alias="all", exclude_if=lambda value: not value + ) + any_of: tuple[WorkflowPredicate, ...] = Field( + default=(), alias="any", exclude_if=lambda value: not value + ) + not_: WorkflowPredicate | None = Field(default=None, alias="not") + + @model_validator(mode="after") + def validate_shape(self) -> WorkflowPredicate: + forms = sum( + (self.fact is not None, bool(self.all_of), bool(self.any_of), self.not_ is not None) + ) + if forms != 1: + raise ValueError("predicate requires exactly one of fact, all, any, or not") + if self.fact is not None: + if self.op is None: + raise ValueError("fact predicate requires op") + if self.op != "isNull" and self.value is None: + raise ValueError("predicate requires value") + if self.op == "isNull" and self.value is not None: + raise ValueError("isNull does not accept value") + elif self.op is not None or self.value is not None: + raise ValueError("group predicate cannot declare op or value") + if len(self.all_of) > 8 or len(self.any_of) > 8: + raise ValueError("predicate group may contain at most 8 items") + return self + + +class WorkflowCase(StrictModel): + when: WorkflowPredicate + next: str + + class WorkflowStep(StrictModel): next: str | None = None route: str | None = None branches: dict[str, str] = Field(default_factory=dict) + cases: tuple[WorkflowCase, ...] = Field(default=(), exclude_if=lambda value: not value) + otherwise: str | None = None + fact_version: Literal["1"] | None = Field(default=None, alias="factVersion") dynamic_route: bool = Field(default=False, alias="dynamicRoute") # Legacy router capability metadata. New definitions omit it; the trusted # router catalog owns the possible destinations. @@ -68,8 +109,17 @@ class WorkflowStep(StrictModel): @model_validator(mode="after") def validate_transition(self) -> WorkflowStep: - if bool(self.next) == bool(self.route): - raise ValueError("exactly one of 'next' or 'route' is required") + if sum((bool(self.next), bool(self.route), bool(self.cases))) != 1: + raise ValueError("exactly one of 'next', 'route', or 'cases' is required") + if self.cases: + if not self.otherwise: + raise ValueError("conditional cases require otherwise") + if len(self.cases) > MAX_BRANCHES: + raise ValueError(f"a conditional step may have at most {MAX_BRANCHES} cases") + elif self.otherwise is not None: + raise ValueError("otherwise is only valid with cases") + if self.fact_version is not None and not self.cases: + raise ValueError("factVersion is only valid with cases") if self.next and self.branches: raise ValueError("branches are only valid with 'route'") if self.route and not self.branches and not self.dynamic_route: diff --git a/src/forge/workflow/declarative/predicates.py b/src/forge/workflow/declarative/predicates.py new file mode 100644 index 00000000..d9429085 --- /dev/null +++ b/src/forge/workflow/declarative/predicates.py @@ -0,0 +1,82 @@ +"""Pure, typed facts and predicates for authored workflow edges.""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from typing import Any + +from forge.workflow.declarative.models import WorkflowPredicate + + +@dataclass(frozen=True) +class Fact: + value_type: type + project: Callable[[Mapping[str, Any]], Any] + values: frozenset[Any] | None = None + + +# Keep this registry narrow. Future template nodes can register a versioned, +# typed result projection without exposing arbitrary checkpoint contents. +FACTS: dict[str, Fact] = { + "workflow.paused": Fact(bool, lambda state: bool(state.get("is_paused"))), + "workflow.has_error": Fact(bool, lambda state: bool(state.get("last_error"))), + "ci.status": Fact( + str, + lambda state: state.get("ci_status"), + frozenset( + {"fixing", "failed", "blocked", "no_prs", "pending", "passed", "external_failure"} + ), + ), + "pr.merged": Fact(bool, lambda state: bool(state.get("pr_merged"))), + "review.revision_requested": Fact(bool, lambda state: bool(state.get("revision_requested"))), +} +PROFILE_FACTS: dict[str, frozenset[str]] = { + "feature": frozenset(FACTS), + "bug": frozenset(FACTS), + "task_takeover": frozenset(FACTS), +} +FACT_CONTRACT_VERSION = "1" + + +def validate_predicate(predicate: WorkflowPredicate, state_profile: str, depth: int = 0) -> None: + if depth > 3: + raise ValueError("predicate nesting exceeds 3 levels") + if predicate.fact is None: + children = (*predicate.all_of, *predicate.any_of) + if predicate.not_ is not None: + children = (*children, predicate.not_) + for child in children: + validate_predicate(child, state_profile, depth + 1) + return + if predicate.fact not in PROFILE_FACTS[state_profile]: + raise ValueError(f"unknown conditional fact '{predicate.fact}' for state '{state_profile}'") + fact = FACTS[predicate.fact] + if predicate.op == "isNull": + return + values = predicate.value if predicate.op == "in" else [predicate.value] + if predicate.op == "in" and (not isinstance(values, list) or not 1 <= len(values) <= 32): + raise ValueError("in predicate requires a list of 1 to 32 values") + for value in values: + if type(value) is not fact.value_type: + raise ValueError( + f"conditional fact '{predicate.fact}' requires {fact.value_type.__name__} values" + ) + if fact.values is not None and value not in fact.values: + raise ValueError(f"unknown value {value!r} for conditional fact '{predicate.fact}'") + + +def evaluate_predicate(predicate: WorkflowPredicate, state: Mapping[str, Any]) -> bool: + if predicate.all_of: + return all(evaluate_predicate(child, state) for child in predicate.all_of) + if predicate.any_of: + return any(evaluate_predicate(child, state) for child in predicate.any_of) + if predicate.not_ is not None: + return not evaluate_predicate(predicate.not_, state) + assert predicate.fact is not None + value = FACTS[predicate.fact].project(state) + if predicate.op == "isNull": + return value is None + if predicate.op == "equals": + return value == predicate.value + return value in predicate.value diff --git a/src/forge/workflow/declarative/router_contracts.py b/src/forge/workflow/declarative/router_contracts.py new file mode 100644 index 00000000..5e5a11ca --- /dev/null +++ b/src/forge/workflow/declarative/router_contracts.py @@ -0,0 +1,335 @@ +"""Reviewed output contracts for registered static routers by source node.""" + +ROUTER_CONTRACTS: dict[str, dict[str, tuple[str, frozenset[str]]]] = { + "bug": { + "analyze_bug": ( + "route_after_analyze_bug", + frozenset(["__end__", "escalate_blocked", "reflect_rca"]), + ), + "answer_question": ( + "route_after_answer", + frozenset(["plan_approval_gate", "rca_option_gate", "triage_gate"]), + ), + "attempt_ci_fix": ( + "route_current_node", + frozenset(["attempt_ci_fix", "ci_evaluator", "escalate_blocked", "human_review_gate"]), + ), + "ci_evaluator": ( + "route_ci_evaluation", + frozenset(["attempt_ci_fix", "escalate_blocked", "human_review_gate"]), + ), + "create_pr": ( + "route_after_pr_creation", + frozenset(["escalate_blocked", "teardown_workspace"]), + ), + "decompose_plan": ( + "route_after_decompose_plan", + frozenset(["__end__", "escalate_blocked", "setup_workspace"]), + ), + "human_review_gate": ( + "route_human_review_bug", + frozenset( + [ + "__end__", + "ci_evaluator", + "complete_tasks", + "implement_review", + "post_merge_summary", + ] + ), + ), + "implement_review": ( + "route_current_node", + frozenset( + [ + "escalate_blocked", + "human_review_gate", + "implement_review", + "review_response_gate", + ] + ), + ), + "implement_work": ( + "route_after_implementation", + frozenset(["escalate_blocked", "implement_work", "local_review"]), + ), + "local_review": ( + "route_after_local_review", + frozenset( + [ + "create_pr", + "escalate_blocked", + "implement_work", + "local_review", + "update_documentation", + ] + ), + ), + "plan_approval_gate": ( + "route_plan_approval", + frozenset(["__end__", "answer_question", "decompose_plan", "regenerate_plan"]), + ), + "plan_bug_fix": ( + "route_after_plan_bug_fix", + frozenset(["__end__", "escalate_blocked", "plan_approval_gate", "plan_bug_fix"]), + ), + "rca_option_gate": ( + "route_rca_option", + frozenset(["__end__", "answer_question", "plan_bug_fix", "regenerate_rca"]), + ), + "reflect_rca": ( + "route_after_reflect_rca", + frozenset(["__end__", "analyze_bug", "escalate_blocked", "rca_option_gate"]), + ), + "regenerate_plan": ( + "route_after_regenerate_plan", + frozenset(["__end__", "escalate_blocked", "plan_approval_gate", "regenerate_plan"]), + ), + "review_response_gate": ( + "route_review_response", + frozenset(["__end__", "human_review_gate", "implement_review"]), + ), + "setup_workspace": ( + "route_after_workspace_setup", + frozenset(["escalate_blocked", "implement_work"]), + ), + "teardown_workspace": ( + "route_after_teardown", + frozenset(["human_review_gate", "setup_workspace"]), + ), + "triage_check": ( + "route_current_node", + frozenset(["analyze_bug", "escalate_blocked", "triage_check", "triage_gate"]), + ), + "triage_gate": ("route_triage_gate", frozenset(["__end__", "triage_check"])), + }, + "feature": { + "answer_question": ( + "route_after_answer", + frozenset( + [ + "plan_approval_gate", + "prd_approval_gate", + "spec_approval_gate", + "task_approval_gate", + ] + ), + ), + "attempt_ci_fix": ( + "route_current_node", + frozenset(["attempt_ci_fix", "ci_evaluator", "escalate_blocked", "human_review_gate"]), + ), + "ci_evaluator": ( + "route_ci_evaluation", + frozenset(["attempt_ci_fix", "escalate_blocked", "human_review_gate"]), + ), + "create_pr": ( + "route_after_pr_creation", + frozenset(["escalate_blocked", "teardown_workspace"]), + ), + "decompose_epics": ( + "route_after_epic_decomposition", + frozenset(["__end__", "plan_approval_gate"]), + ), + "generate_prd": ("route_after_generation", frozenset(["__end__", "prd_approval_gate"])), + "generate_spec": ( + "route_after_spec_generation", + frozenset(["__end__", "spec_approval_gate"]), + ), + "generate_tasks": ( + "route_after_task_generation", + frozenset(["__end__", "task_approval_gate"]), + ), + "human_review_gate": ( + "route_human_review", + frozenset(["__end__", "ci_evaluator", "complete_tasks", "implement_review"]), + ), + "implement_review": ( + "route_current_node", + frozenset( + [ + "escalate_blocked", + "human_review_gate", + "implement_review", + "review_response_gate", + ] + ), + ), + "implement_work": ( + "route_implementation", + frozenset(["escalate_blocked", "implement_work", "local_review"]), + ), + "local_review": ( + "route_current_node", + frozenset(["create_pr", "escalate_blocked", "local_review"]), + ), + "plan_approval_gate": ( + "route_plan_approval", + frozenset( + [ + "__end__", + "answer_question", + "generate_tasks", + "provision_epics", + "regenerate_all_epics", + "update_single_epic", + ] + ), + ), + "prd_approval_gate": ( + "route_prd_approval", + frozenset(["__end__", "answer_question", "generate_spec", "regenerate_prd"]), + ), + "regenerate_all_epics": ( + "route_after_epic_regeneration", + frozenset(["__end__", "plan_approval_gate"]), + ), + "regenerate_all_tasks": ( + "route_after_task_regeneration", + frozenset(["__end__", "task_approval_gate"]), + ), + "regenerate_epic_tasks": ( + "route_after_epic_task_regeneration", + frozenset(["__end__", "task_approval_gate"]), + ), + "regenerate_prd": ( + "route_after_prd_regeneration", + frozenset(["__end__", "prd_approval_gate"]), + ), + "regenerate_spec": ( + "route_after_spec_regeneration", + frozenset(["__end__", "spec_approval_gate"]), + ), + "review_response_gate": ( + "route_review_response", + frozenset(["__end__", "human_review_gate", "implement_review"]), + ), + "setup_workspace": ( + "route_after_workspace_setup", + frozenset(["escalate_blocked", "implement_work"]), + ), + "spec_approval_gate": ( + "route_spec_approval", + frozenset(["__end__", "answer_question", "decompose_epics", "regenerate_spec"]), + ), + "task_approval_gate": ( + "route_task_approval", + frozenset( + [ + "__end__", + "answer_question", + "provision_tasks", + "regenerate_all_tasks", + "regenerate_epic_tasks", + "task_router", + "update_single_task", + ] + ), + ), + "teardown_workspace": ( + "route_after_teardown", + frozenset(["human_review_gate", "setup_workspace"]), + ), + "update_single_epic": ( + "route_after_single_epic_update", + frozenset(["__end__", "plan_approval_gate"]), + ), + "update_single_task": ( + "route_after_single_task_update", + frozenset(["__end__", "task_approval_gate"]), + ), + }, + "task_takeover": { + "answer_question": ("route_after_answer", frozenset(["task_plan_approval_gate"])), + "attempt_ci_fix": ( + "route_current_node", + frozenset(["attempt_ci_fix", "ci_evaluator", "escalate_blocked", "human_review_gate"]), + ), + "ci_evaluator": ( + "route_ci_evaluation", + frozenset(["attempt_ci_fix", "escalate_blocked", "human_review_gate"]), + ), + "create_pr": ( + "route_after_pr_creation", + frozenset(["escalate_blocked", "teardown_workspace"]), + ), + "generate_plan": ( + "route_after_generate_plan", + frozenset(["escalate_blocked", "generate_plan", "task_plan_approval_gate"]), + ), + "human_review_gate": ( + "route_human_review_task_takeover", + frozenset( + [ + "__end__", + "ci_evaluator", + "complete_task_takeover", + "complete_tasks", + "implement_review", + ] + ), + ), + "implement_review": ( + "route_current_node", + frozenset( + [ + "escalate_blocked", + "human_review_gate", + "implement_review", + "review_response_gate", + ] + ), + ), + "implement_work": ( + "route_after_execution", + frozenset(["escalate_blocked", "implement_work", "run_qualitative_review"]), + ), + "review_response_gate": ( + "route_review_response", + frozenset(["__end__", "human_review_gate", "implement_review"]), + ), + "run_qualitative_review": ( + "route_after_qualitative_review", + frozenset( + ["create_pr", "escalate_blocked", "implement_work", "run_qualitative_review"] + ), + ), + "setup_workspace": ( + "route_after_workspace_setup", + frozenset(["escalate_blocked", "implement_work"]), + ), + "task_plan_approval_gate": ( + "route_task_plan_approval", + frozenset(["__end__", "answer_question", "regenerate_plan", "setup_workspace"]), + ), + "teardown_workspace": ( + "route_after_teardown", + frozenset(["human_review_gate", "setup_workspace"]), + ), + "triage_check": ( + "route_after_triage_check", + frozenset(["escalate_blocked", "generate_plan", "triage_check", "triage_gate"]), + ), + "triage_gate": ("route_triage_gate", frozenset(["__end__", "triage_check"])), + }, +} + + +def validate_router_outcomes( + state: str, node_name: str, router_name: str, outcomes: set[str] +) -> None: + """Require every static router result, independent of authored destinations.""" + contract = ROUTER_CONTRACTS[state].get(node_name) + if contract is None or contract[0] != router_name: + raise ValueError( + f"router '{router_name}' has no reviewed outcome contract on '{node_name}'" + ) + expected = set(contract[1]) + missing = expected - outcomes + if missing: + raise ValueError(f"step '{node_name}' omits router outcome '{sorted(missing)[0]}'") + extra = outcomes - expected + if extra: + raise ValueError( + f"step '{node_name}' adds unregistered router outcome '{sorted(extra)[0]}'" + ) diff --git a/tests/unit/workflow/test_conditional_edges.py b/tests/unit/workflow/test_conditional_edges.py new file mode 100644 index 00000000..50e9bbdc --- /dev/null +++ b/tests/unit/workflow/test_conditional_edges.py @@ -0,0 +1,216 @@ +"""Conditional topology uses typed facts and remains revision-safe.""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from forge.workflow.declarative.builtins import builtin_feature_definition +from forge.workflow.declarative.compiler import DeclarativeWorkflowCompiler, WorkflowValidationError +from forge.workflow.declarative.composition import resolve_definition +from forge.workflow.declarative.manifest import build_process_manifest, compare_process_definitions +from forge.workflow.declarative.models import WorkflowDefinition + + +def _definition(steps: dict, *, revision: int = 1) -> WorkflowDefinition: + return WorkflowDefinition.model_validate( + { + "apiVersion": "forge/v1", + "kind": "Workflow", + "metadata": {"name": "conditional-feature", "revision": revision}, + "spec": {"state": "feature", "entry": "generate_prd", "steps": steps}, + } + ) + + +def _steps(target: str = "generate_spec") -> dict: + return { + "generate_prd": { + "cases": [ + { + "when": { + "all": [ + {"fact": "workflow.has_error", "op": "equals", "value": True}, + {"not": {"fact": "workflow.paused", "op": "equals", "value": True}}, + ] + }, + "next": "escalate_blocked", + }, + { + "when": {"fact": "workflow.paused", "op": "equals", "value": False}, + "next": target, + }, + ], + "otherwise": "__end__", + }, + "escalate_blocked": {"next": "__end__"}, + "generate_spec": {"next": "__end__"}, + } + + +def test_conditional_edges_compile_and_select_first_matching_case() -> None: + definition = _definition(_steps()) + compiler = DeclarativeWorkflowCompiler(definition) + compiler.build_graph().compile() + route = compiler._conditional_route(definition.spec.steps["generate_prd"].cases, "__end__") + + assert route({"last_error": "failed", "is_paused": False}) == "escalate_blocked" + assert route({"last_error": None, "is_paused": False}) == "generate_spec" + assert route({"is_paused": True}) == "__end__" + assert route({"is_blocked": True}) == "__end__" + + +@pytest.mark.asyncio +async def test_condition_selection_is_recorded_with_actual_target() -> None: + async def operation(state: dict) -> dict: + return {**state, "last_error": "failed", "is_paused": False} + + step = _definition(_steps()).spec.steps["generate_prd"] + guarded = DeclarativeWorkflowCompiler._guarded_node( + operation, "generate_prd", terminal=False, cases=step.cases, otherwise=step.otherwise + ) + result = await guarded({"ticket_key": "TEST-1"}) + + assert result["transition_history"][-1]["case"] == "0" + assert result["transition_history"][-1]["target"] == "escalate_blocked" + + +@pytest.mark.parametrize( + ("predicate", "message"), + [ + ({"fact": "ticket_key", "op": "equals", "value": "TEST-1"}, "unknown conditional fact"), + ({"fact": "workflow.paused", "op": "equals", "value": 1}, "requires bool values"), + ({"fact": "ci.status", "op": "in", "value": ["invented"]}, "unknown value"), + ], +) +def test_untrusted_or_mistyped_predicates_are_rejected(predicate: dict, message: str) -> None: + steps = _steps() + steps["generate_prd"]["cases"][0]["when"] = predicate + with pytest.raises(WorkflowValidationError, match=message): + DeclarativeWorkflowCompiler(_definition(steps)).validate() + + +def test_conditional_shape_requires_otherwise_and_excludes_router() -> None: + steps = _steps() + del steps["generate_prd"]["otherwise"] + with pytest.raises(ValidationError, match="require otherwise"): + _definition(steps) + + steps = _steps() + steps["generate_prd"]["route"] = "route_after_generation" + with pytest.raises(ValidationError, match="exactly one"): + _definition(steps) + + +def test_manifest_and_diff_include_predicates_even_when_target_is_unchanged() -> None: + previous = _definition(_steps()) + steps = _steps() + steps["generate_prd"]["cases"][1]["when"]["value"] = True + current = _definition(steps, revision=2) + + manifest = build_process_manifest(previous) + assert any(t.source == "generate_prd" and t.condition for t in manifest.transitions) + impact = compare_process_definitions(previous, current) + assert "generate_prd" in impact.changed_transitions + assert "generate_prd" in impact.routing_changes + assert not impact.compatible_for_in_flight + + +@pytest.mark.asyncio +async def test_subworkflow_exit_binds_conditional_targets() -> None: + child = WorkflowDefinition.model_validate( + { + "apiVersion": "forge/v1", + "kind": "Subworkflow", + "metadata": {"name": "conditional-child", "revision": 1}, + "spec": { + "state": "feature", + "entry": "generate_prd", + "steps": { + "generate_prd": { + "cases": [ + { + "when": { + "fact": "workflow.has_error", + "op": "equals", + "value": True, + }, + "next": "@exit/error", + } + ], + "otherwise": "@exit/done", + } + }, + }, + } + ) + parent = WorkflowDefinition.model_validate( + { + "apiVersion": "forge/v1", + "kind": "Workflow", + "metadata": {"name": "conditional-feature", "revision": 1}, + "spec": { + "state": "feature", + "entry": "generate_prd", + "steps": { + "escalate_blocked": {"next": "__end__"}, + "generate_spec": {"next": "__end__"}, + }, + "includes": [ + { + "source": "project", + "name": "conditional-child", + "exits": {"error": "escalate_blocked", "done": "generate_spec"}, + } + ], + }, + } + ) + + async def lookup(_name: str) -> WorkflowDefinition: + return child + + resolved = await resolve_definition(parent, lookup) + DeclarativeWorkflowCompiler(resolved).build_graph().compile() + step = resolved.spec.steps["generate_prd"] + assert step.cases[0].next == "escalate_blocked" + assert step.otherwise == "generate_spec" + + +def test_protected_gate_keeps_trusted_router() -> None: + raw = builtin_feature_definition().canonical_dict() + raw["metadata"]["revision"] += 1 + gate = raw["spec"]["steps"]["prd_approval_gate"] + gate.pop("route") + gate.pop("branches") + gate["cases"] = [ + { + "when": {"fact": "workflow.paused", "op": "equals", "value": False}, + "next": "generate_spec", + } + ] + gate["otherwise"] = "__end__" + with pytest.raises(WorkflowValidationError, match="protected gate"): + DeclarativeWorkflowCompiler( + WorkflowDefinition.model_validate(raw) + ).validate_for_publication() + + +def test_publishable_builtin_can_replace_ci_router_with_authored_conditions() -> None: + raw = builtin_feature_definition().canonical_dict() + raw["metadata"]["revision"] += 1 + ci = raw["spec"]["steps"]["ci_evaluator"] + ci.pop("route") + ci.pop("branches") + ci["cases"] = [ + { + "when": {"fact": "ci.status", "op": "in", "value": ["failed", "blocked", "no_prs"]}, + "next": "escalate_blocked", + }, + {"when": {"fact": "ci.status", "op": "equals", "value": "fixing"}, "next": "attempt_ci_fix"}, + ] + ci["otherwise"] = "human_review_gate" + + candidate = WorkflowDefinition.model_validate(raw) + DeclarativeWorkflowCompiler(candidate).validate_for_publication() + DeclarativeWorkflowCompiler(candidate).build_graph().compile() From fea6748a3f32242986bd21e50dd217b5648a9c15 Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Wed, 30 Sep 2026 15:41:10 +0300 Subject: [PATCH 2/2] fix: validate conditional branches and clarify included completion --- .../references/workflow-format.md | 2 +- docs/reference/declarative-workflows.md | 6 +- src/forge/workflow/declarative/composition.py | 5 +- src/forge/workflow/declarative/models.py | 2 +- tests/unit/workflow/test_conditional_edges.py | 87 ++++++++++++++++++- 5 files changed, 96 insertions(+), 6 deletions(-) diff --git a/.agents/skills/forge-workflow-authoring/references/workflow-format.md b/.agents/skills/forge-workflow-authoring/references/workflow-format.md index 63333989..f776fdd7 100644 --- a/.agents/skills/forge-workflow-authoring/references/workflow-format.md +++ b/.agents/skills/forge-workflow-authoring/references/workflow-format.md @@ -16,7 +16,7 @@ complete workflow with `returnTo` for normal completion. Use `source: builtin` o project. Nodes remain canonical and cannot collide. New runs pin the fully expanded graph; active runs keep their existing pin. -Start at `entry`. Follow `next` for a fixed transition. At a step with `route`, follow every target in `branches`. A branch key is a possible router result; its value is the next step. A step may instead use ordered `cases` with predicates over catalog facts and a required `otherwise` target. `__end__` stops the current invocation and is not itself a declared step. In an included full workflow, fixed `next: __end__` returns to the caller's `returnTo` target; routed `__end__` still pauses the invocation. +Start at `entry`. Follow `next` for a fixed transition. At a step with `route`, follow every target in `branches`. A branch key is a possible router result; its value is the next step. A step may instead use ordered `cases` with predicates over catalog facts and a required `otherwise` target. `__end__` stops the current invocation and is not itself a declared step. In an included full workflow, fixed `next: __end__` returns to the caller's `returnTo` target; routed and conditional `__end__` still stop the invocation. Include a fixed completion step or use a `Subworkflow` with named exits for conditional returns. Use `forge workflow render FILE` instead of tracing a large definition manually. diff --git a/docs/reference/declarative-workflows.md b/docs/reference/declarative-workflows.md index c7c54a82..de3fa550 100644 --- a/docs/reference/declarative-workflows.md +++ b/docs/reference/declarative-workflows.md @@ -114,8 +114,10 @@ Use `kind: Subworkflow` to publish a reusable group of registered steps. Give it steps, and named exit targets such as `@exit/review`. A subworkflow can list `spec.compatibleStates` when the same nodes and routers are valid in several profiles. It is not selectable by a ticket label. A normal `Workflow` can also be included; its fixed -`next: __end__` completion returns to the caller, while routed `__end__` outcomes continue -to pause the invocation. +`next: __end__` completion returns to the caller, while routed or conditional `__end__` +outcomes stop the current invocation. A full workflow included with `returnTo` therefore +needs at least one fixed `next: __end__` completion step. Use a `Subworkflow` with named +exits when distinct conditional outcomes should return to distinct caller steps. ```yaml apiVersion: forge/v1 diff --git a/src/forge/workflow/declarative/composition.py b/src/forge/workflow/declarative/composition.py index ca9ec399..e6f06180 100644 --- a/src/forge/workflow/declarative/composition.py +++ b/src/forge/workflow/declarative/composition.py @@ -131,7 +131,10 @@ async def expand( raise ValueError(f"workflow '{include.name}' requires returnTo and no exits") terminals = [name for name, step in child_steps.items() if step.next == "__end__"] if not terminals: - raise ValueError(f"workflow '{include.name}' has no normal completion") + raise ValueError( + f"workflow '{include.name}' has no fixed normal completion; " + "add a step with next: __end__ or use Subworkflow exits" + ) child_steps = { name: _replace_targets(step, {"__end__": include.return_to}) if name in terminals diff --git a/src/forge/workflow/declarative/models.py b/src/forge/workflow/declarative/models.py index 1c19a469..452200a8 100644 --- a/src/forge/workflow/declarative/models.py +++ b/src/forge/workflow/declarative/models.py @@ -120,7 +120,7 @@ def validate_transition(self) -> WorkflowStep: raise ValueError("otherwise is only valid with cases") if self.fact_version is not None and not self.cases: raise ValueError("factVersion is only valid with cases") - if self.next and self.branches: + if self.branches and not self.route: raise ValueError("branches are only valid with 'route'") if self.route and not self.branches and not self.dynamic_route: raise ValueError("a routed step requires non-empty branches") diff --git a/tests/unit/workflow/test_conditional_edges.py b/tests/unit/workflow/test_conditional_edges.py index 50e9bbdc..181121c6 100644 --- a/tests/unit/workflow/test_conditional_edges.py +++ b/tests/unit/workflow/test_conditional_edges.py @@ -101,6 +101,11 @@ def test_conditional_shape_requires_otherwise_and_excludes_router() -> None: with pytest.raises(ValidationError, match="exactly one"): _definition(steps) + steps = _steps() + steps["generate_prd"]["branches"] = {"ignored": "missing_step"} + with pytest.raises(ValidationError, match="branches are only valid with 'route'"): + _definition(steps) + def test_manifest_and_diff_include_predicates_even_when_target_is_unchanged() -> None: previous = _definition(_steps()) @@ -207,10 +212,90 @@ def test_publishable_builtin_can_replace_ci_router_with_authored_conditions() -> "when": {"fact": "ci.status", "op": "in", "value": ["failed", "blocked", "no_prs"]}, "next": "escalate_blocked", }, - {"when": {"fact": "ci.status", "op": "equals", "value": "fixing"}, "next": "attempt_ci_fix"}, + { + "when": {"fact": "ci.status", "op": "equals", "value": "fixing"}, + "next": "attempt_ci_fix", + }, ] ci["otherwise"] = "human_review_gate" candidate = WorkflowDefinition.model_validate(raw) DeclarativeWorkflowCompiler(candidate).validate_for_publication() DeclarativeWorkflowCompiler(candidate).build_graph().compile() + + +@pytest.mark.asyncio +async def test_included_workflow_keeps_conditional_pause_and_rewrites_fixed_completion() -> None: + child = WorkflowDefinition.model_validate( + { + "apiVersion": "forge/v1", + "kind": "Workflow", + "metadata": {"name": "conditional-child", "revision": 1}, + "spec": { + "state": "feature", + "entry": "generate_prd", + "steps": { + "generate_prd": { + "cases": [ + { + "when": {"fact": "workflow.paused", "op": "equals", "value": True}, + "next": "__end__", + } + ], + "otherwise": "generate_spec", + }, + "generate_spec": {"next": "__end__"}, + }, + }, + } + ) + parent = WorkflowDefinition.model_validate( + { + "apiVersion": "forge/v1", + "kind": "Workflow", + "metadata": {"name": "conditional-parent", "revision": 1}, + "spec": { + "state": "feature", + "entry": "answer_question", + "steps": { + "answer_question": {"next": "generate_prd"}, + "teardown_workspace": {"next": "__end__"}, + }, + "includes": [ + { + "source": "project", + "name": "conditional-child", + "returnTo": "teardown_workspace", + } + ], + }, + } + ) + + async def lookup(_name: str) -> WorkflowDefinition: + return child + + resolved = await resolve_definition(parent, lookup) + assert resolved.spec.steps["generate_prd"].cases[0].next == "__end__" + assert resolved.spec.steps["generate_spec"].next == "teardown_workspace" + DeclarativeWorkflowCompiler(resolved).validate() + + conditional_only = child.model_copy( + update={ + "spec": child.spec.model_copy( + update={ + "steps": { + "generate_prd": child.spec.steps["generate_prd"].model_copy( + update={"otherwise": "__end__"} + ) + } + } + ) + } + ) + + async def lookup_without_completion(_name: str) -> WorkflowDefinition: + return conditional_only + + with pytest.raises(ValueError, match="no fixed normal completion"): + await resolve_definition(parent, lookup_without_completion)