Improve nested workflow usability and documentation

.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
@@ -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

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:
 

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")
 
 
 # ---------------------------------------------------------------------------

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

taskmaestro/workflow_task.py

@@ -9,26 +9,31 @@
 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.
         name: Optional name override; defaults to workflow.name.
         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

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)

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