Metadata-Version: 2.5
Name: taskmaestro
Version: 0.3.0
Summary: A Python 3.12+ library for defining and executing typed DAG task workflows
Project-URL: Homepage, https://github.com/OPM/taskmaestro
Project-URL: Repository, https://github.com/OPM/taskmaestro
Project-URL: Issues, https://github.com/OPM/taskmaestro/issues
Author-email: Ceetron Solutions <info@ceetronsolutions.com>
License-Expression: GPL-3.0-or-later
License-File: LICENSE
Keywords: dag,orchestration,pipeline,pydantic,task,workflow
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: pydantic<3.0,>=2.0
Requires-Dist: pyyaml<7.0,>=6.0
Provides-Extra: dev
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.16; extra == 'dev'
Requires-Dist: types-pyyaml>=6.0; extra == 'dev'
Description-Content-Type: text/markdown

# Taskmaestro

<p align="center">
  <img src="taskmaestro.png" alt="Taskmaestro" width="300">
</p>

A Python 3.12+ library for defining and executing typed DAG task workflows with Pydantic models, lifecycle hooks, and fail-fast semantics.

For an agent-friendly, runnable YAML example with validation and troubleshooting steps, see the [agent quickstart](docs/agent-quickstart.md).

## Installation

```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
```

## Quick Start

### Linear Pipeline

```python
from pydantic import BaseModel
from taskmaestro import Task, Workflow, Job, Runner, ExecutionContext


class NumberInput(BaseModel):
    value: int


class NumberOutput(BaseModel):
    value: int


class AddOne(Task[NumberInput, NumberOutput]):
    def run(self, input: NumberInput, ctx: ExecutionContext) -> NumberOutput:
        return NumberOutput(value=input.value + 1)


class Double(Task[NumberOutput, NumberOutput]):
    def run(self, input: NumberOutput, ctx: ExecutionContext) -> NumberOutput:
        return NumberOutput(value=input.value * 2)


workflow = Workflow(name="math", tasks=[AddOne, Double])
job = Job(workflow=workflow, config=NumberInput(value=5))
result = Runner().run(job)

print(result.status)  # "completed"
print(result.result.value)  # 12
```

### DAG Workflow (Fan-In)

```python
from pydantic import BaseModel
from taskmaestro import Task, Workflow, Job, Runner, ExecutionContext


class Input(BaseModel):
    value: int


class Output(BaseModel):
    value: int


class MergedInput(BaseModel):
    a: Output
    b: Output


class MergedOutput(BaseModel):
    total: int


class BranchA(Task[Input, Output]):
    def run(self, input: Input, ctx: ExecutionContext) -> Output:
        return Output(value=input.value + 1)


class BranchB(Task[Input, Output]):
    def run(self, input: Input, ctx: ExecutionContext) -> Output:
        return Output(value=input.value * 2)


class Merge(Task[MergedInput, MergedOutput]):
    def run(self, input: MergedInput, ctx: ExecutionContext) -> MergedOutput:
        return MergedOutput(total=input.a.value + input.b.value)


workflow = (
    Workflow.builder(name="fan_in")
    .add_task(BranchA)
    .add_task(BranchB)
    .add_task(Merge, depends_on={"a": BranchA, "b": BranchB})
    .build()
)

job = Job(workflow=workflow, config=Input(value=5))
result = Runner().run(job)

print(result.status)  # "completed"
print(result.result.total)  # 16 (6 + 10)
```

## Core Concepts

You define **Tasks** (typed units of work), compose them into a **Workflow** (linear chain or DAG), bind input data via a **Job**, and hand it to a **Runner** for execution. Type safety is enforced at build time — input/output models are validated across the entire graph. Fail-fast semantics stop execution on the first error, and **Hooks** provide cross-cutting lifecycle observations without coupling to task logic.

| Concept | Description |
|---|---|
| **Task** | Subclass `Task[I, O]` with Pydantic models for input and output, then implement `run(input, ctx)`. Each task can declare an optional `timeout_seconds`. For tasks with multiple named outputs, use inline `Inputs`/`Outputs` classes inside the task body. |
| **Workflow** | Build a linear pipeline with `Workflow(tasks=[...])` or a DAG with `Workflow.builder()`. Prefer `builder.task()` and task handles for unambiguous dependencies; the fluent `add_task()` API remains supported. The builder accepts `collect()` for gathering outputs into collection fields and `mapped_over=TaskMap(...)` for sequential expansion over configured mappings. Use `config_fields` to declare which input fields come from `JobConfiguration`. Workflows are validated at build time for cycles, type compatibility, and input completeness. |
| **Job** | Binds a Workflow to a typed config (the root task's input). Tracks `status` (`pending` → `running` → `completed`/`failed`), the final `result`, any `error`, and per-task `task_results`. Optionally accepts a `JobConfiguration` for per-task static config values. A job can only be run once. |
| **Runner** | Executes tasks in topological order, stopping on the first failure (fail-fast). Supports per-task and per-job timeouts via `signal.alarm` (Unix only). Dispatches lifecycle events to registered hooks. |
| **ExecutionContext** | Passed to every `run()` call. Provides a `logger`, an auto-generated `correlation_id` (UUID), a `scratch_dir` (temporary directory), and a service registry (`register()`/`resolve()`) for injecting shared resources like DB connections. |
| **Hooks** | Subclass `BaseHook` and override methods like `on_job_start`, `on_task_complete`, etc. Hook errors are swallowed and reported via `warnings.warn()`, so they never crash the job. Built-ins: `LoggingHook`, `TimingHook`, `ResultPersistenceHook`. |
| **ObjectModel** | Generic `ObjectModel[T]` base model for wrapping arbitrary (non-Pydantic) objects. Enables `arbitrary_types_allowed` so fields can hold native library objects like database connections or API clients. |

For common cases, run a workflow directly without constructing `Job` and `Runner`:

```python
result = workflow.run(
    Input(value=5),
    task_config={"configured_task": {"option": "value"}},
    hooks=[LoggingHook()],
)
```

The explicit `Job` and `Runner` API remains available for advanced lifecycle control.

## Task Handles

`builder.task()` adds a task and returns a handle to that specific instance. Handles avoid ambiguous class and string references, especially when the same task class is registered more than once:

```python
builder = Workflow.builder(name="parallel_wells")
model = builder.task(LoadModel)
well_1 = builder.task(LoadWellPath, name="well_1", depends_on=model)
well_2 = builder.task(LoadWellPath, name="well_2", depends_on=model)
builder.task(Process, name="proc_1", depends_on=well_1)
builder.task(Process, name="proc_2", depends_on=well_2)
workflow = builder.build()
```

Use `handle.field("field_name")` to route one output field. Named input dependencies can be passed directly to `task()`, while keyword arguments to `collect()` provide a concise keyed collection:

```python
merged = builder.task(
    MergeResults,
    primary=producer.field("result"),
    checks=collect(tests=tests, lint=lint, types=types),
)
builder.set_result_task(merged)
```

Handles are accepted anywhere dependency references are accepted. A handle from a different builder is rejected. Use the `depends_on` dictionary form when an input field conflicts with a reserved builder argument such as `name` or `config_fields`. The existing fluent `add_task()` API remains fully supported for backward compatibility.

## Named Task Instances

The same Task class can appear multiple times in a workflow with different names. With the fluent API, use the `name=` parameter in `add_task()`:

```python
workflow = (
    Workflow.builder(name="parallel_wells")
    .add_task(LoadModel)
    .add_task(LoadWellPath, name="well_1", depends_on=LoadModel)
    .add_task(LoadWellPath, name="well_2", depends_on=LoadModel)
    .add_task(Process, name="proc_1", depends_on="well_1")
    .add_task(Process, name="proc_2", depends_on="well_2")
    .build()
)
```

Named instances can be referenced as string dependencies (`depends_on="well_1"`) or in fan-in dicts.

## Per-Task Configuration

`JobConfiguration` provides static config values for individual tasks, merged with upstream outputs at runtime. Declare which fields come from config with `config_fields`:

```python
from taskmaestro import EmptyConfig, Job, JobConfiguration, Workflow

workflow = (
    Workflow.builder(name="configured")
    .add_task(LoadModel, config_fields=["path"])  # root: all input from config
    .add_task(Transform, depends_on=LoadModel, config_fields=["scale_factor"])  # mixed
    .build()
)

job_config = JobConfiguration(
    {
        "load_model": {"path": "/data/model.egrid"},
        "transform": {"scale_factor": 2.5},
    }
)

job = Job(workflow=workflow, config=EmptyConfig(), job_configuration=job_config)
result = Runner().run(job)
```

Every declared config field, on root, dependent and mapped tasks alike, must have a value in the `JobConfiguration`; otherwise `Job(...)` raises `WorkflowDefinitionError`.

## Nested Workflows

A workflow can be wrapped as a typed task and used inside a larger workflow. Its input
type is inferred from its root task and its output type from its result task, so normal
workflow type validation still applies at both boundaries:

```python
inner = Workflow(name="normalize", tasks=[CleanText, NormalizeText])
Normalize = inner.as_task(name="normalize_text")

outer = (
    Workflow.builder("document_pipeline")
    .add_task(LoadDocument)
    .add_task(Normalize, depends_on=LoadDocument)
    .add_task(IndexDocument, depends_on=Normalize)
    .build()
)
```

`workflow_task(inner, name="normalize_text")` is the equivalent factory-style API.
The inner workflow must have exactly one root that receives input from the outer
workflow. Roots supplied entirely by `JobConfiguration` are excluded; if every root is
configured, pass that configuration to `as_task()` and the wrapper accepts
`EmptyConfig`:

```python
ConfiguredPipeline = inner.as_task(
    name="configured_pipeline",
    job_configuration=inner_config,
)
```

The inner tasks share the outer `ExecutionContext`, including services, scratch directory,
and correlation ID. The wrapper is an opaque lifecycle boundary: outer runner hooks and
`Job.task_results` see one wrapper task, while an inner failure is reported with the inner
workflow and failed task names. Mermaid visualization expands wrappers as subgraphs.

## Output Field Routing

Route a specific field from an upstream task's output (rather than the whole output) using `(Task, "field")` tuples:

```python
class ExtractKeywords(Task):
    class Inputs(BaseModel):
        content: TextContent

    class Outputs(BaseModel):
        keywords: KeywordsOutput
        num_words_removed: int

    def run(self, input: Inputs, ctx: ExecutionContext) -> Outputs: ...


workflow = (
    Workflow.builder(name="analysis")
    .add_task(ExtractKeywords, depends_on=PrepareText)
    .add_task(
        BuildReport,
        depends_on={
            "keywords": (ExtractKeywords, "keywords"),  # routes .keywords field
            "num_words_removed": (
                ExtractKeywords,
                "num_words_removed",
            ),  # routes .num_words_removed
            "stats": ComputeWordStats,  # whole output
        },
    )
    .build()
)
```

## Collecting Multiple Outputs

Use `collect()` when several task outputs should populate one `list[T]` or
`dict[str, T]` field. Positional members preserve declaration order:

```python
from taskmaestro import collect

class GridInput(BaseModel):
    surfaces: list[Surface]

workflow = (
    Workflow.builder("create_grid")
    .add_task(LoadSurface, name="top")
    .add_task(GenerateSurface, name="middle")
    .add_task(LoadSurface, name="base")
    .add_task(
        CreateGrid,
        depends_on={"surfaces": collect("top", "middle", "base")},
    )
    .build()
)
```

Use a mapping to preserve aliases in a `dict[str, T]`, and use `(task, "field")`
to collect a specific output field:

```python
depends_on={
    "surfaces": collect({
        "top": ("top_loader", "surface"),
        "base": ("base_loader", "surface"),
    })
}
```

The equivalent YAML forms are:

```yaml
depends_on:
  surfaces:
    collect:
      - top
      - [middle, generated_surface]
      - base
```

```yaml
depends_on:
  surfaces:
    collect:
      top: [top_loader, surface]
      base: [base_loader, surface]
```

Every member is checked against the field's element type when the workflow is
built. Subtypes are accepted. `collect()` and `collect({})` explicitly create
empty list and dictionary inputs, respectively.

## Mapped Tasks

A mapped task invokes one task declaration for every entry in a configured
mapping. Mapped items execute sequentially in mapping declaration order.
Each item gets a fresh task instance and child `ExecutionContext`.

```python
builder = Workflow.builder("create_grid")
connection = builder.task(ConnectToResInsight)
surfaces = builder.map_task(
    LoadRegularSurface,
    name="load_surfaces",
    depends_on={"resinsight": connection},
    config_fields=["unit"],
    over="surfaces",
    key_as="surface_name",
    value_as="path",
    error_mode="fail_fast",
)
builder.task(CreateGrid, depends_on={"surfaces": surfaces})
workflow = builder.build()
```

The mapped task's input model contains the injected key and value fields, not
the source mapping:

```python
class LoadSurfaceInput(BaseModel):
    resinsight: RipsInstance
    unit: str
    surface_name: str  # key_as
    path: str          # value_as
```

Configure the source through `JobConfiguration`:

```python
job_configuration = JobConfiguration({
    "load_surfaces": {
        "unit": "meters",
        "surfaces": {
            "top": "/data/top.irap",
            "base": "/data/base.irap",
        },
    },
})
```

The logical output is a `MappedOutput[O]` Pydantic root model containing an
insertion-ordered `dict[str, O]`, where `O` is the task's declared output type.
When a mapped task is connected to a named `dict[str, O]` input field, its
`root` value is unwrapped automatically:

```python
class CreateGridInput(BaseModel):
    surfaces: dict[str, RegularSurface]
```

The equivalent YAML task declaration is:

```yaml
- task: resinsight.load_regular_surface
  name: load_surfaces
  map:
    over: surfaces
    key_as: surface_name
    value_as: path
    error_mode: fail_fast
  depends_on:
    resinsight: resinsight.connect
  config_fields: [unit]
```

Input YAML:

```yaml
load_surfaces:
  unit: meters
  surfaces:
    top: /data/top.irap
    base: /data/base.irap
```

`fail_fast` stops at the first failed item. `collect_all` attempts every item
and reports an aggregate `MappedTaskExecutionError`. An empty mapping succeeds
with `MappedOutput(root={})`. Per-item records are available in
`job.mapped_item_results`, and
built-in logging, timing, and persistence hooks observe individual items.
Concurrent mapped execution is intentionally deferred.

## 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:

```python
from taskmaestro import ObjectModel

# Type alias — no extra fields needed
GridCase = ObjectModel[rips.EclipseCase]
WellPath = ObjectModel[rips.WellPath]


# Subclass — adds fields alongside the wrapped object
class AddPerforationInput(ObjectModel[rips.WellPath]):
    start_md: float
    end_md: float


# Access the wrapped object via .value
grid = GridCase(value=eclipse_case)
print(grid.value.name)
```

## Plugin discovery

Installed packages can publish tasks and workflows using standard Python entry points:

```toml
[project.entry-points."taskmaestro.tasks"]
"acme.prepare" = "acme_tasks.prepare:Prepare"

[project.entry-points."taskmaestro.workflows"]
"acme.analysis" = "acme_tasks.workflows:analysis_workflow"
```

A task entry point must resolve to a `Task` subclass and a workflow entry point must
resolve to a `Workflow` instance. Prefix names with the provider name to avoid clashes.
Consumers can discover plugins without scanning package directories:

```python
from taskmaestro import registered_tasks, registered_workflows

tasks = registered_tasks()  # dict[str, type[Task]]
workflows = registered_workflows()  # dict[str, Workflow]
```

Use `registered_task_names()` and `registered_workflow_names()` to inspect identifiers
without importing plugin modules, or `get_registered_task(name)` and
`get_registered_workflow(name)` to load one plugin. Duplicate names and invalid plugin
types raise `PluginLoadError`.

Agents can inspect installed task plugins from the CLI without loading every plugin:

```bash
taskmaestro tasks list --json
taskmaestro tasks describe acme.prepare --json
```

`list --json` emits `{"tasks": ["acme.prepare", ...]}` in sorted order; an empty list
means no task plugins are installed. `describe --json` loads only the named plugin
and emits its registered `identifier`, task `name`, `timeout_seconds`, and Pydantic
`input_schema` / `output_schema` (JSON Schema objects). These commands inspect
**installed entry points**, not task classes local to a workflow YAML file. Without
`--json`, `list` prints one identifier per line and `describe` prints indented JSON.
For runtime-only Python objects (such as `ObjectModel[rips.EclipseCase]`), schema
fields include `"not": {}`, `"x-taskmaestro-opaque": true`, and
`"x-taskmaestro-python-type"`. They cannot be supplied as JSON; wire them from
upstream tasks or a Python context instead. Other unsupported schema constructs,
unknown identifiers, and invalid plugins report an error on stderr and exit with status 2.

## YAML Configuration

Workflows can be defined entirely in YAML instead of Python. A `task:` value may be
either a registered task identifier or a dotted Python class path. The loader resolves
registered identifiers first and validates the full configuration:

```yaml
# workflow.yaml
workflow:
  name: text_analysis
  tasks:
    - task: pipeline.PrepareText
    - task: pipeline.GenerateStopWords
    - task: pipeline.ComputeWordStats
      depends_on:
        content: pipeline.PrepareText
        stop_words: pipeline.GenerateStopWords
    - task: pipeline.ExtractKeywords
      depends_on:
        content: pipeline.PrepareText
        stop_words: pipeline.GenerateStopWords
    - task: pipeline.ScoreReadability
      depends_on: pipeline.PrepareText
    - task: pipeline.BuildReport
      depends_on:
        stats: pipeline.ComputeWordStats
        keywords: [pipeline.ExtractKeywords, keywords]                    # output field routing
        readability: pipeline.ScoreReadability
        num_words_removed: [pipeline.ExtractKeywords, num_words_removed]  # output field routing

runner:
  hooks:
    - hook: taskmaestro.hooks.logging.LoggingHook
    - hook: taskmaestro.hooks.timing.TimingHook

context:
  services:
    title: "Python Overview"
```

```yaml
# input.yaml
prepare_text:
  text: "Python is a high-level programming language..."
  title: "Python Overview"
```

Load and run:

```python
from taskmaestro import load_workflow_from_yaml, run_workflow_from_yaml

# Load for inspection, then run
loaded = load_workflow_from_yaml("workflow.yaml", "input.yaml")
result = loaded.run()

# Or run directly
result = run_workflow_from_yaml("workflow.yaml", "input.yaml")
```

YAML input always uses per-task configuration: every top-level key in `input.yaml` must be a registered task instance name, and its value must be a mapping or `null`. Unknown task names and scalar task values are rejected. Fields are validated against the task's input model and can configure root tasks, downstream tasks, and mapped tasks.

YAML also supports named task instances (`name:`), fan-in dictionaries, and output field routing via `[task, field]` lists. Named instances use their instance name as the input key:

```yaml
load_well_path_1:
  path: first.dev
load_well_path_2:
  path: second.dev
```

When the same task class (or the same inner YAML file) appears more than once under different `name:`s, `depends_on` and `result_task` must use the instance name — referencing the class path is rejected as ambiguous.

Use `workflow:` instead of `task:` to compose another YAML workflow. Paths are resolved
relative to the containing workflow file, and `workflow_input:` optionally supplies the
inner workflow's per-task configuration:

```yaml
workflow:
  name: document_pipeline
  tasks:
    - task: pipeline.LoadDocument
    - workflow: normalize/workflow.yaml
      workflow_input: normalize/input.yaml
      name: normalize_text
      depends_on: pipeline.LoadDocument
    - task: pipeline.IndexDocument
      depends_on: normalize_text
```

The same root/result type inference and single-unconfigured-root requirement apply as for
`Workflow.as_task()`.

## Visualization

Generate Mermaid diagrams of workflow topology:

```python
print(workflow.to_mermaid())
# or with config nodes:
print(workflow.to_mermaid(job_configuration=job_config))
```

Output:

```mermaid
---
title: text_analysis
---
graph TD
    _start_(("start"))
    _end_(("end"))
    prepare_text["prepare_text"]
    generate_stop_words["generate_stop_words"]
    compute_word_stats["compute_word_stats"]
    build_report["build_report"]
    _start_ -->|TextInput| prepare_text
    _start_ -->|TextInput| generate_stop_words
    prepare_text -->|content: TextContent| compute_word_stats
    compute_word_stats -->|WordStatsOutput| build_report
    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.

## Error Handling

```
WorkflowRunnerError (base)
├── WorkflowDefinitionError       # Invalid workflow definition
│   ├── CycleDetectedError        # Dependency cycle
│   └── IncompleteInputError      # Missing fan-in field mappings
├── JobStateError                 # e.g., re-running a completed job
├── ConfigLoadError               # YAML config loading failure
└── TaskExecutionError            # Runtime task failure
    ├── MappedTaskExecutionError  # One or more mapped items failed
    ├── TaskOutputTypeError       # Output type mismatch
    └── TaskTimeoutError          # Task exceeded timeout
```

## Command-Line Interface

Installed packages provide a `taskmaestro` command for YAML workflows:

```bash
taskmaestro validate workflow.yaml --input input.yaml
taskmaestro graph workflow.yaml --input input.yaml
taskmaestro run workflow.yaml --input input.yaml --log-level INFO
taskmaestro workflow describe workflow.yaml --json
```

By default, `run` prints the final output as JSON and reports errors on stderr;
`validate` prints a human-readable confirmation. `graph` prints Mermaid markup.

For automation, use `--json` with `validate` or `run`:

```bash
taskmaestro validate workflow.yaml --input input.yaml --json
taskmaestro run workflow.yaml --input input.yaml --json
```

Successful validation emits `{"status":"valid","workflow":"..."}`. A successful
run emits `{"status":"completed","workflow":"...","result":{...}}`. A failure
emits one JSON object on stdout, for example:

```json
{"status":"failed","workflow":"example","failed_task":"prepare","error":{"code":"task_failed","type":"ValueError","message":"Task failed","task":"prepare","field":null,"issues":[]}}
```

Errors contain `code`, exception `type`, a safe `message`, nullable `task` and
`field`, and `issues` (field paths and error codes, without input values).
Loading failures have status `invalid` and `code: "configuration_error"`; task
failures have status `failed` and `code: "task_failed"`. If a completed result
cannot be encoded as JSON (e.g. a Python-only object), `run --json` returns
`code: "serialization_error"`. Missing configuration fields produce `issues`
with code `missing`; when no structured field metadata is available, `field`
is `null`. Use text mode when you need the original exception message.

Exit codes: `0` success, `1` task or result-serialization failure, `2` workflow
configuration failure. In JSON mode logs and ordinary Python `print()` output
from imports/tasks go to stderr, reserving stdout for the result document.
Application code can still write directly to file descriptor 1; JSON mode is
not a sandbox. Error objects intentionally omit raw exception messages, but
application-generated stderr may contain sensitive data.

Inspect a workflow **before its input file is complete** with
`taskmaestro workflow describe workflow.yaml --json`. The JSON contains the
workflow name and result task plus topologically ordered task instances. Each
instance includes its `name`, `python_type`, `depends_on` references (`task` /
`field`), `config_fields`, `required_input_fields`, optional `map` configuration,
and Pydantic input/output schemas. Collection dependencies include their kind
(`positional` or `keyed`) and member references. `python_type` identifies the
loaded class; it may differ from the plugin entry-point identifier used in YAML.

Pass `--input input.yaml` to check the input against the workflow. The input
may be incomplete (for example a template whose runtime-picker fields a UI
fills in): `provided_config_fields` lists the field names set for each task,
`missing_config_fields` lists declared `config_fields` the input does not set,
and `config_values` maps exactly the provided names to their values; without
`--input`, all three are `null`. Other problems, such as unknown task keys or a
missing or invalid map source, still fail. `validate` and `run` reject missing
configuration fields. Values are what `input.yaml` contains for
the task, **before** Pydantic validation, so opaque markers (for example
`{"__resinsight_ref__": "EclipseCase", "case_id": 0}`) come through unchanged.
Dates and datetimes become ISO 8601 strings, paths become strings, sets become
sorted lists and non-string mapping keys become JSON key strings; binary values,
NaN/infinity and keys that collide after conversion are rejected as invalid.
Fields without a value fall back to `input_schema.properties.<field>.default`.
**The output contains the input values, including any secrets.** This check
does not run tasks or hooks and does not validate every configured value's
runtime type. Loading YAML still imports
Python modules (and nested workflows); **do not inspect untrusted YAML or plugins**
under a privileged account. In JSON mode inspection failures have status
`invalid`, an `error` object, and exit code 2. Without `--json`, inspection
prints indented JSON.

`python -m taskmaestro ...` is equivalent, which is useful when the scripts directory is not on `PATH`.

## Examples

Four 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 |

Run an example:

```bash
python examples/text_analysis/pipeline.py              # Python API
python examples/text_analysis/pipeline.py --yaml       # YAML config
```

## Development

```bash
source .venv/bin/activate
pytest -v                  # run tests
ruff check .               # lint
ruff format .              # format
mypy taskmaestro       # type check (strict)
```
