Implement control-flow routing: on_false, on_success, on_failure (#333)

docs/actions.md

@@ -81,3 +81,24 @@ written as instructions rather than identifiers.
 handed to the model to interpret. That is an explicit request, which is why
 it remains supported while the implicit "unrecognised action becomes a
 prompt" behaviour does not.
+
+## Control-flow routing
+
+A step may name where execution goes next. Routing jumps **forward**;
+steps between the source and the target are marked `skipped`. Skipping is
+not failing -- a pipeline that routed around a step still succeeds.
+
+| Key | Fires when |
+|-|-|
+| `on_false` | the step's own `condition:` was false, so it was skipped |
+| `on_success` | the step ran and succeeded |
+| `on_failure` | the step ran and did not succeed |
+
+`on_failure` means two things, told apart by value. These are failure
+**policies**: `continue`, `fail`, `retry`, `skip`. Any other value names a **step to jump to**,
+and the run continues there instead of aborting.
+
+A step whose id is one of those policy words is refused at compile time,
+since routing to it could not be distinguished from selecting the policy.
+Routing targets are checked at compile time too: a jump to a step that
+does not exist, or a step routing to itself, is exit 2.

docs/adr/0001-product-contract.md

@@ -114,6 +114,49 @@ missing extra must degrade one feature, never break the package import.
 Exit codes: `0` success, `1` execution failure, `2` validation/compile failure,
 `130` interrupted.
 
+## Control-flow routing
+
+A step may say where execution goes next:
+
+```yaml
+- id: check_topic
+  action: evaluate_condition
+  condition: "{{ topic | length > 10 }}"
+  parameters:
+    condition: "{{ topic | length > 10 }}"
+  on_false: short_topic_handler
+  on_success: main_processing
+  on_failure: error_recovery
+```
+
+| Key | Fires when |
+|-|-|
+| `on_false` | the step's own `condition:` was false, so the step was skipped |
+| `on_success` | the step ran and succeeded |
+| `on_failure` | the step ran and did not succeed |
+
+Routing jumps **forward**. Steps between the source and the target are marked
+`skipped`, which is the same machinery `goto` already used rather than a second
+one beside it. Skipping is not failing: a pipeline that routed around a step
+still reports `success`.
+
+"Did not succeed" means `StepResult.success`, not status — a tool returning
+`{"success": False}` without raising leaves its task `completed`, and routing
+on status alone sent a failing step down the success path.
+
+**`on_failure` means two things, told apart by value.** It already selected a
+failure *policy* — `fail`, `continue`, `skip`, `retry` — and now also names a
+step to jump to. A reserved policy word keeps its policy meaning; anything else
+is a step id. When `on_failure` names a step, the run continues there instead
+of aborting.
+
+That is only safe because the ambiguous case is refused rather than guessed:
+a pipeline containing a step whose id *is* a policy word fails to compile,
+because routing to it could not be distinguished from selecting the policy.
+
+Routing targets are validated at compile time — a jump to a step that does not
+exist, or a step routing to itself, is exit 2 naming the offending target.
+
 ## The result contract
 
 `Orchestrator.execute_pipeline` returns a `PipelineResult`

scripts/generate_action_docs.py

@@ -20,6 +20,7 @@
 sys.path.insert(0, str(ROOT / "src"))
 
 from orchestrator.core.actions import ACTION_FAMILIES, ACTION_SPECS  # noqa: E402
+from orchestrator.core.routing import FAILURE_POLICIES  # noqa: E402
 
 TARGET = ROOT / "docs" / "actions.md"
 
@@ -106,6 +107,33 @@ def render() -> str:
         "it remains supported while the implicit \"unrecognised action becomes a\n"
         "prompt\" behaviour does not.\n"
     )
+
+    lines.append("## Control-flow routing\n")
+    lines.append(
+        "A step may name where execution goes next. Routing jumps **forward**;\n"
+        "steps between the source and the target are marked `skipped`. Skipping is\n"
+        "not failing -- a pipeline that routed around a step still succeeds.\n"
+    )
+    lines.append("| Key | Fires when |")
+    lines.append("|-|-|")
+    lines.append(
+        "| `on_false` | the step's own `condition:` was false, so it was skipped |"
+    )
+    lines.append("| `on_success` | the step ran and succeeded |")
+    lines.append("| `on_failure` | the step ran and did not succeed |")
+    lines.append("")
+    policies = ", ".join(f"`{p}`" for p in sorted(FAILURE_POLICIES))
+    lines.append(
+        f"`on_failure` means two things, told apart by value. These are failure\n"
+        f"**policies**: {policies}. Any other value names a **step to jump to**,\n"
+        f"and the run continues there instead of aborting.\n"
+    )
+    lines.append(
+        "A step whose id is one of those policy words is refused at compile time,\n"
+        "since routing to it could not be distinguished from selecting the policy.\n"
+        "Routing targets are checked at compile time too: a jump to a step that\n"
+        "does not exist, or a step routing to itself, is exit 2.\n"
+    )
     return "\n".join(lines)
 
 

src/orchestrator/compiler/schema_validator.py

@@ -133,9 +133,20 @@ def _get_default_schema(self) -> Dict[str, Any]:
                                     },
                                     "timeout": {"type": "integer", "minimum": 1},
                                     "max_retries": {"type": "integer", "minimum": 0},
-                                    "on_failure": {
+                                    # Either a failure policy or a step to jump
+                                    # to (#333). Which one is decided by the
+                                    # value; `validate_dependencies` checks
+                                    # that a non-policy value names a real
+                                    # step, so an enum here would reject valid
+                                    # routing outright.
+                                    "on_failure": {"type": "string"},
+                                    "on_false": {
                                         "type": "string",
-                                        "enum": ["continue", "fail", "retry", "skip"],
+                                        "pattern": "^[a-zA-Z][a-zA-Z0-9_-]*$",
+                                    },
+                                    "on_success": {
+                                        "type": "string",
+                                        "pattern": "^[a-zA-Z][a-zA-Z0-9_-]*$",
                                     },
                                     "requires_model": {
                                         "oneOf": [

src/orchestrator/compiler/yaml_compiler.py

@@ -1346,6 +1346,13 @@ def _build_task(self, task_def: Dict[str, Any], available_steps: List[str]) -> T
         # Add additional metadata from task definition
         if "on_failure" in task_def:
             metadata["on_failure"] = task_def["on_failure"]
+
+        # Control-flow routing (#333). `on_failure` is deliberately not copied
+        # again here: it is already above, and whether it names a policy or a
+        # step is decided by its value, not by a second key.
+        for routing_key in ("on_false", "on_success"):
+            if routing_key in task_def:
+                metadata[routing_key] = task_def[routing_key]
         if "requires_model" in task_def:
             metadata["requires_model"] = task_def["requires_model"]
         if "tool" in task_def:

src/orchestrator/core/routing.py

@@ -0,0 +1,62 @@
+"""Control-flow routing between steps.
+
+A step may name where execution goes next:
+
+    - id: check_topic
+      action: evaluate_condition
+      parameters:
+        condition: "{{ topic | length > 10 }}"
+      on_false: short_topic_handler
+      on_success: main_processing
+      on_failure: error_recovery
+
+`on_false` applies when the step's own `condition:` evaluates false, so the
+step is skipped. `on_success` and `on_failure` apply to how the step itself
+ended. Routing jumps forward: steps between the source and the target are
+marked skipped, which is the same machinery `goto` already used.
+
+## The `on_failure` collision
+
+`on_failure` already existed, meaning a *failure policy*: one of `fail`,
+`continue`, `skip`, `retry`. #333 asks for the same key to name a step to jump
+to. Both readings are useful and both are now supported, disambiguated by
+value: a reserved policy word keeps its policy meaning, anything else is a
+step id.
+
+That is only safe because the ambiguity is refused rather than guessed at --
+naming a step `retry` and routing to it would silently become a policy, so a
+pipeline that does so fails at compile time instead.
+"""
+
+from __future__ import annotations
+
+from typing import FrozenSet
+
+#: Values of `on_failure` that select a policy rather than a routing target.
+FAILURE_POLICIES: FrozenSet[str] = frozenset({"fail", "continue", "skip", "retry"})
+
+#: Step keys that name another step to jump to.
+ROUTING_KEYS: FrozenSet[str] = frozenset({"on_false", "on_success", "on_failure"})
+
+
+def is_failure_policy(value: object) -> bool:
+    """Whether an `on_failure` value selects a policy instead of a step."""
+    return isinstance(value, str) and value.strip().lower() in FAILURE_POLICIES
+
+
+def routing_targets(step: dict) -> dict:
+    """The routing keys of `step` that name a step id, as {key: target}.
+
+    `on_failure` is included only when its value is not a reserved policy.
+    """
+    targets = {}
+    for key in ("on_false", "on_success"):
+        value = step.get(key)
+        if isinstance(value, str) and value.strip():
+            targets[key] = value.strip()
+
+    on_failure = step.get("on_failure")
+    if isinstance(on_failure, str) and on_failure.strip() and not is_failure_policy(on_failure):
+        targets["on_failure"] = on_failure.strip()
+
+    return targets

src/orchestrator/orchestrator.py

@@ -12,6 +12,7 @@
 from .core.error_handler import ErrorHandler
 from .core.pipeline import Pipeline
 from .core.pipeline_result import PipelineResult, StepResult
+from .core.routing import is_failure_policy
 from .core.pipeline_status_tracker import PipelineStatusTracker
 from .core.pipeline_resume_manager import PipelineResumeManager, ResumeStrategy
 from .core.resource_allocator import ResourceAllocator
@@ -1029,9 +1030,23 @@ async def _process_goto_jumps(
         # Check recently completed tasks for goto directives
         for task_id in completed_tasks:
             task = pipeline.get_task(task_id)
-            if not task or task.status != TaskStatus.COMPLETED:
+            if not task:
                 continue
-
+
+            # Control-flow routing (#333), which jumps the same way `goto`
+            # does. Unlike goto it also applies to steps that were skipped or
+            # that failed, so the status filter cannot come first.
+            routed = self._routing_target(task, results.get(task_id))
+            if routed and routed in pipeline.tasks:
+                self.logger.info(
+                    "Task %s routing to %s (%s)", task_id, routed, task.status.value
+                )
+                self._skip_tasks_between(pipeline, task_id, routed)
+                return routed
+
+            if task.status != TaskStatus.COMPLETED:
+                continue
+
             # Check for goto in task metadata
             goto_target = task.metadata.get("goto")
             if goto_target:
@@ -1048,6 +1063,34 @@ async def _process_goto_jumps(
 
         return None
 
+    def _routing_target(self, task: Task, value: Any = None) -> Optional[str]:
+        """Where this task's outcome says to go next, if anywhere.
+
+        `on_false` fires when the step's own `condition:` was false, which is
+        recorded as a skip. `on_success` and `on_failure` fire on how the step
+        itself ended, and "ended badly" is not the same as status FAILED: a
+        tool returning {"success": False} without raising leaves its task
+        COMPLETED. StepResult already owns that distinction, so it is reused
+        here rather than restated -- routing on status alone sent a failing
+        step down the success path.
+
+        `on_failure` names a step only when its value is not one of the
+        reserved failure policies -- see core/routing.py.
+        """
+        if task.status is TaskStatus.SKIPPED:
+            return task.metadata.get("on_false")
+
+        succeeded = StepResult.from_task(task, value).success
+
+        if succeeded:
+            return task.metadata.get("on_success")
+
+        on_failure = task.metadata.get("on_failure")
+        if isinstance(on_failure, str) and not is_failure_policy(on_failure):
+            return on_failure.strip() or None
+
+        return None
+
     def _skip_tasks_between(self, pipeline: Pipeline, from_task: str, to_task: str) -> None:
         """Skip tasks between a goto source and target.
 
@@ -2059,6 +2102,11 @@ async def _handle_task_failures(
             task = pipeline.get_task(task_id)
             failure_policy = task.metadata.get("on_failure", "fail")
 
+            # `on_failure` naming a step is routing, not a policy: the run
+            # continues at that step rather than aborting here (#333).
+            if not is_failure_policy(failure_policy):
+                continue
+
             if failure_policy == "continue":
                 # Continue with other tasks
                 continue

src/orchestrator/validation/dependency_validator.py

@@ -23,6 +23,7 @@
     nx = None
 
 from ..core.exceptions import ValidationError
+from ..core.routing import FAILURE_POLICIES, routing_targets
 
 logger = logging.getLogger(__name__)
 
@@ -392,7 +393,54 @@ def _validate_dependency_references(self, tasks: List[Dict[str, Any]]) -> List[D
                         dependency_chain=[task_id, task_id],
                         recommendation=f"Remove self-dependency from task '{task_id}'"
                     ))
-
+
+        # Control-flow routing targets (#333). A jump to a step that does not
+        # exist has to fail at compile time with the bad name, not at runtime
+        # when the executor tries to route into nothing.
+        for task in tasks:
+            task_id = task.get("id")
+            if not task_id:
+                continue
+            for key, target in routing_targets(task).items():
+                if target not in task_ids:
+                    issues.append(DependencyIssue(
+                        issue_type="missing_routing_target",
+                        severity="error",
+                        message=(
+                            f"Task '{task_id}' has {key}: '{target}', which is "
+                            f"not a step in this pipeline"
+                        ),
+                        involved_tasks=[task_id, target],
+                        recommendation=(
+                            f"Either add a step '{target}' or correct the "
+                            f"{key} of task '{task_id}'"
+                        ),
+                    ))
+                elif target == task_id:
+                    issues.append(DependencyIssue(
+                        issue_type="self_routing",
+                        severity="error",
+                        message=f"Task '{task_id}' routes {key} to itself",
+                        involved_tasks=[task_id],
+                        recommendation=f"Route {key} to a different step",
+                    ))
+
+        # A step id that is also a failure policy makes `on_failure` ambiguous:
+        # routing to it cannot be told apart from selecting the policy. Refuse
+        # rather than silently pick one reading.
+        for policy_name in sorted(FAILURE_POLICIES & set(task_ids)):
+            issues.append(DependencyIssue(
+                issue_type="ambiguous_task_id",
+                severity="error",
+                message=(
+                    f"Task id '{policy_name}' collides with the on_failure "
+                    f"policy of the same name, so routing to it could not be "
+                    f"distinguished from selecting the policy"
+                ),
+                involved_tasks=[policy_name],
+                recommendation=f"Rename the step '{policy_name}'",
+            ))
+
         return issues
 
     def _validate_circular_dependencies(