Conditional branches

AGENTS.md

@@ -23,7 +23,9 @@ mypy taskmaestro               # type check (strict mode)
 | `taskmaestro/context.py` | `ExecutionContext` with correlation ID, logger, scratch dir, service registry |
 | `taskmaestro/task.py` | `Task[I, O]` ABC, type introspection (`get_input_type`, `get_output_type`) |
 | `taskmaestro/workflow.py` | `Workflow` (linear + DAG), `WorkflowBuilder`, validation (cycles, types, fan-in) |
-| `taskmaestro/dependencies.py`, `taskmaestro/mapping.py` | Task handles, output references, `collect()`, mapped-task configuration |
+| `taskmaestro/dependencies.py`, `taskmaestro/mapping.py` | Task handles (incl. guard sugar `on`/`otherwise`/`if_true`/`if_false`), output references, `collect()`, `first_of()`, mapped-task configuration |
+| `taskmaestro/conditions.py` | `SwitchTask[I, L]`/`ConditionTask[I]` (label in a `Branch` output), `Condition` (unresolved `when=`), `Guard` (resolved), `when()`, `ContextFlag` (condition on a context value; `Runner.run` checks configured keys are registered before running) |
+| `taskmaestro/condition_tasks.py` | Ready-made conditions registered as `condition.*` task entry points: `Compare`, `IsTrue`, `PathExists` |
 | `taskmaestro/workflow_task.py` | Nested workflows wrapped as tasks |
 | `taskmaestro/yaml_config.py` | YAML parsing, task imports, workflow and input validation; `load_registered_workflow` pairs an entry-point workflow with an input file |
 | `taskmaestro/cli.py`, `taskmaestro/discovery.py` | CLI (`run`, `validate`, `graph`, `tasks list/describe`, `workflow list/describe`; workflow arguments accept a YAML path or a registered workflow name), plugin entry-point discovery (`load_registry` collects broken entry points instead of raising) |
@@ -43,7 +45,8 @@ mypy taskmaestro               # type check (strict mode)
 - **Timeouts**: `signal.setitimer`/`SIGALRM` (Unix only, main thread); gracefully warns if unavailable. All armed timeouts go through the process-wide `_AlarmScheduler`, which always arms the nearest expiry, so nested runs (`workflow_task`) cannot cancel or extend an enclosing run's timeouts
 - **Hook error swallowing**: `_emit()` wraps each hook call in try/except, reports via `warnings.warn(..., HookError, source=exc)` — message includes `repr(exc)`; `HookError` subclasses `UserWarning` so it can be filtered or escalated
 - **Inner-workflow failures**: `workflow_task` raises `WorkflowTaskError` (a `TaskExecutionError`) carrying `inner_job` and chaining the original exception via `__cause__`; `Job.exception` keeps the raw exception alongside `Job.error`
-- **Validation order**: unique names → acyclic (DFS) → type chain → result task detection
+- **Conditional branches**: a `when=` guard is a control edge (ordering + predicate, no data); guard sources count as upstreams for topo order, cycles and sinks. A task that is not run gets `TaskStatus.SKIPPED` + `skip_reason` and `on_task_skip(job, task_name, reason)` (no `task_start`). Skips propagate through plain deps; `collect()` drops skipped members; `first_of` takes the first member that ran; a skipped result task fails the job with `ResultTaskSkippedError`. Every switch label needs an `on()` or `otherwise()`
+- **Validation order**: unique names → references → acyclic (DFS) → task maps → guards (incl. switch label coverage) → type chain (incl. `first_of`) → result task detection
 
 ## Testing Conventions
 

README.md

@@ -423,6 +423,123 @@ with `MappedOutput(root={})`. Per-item records are available in
 built-in logging, timing, and persistence hooks observe individual items.
 Concurrent mapped execution is intentionally deferred.
 
+## Conditional Branches
+
+A workflow can choose between branches at run time. A *switch task* returns a
+branch label; downstream tasks are guarded with `when=` and run only when their
+guard matches. Tasks whose guard does not match are **skipped**, not failed.
+
+```python
+from typing import Literal
+
+from taskmaestro import ConditionTask, SwitchTask, Workflow, first_of
+
+
+class ChooseShipping(SwitchTask[Parcel, Literal["express", "freight", "standard"]]):
+    def decide(self, input: Parcel, ctx: ExecutionContext) -> Literal["express", "freight", "standard"]:
+        return "freight" if input.weight_kg > 100 else "standard"
+
+
+builder = Workflow.builder("shipping")
+parcel = builder.task(LoadParcel)
+method = builder.task(ChooseShipping, depends_on=parcel)
+express = builder.task(ExpressShipping, depends_on=parcel, when=method.on("express"))
+freight = builder.task(FreightShipping, depends_on=parcel, when=method.on("freight"))
+standard = builder.task(StandardShipping, depends_on=parcel, when=method.otherwise())
+builder.task(PrintLabel, parcel=parcel, shipment=first_of(express, freight, standard))
+```
+
+- `SwitchTask[I, L]`: implement `decide()` and return a label of type `L`, a
+  `Literal[...]` or an `Enum`. The task output is a `Branch` model with a
+  `branch` field.
+- `ConditionTask[I]` is a switch with the labels `True`/`False`; guard with
+  `handle.if_true()` and `handle.if_false()`.
+- `handle.on(*labels)` runs a task for any of the labels. `handle.otherwise()`
+  runs it for every label that no `on()` guard claims.
+- Any output field can be a guard too: `when(handle.field("kind"), equals=...)`,
+  `in_=[...]` or `truthy=True|False` (truthy/falsy needs a `bool` or optional
+  field; a non-bool value counts as truthy when it is not `None`, so `0` or `""`
+  still count as set). Guard values are validated strictly against the field
+  type: `"no"` or `1` is not accepted for a `bool` field, while the string value
+  of an enum member is. Shorthands: `handle.field("kind").equals(...)`, `.is_in(...)`,
+  `.truthy()`, `.falsy()`.
+- `first_of(a, b, ...)` joins branches: the input gets the output of the first
+  listed member that ran.
+- `ContextFlag` is a ready-made condition on the execution context: it is
+  `True` when the context value registered under its configured `key` is
+  truthy. `Runner.run` raises `ContextValueMissingError` (a
+  `WorkflowDefinitionError`) before any task runs if
+  a configured key is not registered. The `taskmaestro` CLI does not register
+  context values, so it reports such workflows as a configuration error.
+
+  ```python
+  dry = builder.task(ContextFlag, name="dry_run", config_fields=["key"])
+  builder.task(Deploy, depends_on=data, when=dry.if_false())
+  # job configuration: {"dry_run": {"key": "dry_run"}}
+  ctx.register("dry_run", True)
+  ```
+
+- Ready-made conditions, registered as task plugins so they show up in
+  `taskmaestro tasks list` and workflow editors
+  (`taskmaestro.condition_tasks`):
+
+  | Task id                 | Class        | `True` when                                                    |
+  |-------------------------|--------------|----------------------------------------------------------------|
+  | `condition.compare`     | `Compare`    | `value <op> threshold`; `op` is one of `<`, `<=`, `==`, `!=`, `>=` (default), `>` |
+  | `condition.is_true`     | `IsTrue`     | `value` is truthy                                              |
+  | `condition.path_exists` | `PathExists` | `path` exists; `kind` may require a `file` or `directory`     |
+
+  ```yaml
+  - task: condition.compare
+    name: is_large
+    depends_on:
+      value: [pipeline.LoadOrder, amount]
+    config_fields: [op, threshold]
+  - task: pipeline.ManualReview
+    depends_on: pipeline.LoadOrder
+    when: {switch: is_large, case: true}
+  ```
+
+Validation happens at build time. Labels must belong to the switch's label
+type, so typos are caught. Every label needs a guarded task or an
+`otherwise()`. An `otherwise()` that can never match is an error. A switch with
+no guarded tasks is an ordinary task.
+
+At run time:
+
+- A task is skipped when its guard does not match, when the guard's source was
+  skipped, when any plain dependency was skipped, or when every `first_of`
+  member was skipped. Skips propagate downstream.
+- `collect()` leaves out skipped members.
+- Skipped tasks get a `TaskResult` with `status=TaskStatus.SKIPPED` and a
+  `skip_reason`, and are listed in `job.skipped_tasks`. Hooks receive
+  `on_task_skip(job, task_name, reason)`; `on_task_start` is not called for
+  them.
+- If the result task is skipped, the job fails with `ResultTaskSkippedError`.
+
+In YAML:
+
+```yaml
+- task: pipeline.ExpressShipping
+  depends_on: pipeline.LoadParcel
+  when: {switch: pipeline.ChooseShipping, case: express}       # case may be a list
+- task: pipeline.StandardShipping
+  depends_on: pipeline.LoadParcel
+  when: {switch: pipeline.ChooseShipping, otherwise: true}
+- task: pipeline.Audit
+  depends_on: pipeline.LoadParcel
+  when: {ref: [pipeline.LoadParcel, priority], truthy: true}   # or equals: / in: [...]
+- task: pipeline.PrintLabel
+  depends_on:
+    parcel: pipeline.LoadParcel
+    shipment:
+      first_of: [pipeline.ExpressShipping, pipeline.StandardShipping]
+```
+
+In Mermaid diagrams, switch tasks are diamonds and guards are dashed edges
+labeled with their case (`express`, `otherwise`, `true`, `.kind = a`, ...).
+`first_of` joins are drawn as rounded nodes.
+
 ## ObjectModel
 
 `ObjectModel[T]` wraps arbitrary (non-Pydantic) objects so they can flow through workflows. Use it as a type alias for simple wrappers, or subclass it to add extra fields:
@@ -616,7 +733,7 @@ graph TD
     build_report -->|AnalysisReport| _end_
 ```
 
-Edges are labeled with data types. Fan-in edges show field names, and field routing edges show `.field: Type`. When a `JobConfiguration` is provided, configured tasks get dashed edges from a `JobConfiguration` node.
+Edges are labeled with data types. Fan-in edges show field names, and field routing edges show `.field: Type`. When a `JobConfiguration` is provided, configured tasks get dashed edges from a `JobConfiguration` node. Switch tasks are drawn as diamonds with dashed, labeled guard edges (see [Conditional Branches](#conditional-branches)).
 
 ## Error Handling
 
@@ -630,6 +747,7 @@ WorkflowRunnerError (base)
 │   └── WorkflowLookupError       # Unknown or broken registered workflow name
 └── TaskExecutionError            # Runtime task failure
     ├── MappedTaskExecutionError  # One or more mapped items failed
+    ├── ResultTaskSkippedError    # The result task was skipped by a condition
     ├── TaskOutputTypeError       # Output type mismatch
     └── TaskTimeoutError          # Task exceeded timeout
 ```
@@ -738,14 +856,16 @@ prints indented JSON.
 
 ## Examples
 
-Four full example pipelines are included in the `examples/` directory:
+Six full example pipelines are included in the `examples/` directory:
 
 | Example | Features |
 |---|---|
 | `examples/text_analysis/` | DAG with fan-out/fan-in, output field routing, inline `Inputs`/`Outputs` classes, YAML config, Mermaid visualization |
 | `examples/resinsight/` | `ObjectModel[T]` for gRPC objects, `JobConfiguration` with per-task config, named task instances, `config_fields`, YAML config |
 | `examples/image_processing/` | Nested workflows through `Workflow.as_task()` and YAML `workflow:`, typed boundaries, expanded Mermaid subgraph |
 | `examples/release_pipeline/` | Keyed `collect()` dependencies, mapped tasks, mapped output routing, per-task YAML config |
+| `examples/conditional_if/` | If/else with `ConditionTask`, `if_true()`/`if_false()` guards, `first_of` join, YAML `when:` |
+| `examples/conditional_switch/` | Multi-way `SwitchTask` with `on(...)` and `otherwise()`, `first_of` join, YAML `when:` |
 
 Run an example: