feat: Phase 1 of typed on_error routing (#227)

README.md

@@ -370,6 +370,7 @@ See the [`examples/`](./examples/) directory for complete workflows:
 | [parallel-research.yaml](./examples/parallel-research.yaml) | Static parallel execution |
 | [design-review.yaml](./examples/design-review.yaml) | Human gate with loop pattern |
 | [script-step.yaml](./examples/script-step.yaml) | Script step with exit_code routing |
+| [error-routing.yaml](./examples/error-routing.yaml) | Typed script failures with deterministic error routes |
 | [set-step.yaml](./examples/set-step.yaml) | Set step deriving named values + boolean-routed branching |
 | [wait-step.yaml](./examples/wait-step.yaml) | Wait step + script for a polling loop-back pattern |
 | [wait-smoke.yaml](./examples/wait-smoke.yaml) | Minimal wait-only smoke test (no provider required) |

docs/workflow-syntax.md

@@ -146,6 +146,8 @@ agents:
     routes:                         # Optional: Routing logic
       - to: next_agent              # Agent name or $end
         when: "{{ condition }}"     # Optional: Route condition
+      - to: recovery_agent          # Script steps only in Phase 1
+        on_error: external.git.drift
 ```
 
 ### Retry Policy
@@ -441,7 +443,74 @@ routes:
   - to: $end
 ```
 
-**Restrictions** — script steps cannot have `prompt`, `model`, `provider`, `tools`, `system_prompt`, `options`, or `validator`. Script steps also cannot be used inside `parallel` groups or `for_each` groups.
+**Typed script failures** — scripts may intentionally publish a language-neutral
+error envelope through the path in `CONDUCTOR_ERROR_OUT`:
+
+```json
+{
+  "kind": "external.git.drift",
+  "message": "remote changed while preparing the update",
+  "details": {"branch": "main"}
+}
+```
+
+`CONDUCTOR_ERROR_OUT` is a file path, not JSON stored in an environment variable.
+Conductor creates a private temporary directory for each script process, injects
+the path only into that subprocess, reads the file after the process exits, and
+then removes the directory. The script may use any language capable of writing
+UTF-8 JSON; no Conductor helper library is required or shipped.
+
+The file is intentionally out-of-band because script stdout already carries the
+step's output contract. File presence means the script intended to raise a typed
+failure, regardless of its exit code. A missing file means no typed failure:
+ordinary nonzero exits, `exit_code` conditions, stdout parsing, and existing
+routes retain their previous behavior.
+
+When a typed envelope is present, error routing takes precedence over validating
+the script's success `output:` schema, so handlers can inspect partial stdout and
+stderr that do not satisfy the success contract.
+
+Malformed JSON, invalid UTF-8, an invalid envelope, or a read failure becomes the
+engine-owned `internal.script_error_transport` envelope. This fail-safe behavior
+prevents an intended typed failure from silently becoming success.
+Scripts cannot publish kinds under the engine-owned `internal.`, `provider.`,
+`subworkflow.`, or `retry.` namespaces; attempts are treated as invalid
+transport envelopes.
+
+```yaml
+agents:
+  - name: fetch
+    type: script
+    command: python
+    args: ["scripts/fetch.py"]
+    raises:                         # Optional documentation/load-time validation
+      - external.git.drift
+    routes:
+      - to: continue_pipeline       # Success bucket
+      - to: recover_drift           # Exact error-kind match
+        on_error: external.git.drift
+      - to: diagnose_unknown        # Catch-all; must be the last error route
+        on_error: true
+```
+
+`raises:` is optional documentation and validation metadata. When present,
+specific `on_error` kinds must be declared (engine-owned kinds such as
+`internal.script_error_transport` are exempt). It does not opt a script into
+error synthesis, does not rewrite undeclared runtime kinds, and does not limit
+`on_error: true`; the catch-all receives the original runtime kind.
+
+Error handlers can inspect both the script's partial output and its envelope:
+
+```yaml
+prompt: |
+  {{ fetch.error.kind }}: {{ fetch.error.message }}
+  stderr: {{ fetch.output.stderr }}
+```
+
+In `context.mode: explicit`, declare `fetch.error` (or a nested field such as
+`fetch.error.kind`) in the handler's `input:` list, just as for `fetch.output`.
+
+**Restrictions** — script steps cannot have `prompt`, `model`, `provider`, `tools`, `system_prompt`, `options`, `validator`, or `retry`. Script steps also cannot be used inside `parallel` groups or `for_each` groups. `CONDUCTOR_ERROR_OUT` is engine-owned and overrides a value with the same name in `env:`.
 
 **Environment variable note** — values in `env` are passed as-is to the subprocess (they are not rendered as Jinja2 templates). Use `${VAR}` syntax in the workflow YAML loader if you need environment variable substitution in env values.
 
@@ -988,6 +1057,59 @@ Structure:
 
 Routes define workflow control flow. Routes are evaluated in order, and the first matching route is taken.
 
+Routes are separated into success and typed-error buckets. A route without
+`on_error` is considered only after successful execution. A route with
+`on_error` is considered only when a script publishes a typed envelope. Within
+the selected bucket, declaration order and `when:` behavior are unchanged.
+
+`on_error` accepts an exact kind, a list of exact kinds, or `true` as a catch-all:
+
+```yaml
+routes:
+  - to: continue_pipeline
+  - to: recover
+    on_error:
+      - external.git.drift
+      - external.git.fetch_failed
+    when: "{{ output.exit_code == 0 and error.details.branch == 'main' }}"
+  - to: diagnose
+    on_error: true
+```
+
+For error-route conditions, both `output` and `error` are in scope. A script
+with error routes must also define at least one success route, and a catch-all
+error route must be last among error routes. Phase 1 supports `on_error` and
+`raises` only on top-level `type: script` steps; provider agents, workflow
+steps, terminate steps, parallel groups, and for-each groups are rejected at
+validation rather than accepted as handlers that never run.
+
+Jinja2 conditions use `error.kind`, `error.message`, and `error.details`.
+Arithmetic-style conditions use the router's flattened names such as
+`error_kind` and `error_message`.
+
+### Error precedence and checkpoints
+
+- **Retry before routing:** existing provider-agent `retry:` behavior completes
+  inside the provider before the workflow engine observes a failure. Provider
+  failures are not routable in Phase 1. Script steps cannot configure `retry`,
+  so typed script failures are never retried implicitly.
+- **Explicit terminate:** `type: terminate, status: failed` remains a terminal
+  control-flow signal. It is not converted to an error envelope, does not run
+  `on_error` routes, and does not create a failure checkpoint.
+- **Sub-workflow terminate:** a child's failed terminate remains
+  `SubworkflowTerminatedError` at the parent seam. `type: workflow` cannot use
+  `on_error` in Phase 1, so existing parent hook and checkpoint behavior is
+  unchanged.
+- **Routed failure:** the failing script's output and envelope are committed to
+  context before its handler runs. The routed failure is handled control flow,
+  so it creates no failure checkpoint. If the handler later fails, the normal
+  failure checkpoint points at the handler and includes the earlier envelope.
+- **Unhandled failure:** if no error route matches, Conductor raises
+  `UnhandledNodeError` before committing the failing step's output/error or
+  execution count. The normal failure checkpoint points at the script so
+  `conductor resume` re-runs it. Any periodic boundary checkpoint taken before
+  the script remains valid.
+
 ### Basic Route
 
 ```yaml
@@ -1236,6 +1358,9 @@ How it works:
   the run reaches a terminal, non-resumable outcome** (clean completion or an
   explicit `status: failed` terminate). On an unexpected failure they are kept
   alongside the on-failure checkpoint.
+- A typed script failure that is successfully routed is not a workflow failure,
+  so it does not create an on-failure checkpoint. An unhandled typed failure
+  follows the normal failure path and checkpoints the failing script for replay.
 - If a periodic save itself fails (e.g. the disk fills), the run is not
   interrupted; the failure is surfaced via a `checkpoint_save_failed` event and
   a console warning so you know recovery may be unavailable.

examples/README.md

@@ -243,6 +243,22 @@ Demonstrates:
 conductor run examples/script-step.yaml
 ```
 
+### error-routing.yaml
+
+Typed script failures using the language-neutral `CONDUCTOR_ERROR_OUT` file
+contract. Demonstrates:
+
+- Exact-kind and catch-all `on_error` routes
+- Optional documentation-only `raises:`
+- Handler access to both `step.error` and partial `step.output`
+- Backward-compatible success behavior without helper libraries
+
+```bash
+conductor run examples/error-routing.yaml --input failure=none
+conductor run examples/error-routing.yaml --input failure=drift
+conductor run examples/error-routing.yaml --input failure=unexpected
+```
+
 ### script-stdin.yaml
 
 Hand a structured payload to a script step via **stdin** instead of

examples/error-routing.yaml

@@ -0,0 +1,82 @@
+# Typed Script Error Routing
+#
+# Usage:
+#   conductor run examples/error-routing.yaml --input failure=none
+#   conductor run examples/error-routing.yaml --input failure=drift
+#   conductor run examples/error-routing.yaml --input failure=unexpected
+
+workflow:
+  name: typed-script-error-routing
+  description: Typed script failures with exact and catch-all error routes
+  version: "1.0.0"
+  entry_point: check_remote
+
+  input:
+    failure:
+      type: string
+      required: false
+      default: none
+
+agents:
+  - name: check_remote
+    type: script
+    command: python
+    args:
+      - -c
+      - |
+        import json
+        import os
+        import sys
+
+        failure = sys.argv[1]
+        if failure != "none":
+            kind = (
+                "external.git.drift"
+                if failure == "drift"
+                else "external.git.unexpected"
+            )
+            envelope = {
+                "kind": kind,
+                "message": f"simulated {failure} failure",
+                "details": {"branch": "main"},
+            }
+            with open(os.environ["CONDUCTOR_ERROR_OUT"], "w", encoding="utf-8") as stream:
+                json.dump(envelope, stream)
+
+        print(json.dumps({"checked": True, "failure": failure}))
+      - "{{ workflow.input.failure }}"
+    raises:
+      - external.git.drift
+    routes:
+      - to: success
+      - to: recover_drift
+        on_error: external.git.drift
+      - to: recover_unexpected
+        on_error: true
+
+  - name: success
+    type: set
+    value: "remote is current"
+    output_type: string
+    routes:
+      - to: $end
+
+  - name: recover_drift
+    type: set
+    value: "recovering {{ check_remote.error.kind }} on {{ check_remote.error.details.branch }}"
+    output_type: string
+    routes:
+      - to: $end
+
+  - name: recover_unexpected
+    type: set
+    value: "caught undeclared kind {{ check_remote.error.kind }}"
+    output_type: string
+    routes:
+      - to: $end
+
+output:
+  result: >-
+    {% if success is defined %}{{ success.output }}
+    {% elif recover_drift is defined %}{{ recover_drift.output }}
+    {% else %}{{ recover_unexpected.output }}{% endif %}

src/conductor/config/schema.py

@@ -20,6 +20,7 @@
 )
 
 from conductor.duration import parse_duration
+from conductor.error_kinds import is_reserved_error_kind, validate_error_kind
 from conductor.file_string import FileString
 from conductor.providers.context_tier import ContextTier
 from conductor.providers.reasoning import ReasoningEffort
@@ -123,6 +124,9 @@ class RouteDef(BaseModel):
     output: dict[str, str] | None = None
     """Optional output transformation (template expressions)."""
 
+    on_error: bool | str | list[str] | None = None
+    """Select typed failures for this route; omitted routes handle success."""
+
     @field_validator("to")
     @classmethod
     def validate_target(cls, v: str) -> str:
@@ -131,6 +135,24 @@ def validate_target(cls, v: str) -> str:
             raise ValueError("Route target cannot be empty")
         return v
 
+    @field_validator("on_error", mode="before")
+    @classmethod
+    def validate_on_error(cls, value: Any) -> bool | str | list[str] | None:
+        """Validate exact, list, and catch-all typed failure selectors."""
+        if value is None:
+            return None
+        if isinstance(value, bool):
+            if not value:
+                raise ValueError("on_error: false is invalid; omit on_error for success routes")
+            return True
+        if isinstance(value, str):
+            return validate_error_kind(value)
+        if isinstance(value, list):
+            if not value:
+                raise ValueError("on_error kind list cannot be empty")
+            return [validate_error_kind(kind) for kind in value]
+        raise ValueError("on_error must be true, an error kind, or a list of error kinds")
+
 
 class ParallelGroup(BaseModel):
     """Definition for a parallel agent execution group."""
@@ -750,6 +772,9 @@ class AgentDef(BaseModel):
     routes: list[RouteDef] = Field(default_factory=list)
     """Routing rules evaluated in order after execution."""
 
+    raises: list[str] | None = None
+    """Optional documentation and load-time validation for script error kinds."""
+
     options: list[GateOption] | None = None
     """Options for human_gate type agents."""
 
@@ -1118,6 +1143,25 @@ def preserve_system_prompt_file_str(
             return value
         return handler(value)
 
+    @field_validator("raises")
+    @classmethod
+    def validate_raises(cls, value: list[str] | None) -> list[str] | None:
+        """Validate optional author-owned error-kind declarations."""
+        if value is None:
+            return None
+        if not value:
+            raise ValueError("raises cannot be empty; omit the field instead")
+
+        validated: list[str] = []
+        for kind in value:
+            validate_error_kind(kind)
+            if is_reserved_error_kind(kind):
+                raise ValueError(f"raises kind '{kind}' uses an engine-owned namespace")
+            if kind in validated:
+                raise ValueError(f"raises kind '{kind}' is declared more than once")
+            validated.append(kind)
+        return validated
+
     @model_validator(mode="after")
     def validate_agent_type(self) -> AgentDef:
         """Ensure agent has required fields for its type."""