feat: support conditional edges in composed workflows

.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;

.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 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.
 
@@ -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

docs/reference/declarative-workflows.md

@@ -81,16 +81,43 @@ 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
 
 Use `kind: Subworkflow` to publish a reusable group of registered steps. Give it an entry,
 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
@@ -157,8 +184,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.
@@ -228,7 +255,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

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),

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: