From dbf45ff319129e051a55654d04ed7c541915f55a Mon Sep 17 00:00:00 2001 From: Kristian Bendiksen Date: Tue, 18 Aug 2026 15:37:37 +0200 Subject: [PATCH 1/4] Improve nested workflow usability and documentation --- README.md | 60 ++++++++++++++++++++++++++- examples/image_processing/pipeline.py | 5 +-- taskmaestro/workflow.py | 16 +++++++ taskmaestro/workflow_task.py | 13 ++++-- tests/test_workflow_task.py | 31 ++++++++++++++ 5 files changed, 117 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index e439a48..c0186e0 100644 --- a/README.md +++ b/README.md @@ -160,6 +160,43 @@ job = Job(workflow=workflow, config=EmptyConfig(), job_configuration=job_config) result = Runner().run(job) ``` +## 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: @@ -306,6 +343,26 @@ result = run_workflow_from_yaml("workflow.yaml", "input.yaml") YAML supports named task instances (`name:`), per-task input config (keyed by task name in the input file), fan-in dicts, and output field routing via `[task, field]` lists. +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: @@ -354,12 +411,13 @@ WorkflowRunnerError (base) ## Examples -Two full example pipelines are included in the `examples/` directory: +Three 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 | Run an example: diff --git a/examples/image_processing/pipeline.py b/examples/image_processing/pipeline.py index cc53ab7..6c999d4 100644 --- a/examples/image_processing/pipeline.py +++ b/examples/image_processing/pipeline.py @@ -39,7 +39,6 @@ Runner, Task, Workflow, - workflow_task, ) from taskmaestro.hooks import LoggingHook, TimingHook @@ -266,8 +265,8 @@ def run(self, input: ImageAnalysis, ctx: ExecutionContext) -> ReportOutput: .build() ) -# Wrap the inner workflow as a single opaque task -AnalyzeImage = workflow_task(inner_workflow, name="analyze_image") +# Wrap the inner workflow as a single typed task +AnalyzeImage = inner_workflow.as_task(name="analyze_image") # --------------------------------------------------------------------------- diff --git a/taskmaestro/workflow.py b/taskmaestro/workflow.py index fc34a38..4b9761e 100644 --- a/taskmaestro/workflow.py +++ b/taskmaestro/workflow.py @@ -343,6 +343,22 @@ def _validate_result_task(self) -> None: f"({sinks}); specify result_task explicitly" ) + def as_task( + self, + *, + name: str | None = None, + job_configuration: JobConfiguration | None = None, + ) -> type[Task[Any, Any]]: + """Wrap this workflow as a task for composition in another workflow. + + The task input is inferred from the sole unconfigured root task and its + output from this workflow's result task. If every root is configured, + pass ``job_configuration`` and the generated task accepts ``EmptyConfig``. + """ + from taskmaestro.workflow_task import workflow_task + + return workflow_task(self, name=name, job_configuration=job_configuration) + def to_mermaid(self, *, job_configuration: JobConfiguration | None = None) -> str: """Return a Mermaid diagram string for this workflow.""" from taskmaestro.visualization import to_mermaid diff --git a/taskmaestro/workflow_task.py b/taskmaestro/workflow_task.py index 42ab44b..6a94e6e 100644 --- a/taskmaestro/workflow_task.py +++ b/taskmaestro/workflow_task.py @@ -9,18 +9,21 @@ from taskmaestro.job import EmptyConfig, Job, JobConfiguration, JobStatus from taskmaestro.runner import Runner from taskmaestro.task import Task, get_input_type, get_output_type +from taskmaestro.workflow import Workflow def workflow_task( - workflow: Any, + workflow: Workflow, *, name: str | None = None, job_configuration: JobConfiguration | None = None, ) -> type[Task[Any, Any]]: """Create a Task subclass that wraps an entire workflow as a single task. - The generated task's input type is derived from the inner workflow's root task, - and its output type from the inner workflow's result task. + The generated task's input type is derived from the inner workflow's sole + unconfigured root task, and its output type from the inner workflow's result + task. Prefer ``workflow.as_task()`` in application code; this function remains + available when a factory-style API is more convenient. Args: workflow: The inner Workflow to wrap. @@ -28,7 +31,9 @@ def workflow_task( job_configuration: Optional JobConfiguration for inner tasks with config_fields. Returns: - A new Task subclass that runs the inner workflow. + A new Task subclass that runs the inner workflow with the outer task's + ``ExecutionContext``. The inner workflow is an opaque lifecycle boundary: + hooks on the outer runner observe the generated task, not its inner tasks. Raises: WorkflowDefinitionError: If the inner workflow does not have exactly one diff --git a/tests/test_workflow_task.py b/tests/test_workflow_task.py index 08cc5cf..a61d1e5 100644 --- a/tests/test_workflow_task.py +++ b/tests/test_workflow_task.py @@ -201,6 +201,37 @@ def test_execution(self) -> None: assert result.result == 12 +# ============================================================ +# TestWorkflowAsTask +# ============================================================ + + +class TestWorkflowAsTask: + def test_infers_boundary_types(self) -> None: + inner_wf = Workflow("inner", tasks=[InnerAdd, InnerDouble]) + + WrappedTask = inner_wf.as_task(name="nested") + + assert WrappedTask.name == "nested" + assert get_input_type(WrappedTask) is InnerInput + assert get_output_type(WrappedTask) is InnerOutput + + def test_runs_inside_outer_workflow(self) -> None: + inner_wf = Workflow("inner", tasks=[InnerAdd, InnerDouble]) + Nested = inner_wf.as_task() + outer_wf = ( + Workflow.builder("outer") + .add_task(Nested) + .add_task(DownstreamTask, depends_on=Nested) + .build() + ) + + result = Runner().run(Job(outer_wf, InnerInput(value=4))) + + assert result.status == JobStatus.COMPLETED + assert result.result == StringResult(text="10") + + # ============================================================ # TestWorkflowTaskInOuterWorkflow # ============================================================ From c5ee6bc182c1b69dffab869bd5737c0279e609cb Mon Sep 17 00:00:00 2001 From: Kristian Bendiksen Date: Tue, 18 Aug 2026 15:49:41 +0200 Subject: [PATCH 2/4] Run checks for python versions. --- .github/workflows/ci.yml | 3 --- 1 file changed, 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f5d8384..cf765c7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,15 +25,12 @@ jobs: run: pip install -e ".[dev]" - name: Ruff check - if: matrix.python-version == '3.12' run: ruff check taskmaestro/ tests/ - name: Ruff format check - if: matrix.python-version == '3.12' run: ruff format --check taskmaestro/ tests/ - name: Mypy type check - if: matrix.python-version == '3.12' run: mypy taskmaestro - name: Run tests with coverage From 39dc4d4260259513750a1fc8681592b4cb8f7816 Mon Sep 17 00:00:00 2001 From: Kristian Bendiksen Date: Tue, 18 Aug 2026 15:50:37 +0200 Subject: [PATCH 3/4] Add python v3.14 to test matrix. --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cf765c7..0b40c22 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -11,7 +11,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.12", "3.13"] + python-version: ["3.12", "3.13", "3.14"] steps: - uses: actions/checkout@v4 From 923c5f901093d1cb18b0c7fa493322ca4477a446 Mon Sep 17 00:00:00 2001 From: Kristian Bendiksen Date: Sat, 12 Sep 2026 07:34:43 +0200 Subject: [PATCH 4/4] Test discovery error paths --- tests/test_discovery.py | 42 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/tests/test_discovery.py b/tests/test_discovery.py index 9f3fe2b..71804d7 100644 --- a/tests/test_discovery.py +++ b/tests/test_discovery.py @@ -10,6 +10,7 @@ from pydantic import BaseModel from taskmaestro import ( + ConfigLoadError, ExecutionContext, PluginLoadError, Task, @@ -104,3 +105,44 @@ def test_rejects_duplicate_names(monkeypatch: pytest.MonkeyPatch) -> None: with pytest.raises(PluginLoadError, match="Multiple entry points"): registered_tasks() + + +def test_reports_entry_point_load_failure(monkeypatch: pytest.MonkeyPatch) -> None: + entry = _entry_point("broken", "tests.test_discovery:missing", "taskmaestro.tasks") + monkeypatch.setattr("taskmaestro.discovery.entry_points", lambda *, group: [entry]) + + with pytest.raises(PluginLoadError, match="Cannot load task entry point 'broken'") as exc_info: + registered_tasks() + + assert isinstance(exc_info.value.__cause__, AttributeError) + + +def test_rejects_wrong_workflow_plugin_type(monkeypatch: pytest.MonkeyPatch) -> None: + entry = _entry_point("invalid", "tests.test_discovery:ExampleTask", "taskmaestro.workflows") + monkeypatch.setattr("taskmaestro.discovery.entry_points", lambda *, group: [entry]) + + with pytest.raises(PluginLoadError, match="must resolve to a Workflow"): + registered_workflows() + + +def test_reports_missing_plugins(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("taskmaestro.discovery.entry_points", lambda *, group: []) + + with pytest.raises(PluginLoadError, match="No task entry point named 'missing'"): + get_registered_task("missing") + with pytest.raises(PluginLoadError, match="No workflow entry point named 'missing'"): + get_registered_workflow("missing") + + +def test_yaml_wraps_invalid_task_plugin_error( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + entry = _entry_point("invalid", "tests.test_discovery:Input", "taskmaestro.tasks") + monkeypatch.setattr("taskmaestro.discovery.entry_points", lambda *, group: [entry]) + workflow_path = tmp_path / "workflow.yaml" + workflow_path.write_text("workflow:\n name: invalid\n tasks:\n - task: invalid\n") + input_path = tmp_path / "input.yaml" + input_path.write_text("{}\n") + + with pytest.raises(ConfigLoadError, match="must resolve to a Task subclass"): + load_workflow_from_yaml(workflow_path, input_path)