diff --git a/AGENTS.md b/AGENTS.md index 1b0b407..8ae8c81 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -23,7 +23,10 @@ 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/scripts.py` | Built-in subprocess tasks `taskmaestro.python` (`PythonScript`) and `taskmaestro.command` (`Command`), `ScriptOutput`; JSON exchange via `TASKMAESTRO_INPUT`/`TASKMAESTRO_OUTPUT` (see `docs/script-tasks.md`) | | `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 +46,9 @@ 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()` +- **Script tasks**: `inputs` is `dict[str, Any]`, wired with a keyed `collect`; subprocesses only (no in-process mode). Relative paths resolve against the process CWD. Failures raise `ScriptError` (a `TaskExecutionError`) +- **Validation order**: unique names → references → acyclic (DFS) → task maps → guards (incl. switch label coverage) → type chain (incl. `first_of`) → result task detection ## Testing Conventions diff --git a/README.md b/README.md index 99f2521..8726fdd 100644 --- a/README.md +++ b/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 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: @@ -586,6 +703,58 @@ workflow: The same root/result type inference and single-unconfigured-root requirement apply as for `Workflow.as_task()`. +## Script and Command Tasks + +Two built-in tasks run user code without a Task class. Both run a subprocess and return +`ScriptOutput` (`result`, `returncode`, `stdout`, `stderr`): + +| Task id | Runs | Main settings | +|---|---|---| +| `taskmaestro.python` | `