Return a typed PipelineResult with an execution trace (#332)

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,87 @@ 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`
+(`core/pipeline_result.py`). It is a `Mapping`, so `result["step_id"]` returns
+the step's raw value exactly as before; the trace arrives as attributes
+alongside.
+
+| | |
+|-|-|
+| `status` / `success` | whether the run as a whole succeeded |
+| `outputs` | declared `outputs:`, resolved |
+| `steps` | `StepResult` per step |
+| `execution_order` / `execution_levels` | dependency order, and what could run together |
+| `started_at` / `completed_at` / `duration` | run timing |
+| `failed_steps` / `skipped_steps` / `retried_steps` | the trace, by category |
+
+Each `StepResult` carries its canonical action, status, success, value,
+structured error (`error` and `error_type`), the tool or model and provider
+that ran it, start/end/duration, retry count and dependencies.
+
+**`status` and `success` are not the same question.** `status` records whether
+the task finished; `success` whether it worked. A tool that returns
+`{"success": False}` without raising *finishes* — reading only the status
+reported a failing pipeline as successful, and that is why the CLI's exit code
+consults `success`.
+
+Declared outputs do not change the shape of the return value. They used to:
+a pipeline with `outputs:` returned `{"steps": …, "outputs": …}` and one
+without returned `{step_id: …}`, so a caller could not index a result without
+first checking which it had been handed, and a step named `outputs` collided
+with the second.
+
+`to_dict()` is the stable serialisation the CLI emits — no `default=str`
+coercion, so it round-trips. `normalized()` drops the execution id and
+wall-clock times, which are the only fields that legitimately differ between
+two runs. **The CLI and the Python API must produce equal normalised
+documents**, compared whole rather than by selected nested values.
+
 ## Actions and template resolution
 
 A step names either a tool and an operation on it (`tool: filesystem` with

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/cli.py

@@ -311,7 +311,12 @@ async def _execute():
         click.echo(f"{name}: {exc}", err=True)
         sys.exit(EXIT_VALIDATION_ERROR if is_validation else EXIT_EXECUTION_ERROR)
 
-    rendered = json.dumps(results, indent=2, default=str)
+    # Serialise the typed result, not whatever the executor happened to
+    # return. `default=str` silently stringified anything unserialisable, so
+    # the JSON was neither stable nor round-trippable and CLI/API equivalence
+    # could not be asserted at all.
+    payload = results.to_dict() if hasattr(results, "to_dict") else results
+    rendered = json.dumps(payload, indent=2, default=str)
     if output:
         Path(output).write_text(rendered)
         click.echo(f"Results written to {output}")
@@ -340,6 +345,13 @@ def _failed_steps(results) -> list:
     Steps report themselves as ``{"success": bool, "error": ...}``. Anything
     that does not look like a step result is ignored rather than guessed at.
     """
+    # A PipelineResult knows which steps failed from their recorded status,
+    # which catches a failure whose value is not a dict at all. The duck-typed
+    # fallback stays for anything still returning a plain mapping.
+    typed = getattr(results, "failed_steps", None)
+    if typed is not None:
+        return sorted(step.id for step in typed)
+
     if not isinstance(results, dict):
         return []
     failed = []

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: