diff --git a/docs/actions.md b/docs/actions.md
index c4f6069..20326f1 100644
--- a/docs/actions.md
+++ b/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.
diff --git a/docs/adr/0001-product-contract.md b/docs/adr/0001-product-contract.md
index 85a03e1..4095e27 100644
--- a/docs/adr/0001-product-contract.md
+++ b/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`
diff --git a/scripts/generate_action_docs.py b/scripts/generate_action_docs.py
index 8027522..4363e6d 100644
--- a/scripts/generate_action_docs.py
+++ b/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)
 
 
diff --git a/src/orchestrator/compiler/schema_validator.py b/src/orchestrator/compiler/schema_validator.py
index 1e3a8af..527ee36 100644
--- a/src/orchestrator/compiler/schema_validator.py
+++ b/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": [
diff --git a/src/orchestrator/compiler/yaml_compiler.py b/src/orchestrator/compiler/yaml_compiler.py
index 345fcd4..203a486 100644
--- a/src/orchestrator/compiler/yaml_compiler.py
+++ b/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:
diff --git a/src/orchestrator/core/routing.py b/src/orchestrator/core/routing.py
new file mode 100644
index 0000000..1b6dd8f
--- /dev/null
+++ b/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
diff --git a/src/orchestrator/orchestrator.py b/src/orchestrator/orchestrator.py
index 96816b6..03c6926 100644
--- a/src/orchestrator/orchestrator.py
+++ b/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
diff --git a/src/orchestrator/validation/dependency_validator.py b/src/orchestrator/validation/dependency_validator.py
index e6f5581..288858a 100644
--- a/src/orchestrator/validation/dependency_validator.py
+++ b/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(
diff --git a/tests/test_control_flow_routing.py b/tests/test_control_flow_routing.py
new file mode 100644
index 0000000..6a5e06b
--- /dev/null
+++ b/tests/test_control_flow_routing.py
@@ -0,0 +1,233 @@
+"""Control-flow routing: `on_false`, `on_success`, `on_failure` (#333).
+
+A step may say where execution goes next. Routing jumps forward, marking the
+steps in between as skipped -- the same machinery `goto` already used, rather
+than a second one beside it.
+
+The awkward part is `on_failure`, which already meant a *failure policy*
+(`fail` / `continue` / `skip` / `retry`) and now also names a step to jump to.
+Both are supported, disambiguated by value, and the ambiguous case -- a step
+whose id *is* a policy word -- is refused at compile time rather than guessed.
+"""
+
+import asyncio
+import os
+from pathlib import Path
+
+import pytest
+
+from orchestrator.compiler.yaml_compiler import YAMLCompiler
+from orchestrator.core.exceptions import YAMLCompilerError
+from orchestrator.core.routing import FAILURE_POLICIES, is_failure_policy, routing_targets
+from orchestrator.models.model_registry import ModelRegistry
+from tests.test_infrastructure import MockTestModel, create_test_orchestrator
+
+pytestmark = [pytest.mark.contract, pytest.mark.e2e]
+
+
+def _compiler():
+    registry = ModelRegistry()
+    registry.register_model(MockTestModel())
+    return YAMLCompiler(model_registry=registry)
+
+
+def _run(yaml_content, cwd):
+    previous = Path.cwd()
+    os.chdir(cwd)
+    try:
+        return asyncio.run(
+            create_test_orchestrator().execute_yaml(yaml_content=yaml_content, context={})
+        )
+    finally:
+        os.chdir(previous)
+
+
+def _write_step(step_id, filename, *, depends_on=None, **extra):
+    lines = [
+        f"  - id: {step_id}",
+        "    tool: filesystem",
+        "    action: write",
+        "    parameters:",
+        f'      path: "out/{filename}"',
+        f'      content: "{step_id}-ran"',
+    ]
+    if depends_on:
+        lines.append(f"    dependencies: [{depends_on}]")
+    for key, value in extra.items():
+        lines.append(f"    {key}: {value}")
+    return "\n".join(lines)
+
+
+def _conditional_pipeline(condition):
+    """check -> (middle) -> recover, where `check` routes on_false to recover."""
+    return f"""
+id: routing_pipeline
+name: Routing Pipeline
+steps:
+  - id: check
+    action: evaluate_condition
+    condition: "{condition}"
+    parameters:
+      condition: "1 == 1"
+    on_false: recover
+{_write_step("middle", "middle.txt", depends_on="check")}
+{_write_step("recover", "recover.txt", depends_on="middle")}
+"""
+
+
+# ---------------------------------------------------------------------------
+# on_false
+# ---------------------------------------------------------------------------
+
+def test_on_false_jumps_and_skips_the_steps_in_between(tmp_path):
+    result = _run(_conditional_pipeline("false"), tmp_path)
+
+    assert result.steps["check"].status == "skipped"
+    assert result.steps["middle"].status == "skipped", (
+        "the step routed over must be skipped, not run"
+    )
+    assert result.steps["recover"].success is True
+
+    assert not (tmp_path / "out" / "middle.txt").exists(), (
+        "a skipped step must not have had its side effect"
+    )
+    assert (tmp_path / "out" / "recover.txt").read_text() == "recover-ran"
+
+
+def test_a_routed_pipeline_still_reports_success(tmp_path):
+    """Skipping is not failing. A pipeline that routed around a step worked."""
+    result = _run(_conditional_pipeline("false"), tmp_path)
+
+    assert result.success is True
+    assert result.status == "completed"
+    assert [s.id for s in result.failed_steps] == []
+    assert sorted(s.id for s in result.skipped_steps) == ["check", "middle"]
+
+
+def test_on_false_does_not_fire_when_the_condition_holds(tmp_path):
+    """The branch not taken must actually not be taken."""
+    result = _run(_conditional_pipeline("true"), tmp_path)
+
+    assert result.steps["check"].status == "completed"
+    assert result.steps["middle"].success is True, "the normal path must run"
+    assert (tmp_path / "out" / "middle.txt").exists()
+
+
+# ---------------------------------------------------------------------------
+# on_success
+# ---------------------------------------------------------------------------
+
+def test_on_success_jumps_past_the_intervening_step(tmp_path):
+    pipeline = f"""
+id: on_success_pipeline
+name: On Success Pipeline
+steps:
+{_write_step("first", "first.txt", on_success="last")}
+{_write_step("skipped_middle", "middle.txt", depends_on="first")}
+{_write_step("last", "last.txt", depends_on="skipped_middle")}
+"""
+    result = _run(pipeline, tmp_path)
+
+    assert result.steps["first"].success is True
+    assert result.steps["skipped_middle"].status == "skipped"
+    assert result.steps["last"].success is True
+    assert not (tmp_path / "out" / "middle.txt").exists()
+
+
+# ---------------------------------------------------------------------------
+# on_failure: policy or step id
+# ---------------------------------------------------------------------------
+
+def test_on_failure_naming_a_step_routes_instead_of_aborting(tmp_path):
+    pipeline = f"""
+id: on_failure_routing
+name: On Failure Routing
+steps:
+  - id: boom
+    tool: filesystem
+    action: read
+    on_failure: handler
+    parameters:
+      path: "/nonexistent/definitely/not/here.txt"
+{_write_step("never", "never.txt", depends_on="boom")}
+{_write_step("handler", "handler.txt", depends_on="never")}
+"""
+    result = _run(pipeline, tmp_path)
+
+    assert result.steps["boom"].success is False
+    assert result.steps["handler"].success is True, (
+        "routing on failure must reach the handler"
+    )
+    assert not (tmp_path / "out" / "never.txt").exists()
+
+
+@pytest.mark.parametrize("policy", sorted(FAILURE_POLICIES))
+def test_reserved_policy_words_keep_their_policy_meaning(policy):
+    """`on_failure: continue` is a policy, not a jump to a step called continue."""
+    assert is_failure_policy(policy)
+    assert routing_targets({"id": "s", "on_failure": policy}) == {}
+
+
+def test_a_non_policy_on_failure_is_a_routing_target():
+    assert routing_targets({"id": "s", "on_failure": "handler"}) == {
+        "on_failure": "handler"
+    }
+
+
+# ---------------------------------------------------------------------------
+# compile-time validation
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("key", ["on_false", "on_success", "on_failure"])
+def test_routing_to_a_step_that_does_not_exist_fails_to_compile(key):
+    """Exit 2 naming the bad target, not a run that jumps into nothing."""
+    pipeline = f"""
+id: bad_routing
+name: Bad Routing
+steps:
+{_write_step("only", "only.txt", **{key: "nowhere"})}
+"""
+    with pytest.raises(YAMLCompilerError) as excinfo:
+        asyncio.run(_compiler().compile(pipeline, {}, resolve_ambiguities=False))
+
+    assert "nowhere" in str(excinfo.value)
+
+
+def test_routing_to_itself_fails_to_compile():
+    pipeline = f"""
+id: self_routing
+name: Self Routing
+steps:
+{_write_step("only", "only.txt", on_success="only")}
+"""
+    with pytest.raises(YAMLCompilerError) as excinfo:
+        asyncio.run(_compiler().compile(pipeline, {}, resolve_ambiguities=False))
+
+    assert "itself" in str(excinfo.value)
+
+
+def test_a_step_named_after_a_failure_policy_is_refused():
+    """The one genuinely ambiguous case, refused rather than guessed at.
+
+    With a step called `retry`, `on_failure: retry` could mean either the
+    policy or a jump to that step, and no rule can tell them apart.
+    """
+    pipeline = f"""
+id: colliding_ids
+name: Colliding Ids
+steps:
+{_write_step("first", "first.txt")}
+{_write_step("retry", "retry.txt", depends_on="first")}
+"""
+    with pytest.raises(YAMLCompilerError) as excinfo:
+        asyncio.run(_compiler().compile(pipeline, {}, resolve_ambiguities=False))
+
+    message = str(excinfo.value)
+    assert "retry" in message and "collides" in message
+
+
+def test_valid_routing_compiles_and_is_carried_on_the_task():
+    pipeline = _conditional_pipeline("false")
+    compiled = asyncio.run(_compiler().compile(pipeline, {}, resolve_ambiguities=False))
+
+    assert compiled.tasks["check"].metadata["on_false"] == "recover"