From 2e32a0034c9a9f9b9d5c52eb3b0f9bb16eab29a0 Mon Sep 17 00:00:00 2001
From: Kristian Bendiksen <kristian.bendiksen@gmail.com>
Date: Wed, 28 Oct 2026 15:12:42 +0100
Subject: [PATCH] Add built-in Python script and command tasks

Add taskmaestro.python and taskmaestro.command entry-point tasks that run
a Python script or an executable/shell command as a subprocess. Upstream
data is passed as JSON via TASKMAESTRO_INPUT (keyed collect into
`inputs`), and scripts may write a JSON result to TASKMAESTRO_OUTPUT.
Failures, timeouts and invalid output raise ScriptError.

Report unknown root-task config fields before uncovered required fields,
so a typo in input.yaml names the misspelled key.

Includes an example under examples/scripts/, a design doc in
docs/script-tasks.md, and README/AGENTS.md updates.

Closes #1
---
 AGENTS.md                       |   2 +
 README.md                       |  57 +++-
 docs/script-tasks.md            | 108 +++++++
 examples/scripts/input.yaml     |   8 +
 examples/scripts/sample.txt     |   3 +
 examples/scripts/save_report.sh |  11 +
 examples/scripts/word_count.py  |  32 +++
 examples/scripts/workflow.yaml  |  25 ++
 pyproject.toml                  |   2 +
 taskmaestro/__init__.py         |  14 +
 taskmaestro/exceptions.py       |  22 ++
 taskmaestro/scripts.py          | 287 +++++++++++++++++++
 taskmaestro/workflow.py         |  15 +-
 tests/test_condition_tasks.py   |   5 +-
 tests/test_scripts.py           | 491 ++++++++++++++++++++++++++++++++
 tests/test_workflow.py          |   6 +
 16 files changed, 1079 insertions(+), 9 deletions(-)
 create mode 100644 docs/script-tasks.md
 create mode 100644 examples/scripts/input.yaml
 create mode 100644 examples/scripts/sample.txt
 create mode 100644 examples/scripts/save_report.sh
 create mode 100644 examples/scripts/word_count.py
 create mode 100644 examples/scripts/workflow.yaml
 create mode 100644 taskmaestro/scripts.py
 create mode 100644 tests/test_scripts.py

diff --git a/AGENTS.md b/AGENTS.md
index 814b773..8ae8c81 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -26,6 +26,7 @@ mypy taskmaestro               # type check (strict mode)
 | `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) |
@@ -46,6 +47,7 @@ mypy taskmaestro               # type check (strict mode)
 - **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`
 - **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 a4d9d7d..1dbf019 100644
--- a/README.md
+++ b/README.md
@@ -705,6 +705,59 @@ 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` | `<current python> <script> <args...>` | `script`, `args` |
+| `taskmaestro.command` | A string through the shell, or a list as an executable with arguments (no shell) | `command` |
+
+Shared settings: `inputs` (upstream data), `env` (extra environment variables), `cwd`,
+`timeout` (seconds; the process is killed) and `check` (default `true`: a nonzero exit
+code fails the task with `ScriptError`, including the last lines of stderr). Absolute paths
+are used as given; relative `script` and `cwd` paths resolve against the current directory
+of the process running the workflow.
+
+Data is exchanged through JSON files in a fresh directory under the scratch directory,
+deleted when the task finishes:
+
+- `TASKMAESTRO_INPUT`: path to a JSON object holding `inputs` (set when `inputs` is not empty).
+- `TASKMAESTRO_OUTPUT`: path where the process may write a JSON value; it becomes `result`.
+
+`inputs` is a `dict[str, Any]`, connected with a keyed `collect` so scripts always receive a
+JSON object. Runtime-only objects (such as `ObjectModel[rips.EclipseCase]`) cannot be sent to a
+process and fail the task; use a regular `Task` for those.
+
+Their settings are ordinary input fields, given per task in `input.yaml`:
+
+```yaml
+# workflow.yaml (tasks)
+- task: taskmaestro.python
+  name: word_count
+- task: taskmaestro.command
+  name: save_report
+  depends_on:
+    inputs:
+      collect: {counts: [word_count, result]}
+```
+
+```yaml
+# input.yaml
+word_count:
+  script: scripts/word_count.py
+  args: [data/sample.txt]
+save_report:
+  command: sh scripts/save_report.sh
+  env: {REPORT_DIR: /tmp/reports}
+```
+
+Downstream tasks can use a single field, e.g. `depends_on: {path: [save_report, result]}`;
+the downstream input model validates the value at run time. See
+[`docs/script-tasks.md`](docs/script-tasks.md) and `examples/scripts/`.
+
 ## Visualization
 
 Generate Mermaid diagrams of workflow topology:
@@ -750,6 +803,7 @@ WorkflowRunnerError (base)
 └── TaskExecutionError            # Runtime task failure
     ├── MappedTaskExecutionError  # One or more mapped items failed
     ├── ResultTaskSkippedError    # The result task was skipped by a condition
+    ├── ScriptError               # A script/command task failed (returncode, stdout, stderr)
     ├── TaskOutputTypeError       # Output type mismatch
     └── TaskTimeoutError          # Task exceeded timeout
 ```
@@ -858,7 +912,7 @@ prints indented JSON.
 
 ## Examples
 
-Six full example pipelines are included in the `examples/` directory:
+Seven full example pipelines are included in the `examples/` directory:
 
 | Example | Features |
 |---|---|
@@ -868,6 +922,7 @@ Six full example pipelines are included in the `examples/` directory:
 | `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:` |
+| `examples/scripts/` | `taskmaestro.python` and `taskmaestro.command` tasks, keyed `collect` into a script (YAML only; run from the repository root) |
 
 Run an example:
 
diff --git a/docs/script-tasks.md b/docs/script-tasks.md
new file mode 100644
index 0000000..868072c
--- /dev/null
+++ b/docs/script-tasks.md
@@ -0,0 +1,108 @@
+# Script and command tasks (issue #1)
+
+## Problem
+Users need small steps in a workflow: post-processing a result, copying files, or
+calling an existing script or executable. Before this change, each such step needed a
+`Task[I, O]` subclass with its own Pydantic input and output models. Issue #1 asks for
+custom tasks without that overhead. The design follows the proposal in the issue
+comments, with the changes listed under "Decisions".
+
+## Agreed design
+
+### Built-in tasks
+Both tasks are registered in the `taskmaestro.tasks` entry-point group. They appear in
+`taskmaestro tasks list`, and `tasks describe` reports their schemas.
+
+| Task id | Class | Runs |
+|---|---|---|
+| `taskmaestro.python` | `PythonScript` | `<sys.executable> <script> <args...>` in a subprocess |
+| `taskmaestro.command` | `Command` | `command` as a string through the shell, or as a list run directly, without a shell |
+
+Input fields:
+
+| Field | Type | Default | Tasks |
+|---|---|---|---|
+| `script` | path | required | python |
+| `args` | list of str | `[]` | python |
+| `command` | str \| list of str | required | command |
+| `inputs` | `dict[str, Any]` | `{}` | both |
+| `env` | mapping str → str | `{}` | both, added to the inherited environment |
+| `cwd` | path | current dir | both |
+| `timeout` | float > 0 (s) | none | both, kills the process |
+| `check` | bool | `true` | both, fails on a nonzero exit code |
+
+Output (`ScriptOutput`): `result: Any` (the JSON from `$TASKMAESTRO_OUTPUT`, or `null`),
+`returncode`, `stdout`, `stderr`.
+
+### Data exchange
+Each run gets a fresh directory under `ctx.scratch_dir` (`script-*`). The directory is
+deleted when the task finishes, also when it fails: `result`, `stdout`, `stderr` and
+`returncode` are returned in `ScriptOutput` or `ScriptError`, and `inputs` may hold
+sensitive data. Processes left running in the background must not rely on these files
+after the task returns.
+
+- `TASKMAESTRO_INPUT` is set only when `inputs` is not empty. It holds the path to
+  `input.json`, which is `inputs` serialized with `pydantic_core.to_json`, so Pydantic
+  models become objects.
+- `TASKMAESTRO_OUTPUT` is always set. The process may write one JSON value there. A
+  missing or empty file gives `result = null`. Invalid JSON fails the task.
+
+### `inputs` is `dict[str, Any]`
+Upstream data is connected with a keyed `collect`:
+`inputs: {collect: {export: [exporter, result]}}`. A `[task, field]` reference whose
+type is `dict[str, ...]` also works. Scripts therefore always receive a JSON object with
+known keys. This type needs no change to the type checker, because keyed collections
+into `dict[str, T]` were already supported.
+
+Alternatives that were rejected:
+- `Any` would accept a whole model, a list or a dict. Collections into `Any` would need a
+  core validation change, and scripts would get an input whose shape varies.
+- `str` would only accept string-typed fields. Most upstream outputs are models.
+
+`result` is `Any`, so it can be connected to any downstream field. The downstream input
+model validates the value at run time.
+
+### Configuration
+The settings are ordinary input fields. They are given per task in `input.yaml`, like
+any other task configuration, and need no new YAML syntax.
+
+The proposal also had a `params:` key on workflow task entries. It would hold fixed
+settings, such as the script path, in `workflow.yaml`. It was left out because it is a
+general YAML feature rather than something scripts need, and it can be proposed
+separately.
+
+### Errors and timeouts
+Failures raise `ScriptError`, a `TaskExecutionError` with `returncode`, `stdout` and
+`stderr`:
+- A missing script is reported before anything runs.
+- Inputs that cannot be converted to JSON fail the task with a message suggesting a
+  regular Task. This applies to runtime-only objects such as `ObjectModel[rips.EclipseCase]`.
+- A process that cannot start (`OSError`) fails the task.
+- A `timeout` kills the process and fails the task.
+- A nonzero exit code with `check: true` fails the task. The message includes the exit
+  code and the last 20 lines of stderr.
+- When a runner or task timeout (`SIGALRM`) fires while a process runs,
+  `subprocess.run` kills and reaps the child before the timeout error propagates.
+
+### Paths
+Absolute paths are used as given. Relative `script` and `cwd` paths resolve against the
+current directory of the process that runs the workflow, not the workflow file's
+directory. `script` is made absolute before the process starts, so `cwd` does not
+affect it. Paths inside `command` or `args` are interpreted by the child process
+in `cwd`.
+
+## Decisions
+- **Function (in-process) mode is postponed.** The proposal had a default `function`
+  mode that calls `main(inputs, options, ctx)` in the workflow's interpreter. That mode
+  could receive live objects. If it is added later, it should be an opt-in `mode`
+  field that defaults to `subprocess`, which keeps existing workflows unchanged.
+- **No `params:` key** (see Configuration).
+- No new dependencies.
+
+## ResInsight integration notes
+- `result` has the empty JSON schema `{}`. ResInsight must accept edges out of a field
+  with that schema.
+- `inputs` is an object schema whose values can be anything. ResInsight connects to it
+  with keyed `collect` edges (`"collect": "dict"`).
+- ResInsight could warn when an opaque (`x-taskmaestro-opaque`) output is collected into
+  `inputs`, since that fails at run time.
diff --git a/examples/scripts/input.yaml b/examples/scripts/input.yaml
new file mode 100644
index 0000000..2902d62
--- /dev/null
+++ b/examples/scripts/input.yaml
@@ -0,0 +1,8 @@
+word_count:
+  script: examples/scripts/word_count.py
+  args: [examples/scripts/sample.txt, "5"]
+count_bytes:
+  command: [wc, -c, examples/scripts/sample.txt]
+save_report:
+  command: sh examples/scripts/save_report.sh
+  env: {REPORT_DIR: /tmp/taskmaestro-script-tasks}
diff --git a/examples/scripts/sample.txt b/examples/scripts/sample.txt
new file mode 100644
index 0000000..b42662d
--- /dev/null
+++ b/examples/scripts/sample.txt
@@ -0,0 +1,3 @@
+Taskmaestro runs typed tasks in a workflow.
+A script task runs a script, and a command task runs a command.
+Tasks pass JSON to scripts and scripts pass JSON back to tasks.
diff --git a/examples/scripts/save_report.sh b/examples/scripts/save_report.sh
new file mode 100644
index 0000000..18a857b
--- /dev/null
+++ b/examples/scripts/save_report.sh
@@ -0,0 +1,11 @@
+#!/bin/sh
+# Run by the taskmaestro.command task. Upstream data is in $TASKMAESTRO_INPUT
+# (JSON); the path of the saved report is returned through $TASKMAESTRO_OUTPUT.
+set -eu
+mkdir -p "$REPORT_DIR"
+report="$REPORT_DIR/word_count.json"
+cp "$TASKMAESTRO_INPUT" "$report"
+# printf does not escape JSON: this output is valid only while REPORT_DIR has no
+# '"' or '\'. Use a Python script (json.dump) for output built from arbitrary values.
+printf '{"report": "%s"}' "$report" > "$TASKMAESTRO_OUTPUT"
+echo "Saved $report"
diff --git a/examples/scripts/word_count.py b/examples/scripts/word_count.py
new file mode 100644
index 0000000..783b342
--- /dev/null
+++ b/examples/scripts/word_count.py
@@ -0,0 +1,32 @@
+"""Count the words in a text file and return the most common ones.
+
+Run by the ``taskmaestro.python`` task: arguments come from ``args`` and the
+result is written as JSON to ``$TASKMAESTRO_OUTPUT``. Taskmaestro is not imported.
+"""
+
+import json
+import os
+import sys
+from collections import Counter
+
+path = sys.argv[1]
+top = int(sys.argv[2]) if len(sys.argv) > 2 else 3
+
+with open(path, encoding="utf-8") as f:
+    text = f.read()
+words = [word.strip(".,;:!?\"'()").lower() for word in text.split()]
+words = [word for word in words if word]
+if not words:
+    print(f"No words in {path}", file=sys.stderr)
+    sys.exit(1)  # Fails the task (check: true).
+
+with open(os.environ["TASKMAESTRO_OUTPUT"], "w", encoding="utf-8") as f:
+    json.dump(
+        {
+            "words": len(words),
+            "lines": len(text.splitlines()),
+            "top_words": [word for word, _count in Counter(words).most_common(top)],
+        },
+        f,
+    )
+print(f"Counted {len(words)} words in {path}")
diff --git a/examples/scripts/workflow.yaml b/examples/scripts/workflow.yaml
new file mode 100644
index 0000000..1bc8967
--- /dev/null
+++ b/examples/scripts/workflow.yaml
@@ -0,0 +1,25 @@
+# Script and command tasks: no Task classes needed. Their settings (script,
+# args, command, env, ...) are ordinary input fields, given in input.yaml.
+# Run from the repository root (relative paths resolve against the current directory):
+#   taskmaestro run examples/scripts/workflow.yaml --input examples/scripts/input.yaml
+
+workflow:
+  name: script_tasks
+  tasks:
+    # Python file run in a subprocess.
+    - task: taskmaestro.python
+      name: word_count
+
+    # Executable with arguments (list form: no shell, no quoting issues).
+    - task: taskmaestro.command
+      name: count_bytes
+
+    # Shell script (string form: runs through the shell). Upstream data
+    # arrives as one JSON object in $TASKMAESTRO_INPUT.
+    - task: taskmaestro.command
+      name: save_report
+      depends_on:
+        inputs:
+          collect:
+            counts: [word_count, result]
+            wc_output: [count_bytes, stdout]
diff --git a/pyproject.toml b/pyproject.toml
index aeffe56..d4f9380 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -35,6 +35,8 @@ taskmaestro = "taskmaestro.cli:main"
 "condition.compare" = "taskmaestro.condition_tasks:Compare"
 "condition.is_true" = "taskmaestro.condition_tasks:IsTrue"
 "condition.path_exists" = "taskmaestro.condition_tasks:PathExists"
+"taskmaestro.command" = "taskmaestro.scripts:Command"
+"taskmaestro.python" = "taskmaestro.scripts:PythonScript"
 
 [project.urls]
 Homepage = "https://github.com/OPM/taskmaestro"
diff --git a/taskmaestro/__init__.py b/taskmaestro/__init__.py
index 5dbecb4..811ad15 100644
--- a/taskmaestro/__init__.py
+++ b/taskmaestro/__init__.py
@@ -33,6 +33,7 @@
     MappedTaskExecutionError,
     PluginLoadError,
     ResultTaskSkippedError,
+    ScriptError,
     TaskExecutionError,
     TaskOutputTypeError,
     TaskTimeoutError,
@@ -45,6 +46,13 @@
 from taskmaestro.mapping import MappedOutput, TaskMap
 from taskmaestro.object_model import ObjectModel
 from taskmaestro.runner import HookError, Runner
+from taskmaestro.scripts import (
+    Command,
+    CommandInput,
+    PythonScript,
+    PythonScriptInput,
+    ScriptOutput,
+)
 from taskmaestro.task import Task
 from taskmaestro.visualization import to_mermaid
 from taskmaestro.workflow import Workflow, WorkflowBuilder
@@ -60,6 +68,8 @@
     "TASK_ENTRY_POINT_GROUP",
     "WORKFLOW_ENTRY_POINT_GROUP",
     "Branch",
+    "Command",
+    "CommandInput",
     "Condition",
     "ConditionTask",
     "ConfigLoadError",
@@ -82,8 +92,12 @@
     "ObjectModel",
     "OutputHandle",
     "PluginLoadError",
+    "PythonScript",
+    "PythonScriptInput",
     "ResultTaskSkippedError",
     "Runner",
+    "ScriptError",
+    "ScriptOutput",
     "SwitchTask",
     "Task",
     "TaskExecutionError",
diff --git a/taskmaestro/exceptions.py b/taskmaestro/exceptions.py
index 9924d71..11b3388 100644
--- a/taskmaestro/exceptions.py
+++ b/taskmaestro/exceptions.py
@@ -86,6 +86,28 @@ def __init__(self, workflow_name: str, inner_job: Any) -> None:
         )
 
 
+class ScriptError(TaskExecutionError):
+    """A script or command run by ``taskmaestro.python``/``taskmaestro.command`` failed.
+
+    ``returncode`` is ``None`` when the process was not started or did not finish
+    (missing script, unserializable inputs, timeout). ``stdout``/``stderr`` hold the
+    captured output, if any.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        returncode: int | None = None,
+        stdout: str = "",
+        stderr: str = "",
+    ) -> None:
+        super().__init__(message)
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
 class ConfigLoadError(WorkflowRunnerError):
     """Raised when YAML config loading fails (parse errors, import failures, validation)."""
 
diff --git a/taskmaestro/scripts.py b/taskmaestro/scripts.py
new file mode 100644
index 0000000..90ccc79
--- /dev/null
+++ b/taskmaestro/scripts.py
@@ -0,0 +1,287 @@
+"""Built-in tasks that run a Python script or a command in a subprocess.
+
+Registered as the ``taskmaestro.python`` and ``taskmaestro.command`` task plugins.
+They make one-off workflow steps possible without writing a Task class.
+
+Data exchange with the process uses JSON files in a fresh directory under the
+run's scratch directory:
+
+- ``TASKMAESTRO_INPUT``: path to a JSON object holding ``inputs`` (set only when
+  ``inputs`` is not empty).
+- ``TASKMAESTRO_OUTPUT``: path where the process may write a JSON value, which
+  becomes ``ScriptOutput.result``.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import signal
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from pydantic import BaseModel, Field
+from pydantic_core import PydanticSerializationError, to_json
+
+from taskmaestro.context import ExecutionContext
+from taskmaestro.exceptions import ScriptError
+from taskmaestro.task import Task
+
+INPUT_ENV = "TASKMAESTRO_INPUT"
+OUTPUT_ENV = "TASKMAESTRO_OUTPUT"
+_STDERR_TAIL_LINES = 20
+# Executable name prefixes accepted as a Python interpreter (python3, python3.14t,
+# pythonw.exe, pypy3, graalpy, ...).
+_PYTHON_NAMES = ("python", "pypy", "graalpy")
+
+
+class ScriptOutput(BaseModel):
+    """Output of :class:`PythonScript` and :class:`Command`."""
+
+    result: Any = Field(default=None, description=f"JSON value written to ${OUTPUT_ENV}, or null")
+    returncode: int = Field(default=0, description="Exit code of the process")
+    stdout: str = Field(default="", description="Captured standard output")
+    stderr: str = Field(default="", description="Captured standard error")
+
+
+class _ProcessInput(BaseModel):
+    """Settings shared by script and command tasks."""
+
+    inputs: dict[str, Any] = Field(
+        default_factory=dict,
+        description=(
+            f"Upstream data, connected with a keyed collect; written as JSON to ${INPUT_ENV}"
+        ),
+    )
+    env: dict[str, str] = Field(
+        default_factory=dict, description="Extra environment variables for the process"
+    )
+    cwd: Path | None = Field(
+        default=None, description="Working directory of the process (default: current directory)"
+    )
+    timeout: float | None = Field(
+        default=None, gt=0, description="Kill the process after this many seconds"
+    )
+    check: bool = Field(default=True, description="Fail the task on a nonzero exit code")
+
+
+class PythonScriptInput(_ProcessInput):
+    """Input of :class:`PythonScript`."""
+
+    script: Path = Field(description="Python file to run")
+    args: list[str] = Field(default_factory=list, description="Command-line arguments")
+
+
+class CommandInput(_ProcessInput):
+    """Input of :class:`Command`."""
+
+    command: str | list[str] = Field(
+        description=(
+            "A string runs through the shell; a list runs the executable directly, without a shell"
+        )
+    )
+
+
+class PythonScript(Task[PythonScriptInput, ScriptOutput]):
+    """Run a Python file with the current interpreter in a subprocess."""
+
+    name = "python"
+
+    def run(self, input: PythonScriptInput, ctx: ExecutionContext) -> ScriptOutput:
+        # Resolve before the process changes directory to ``cwd``.
+        script = input.script.expanduser().absolute()
+        if not script.is_file():
+            raise ScriptError(f"Script not found: {script}")
+        return run_process(
+            [_python_interpreter(), str(script), *input.args],
+            settings=input,
+            ctx=ctx,
+            label=f"Script '{script.name}'",
+        )
+
+
+class Command(Task[CommandInput, ScriptOutput]):
+    """Run a shell command, shell script or executable in a subprocess."""
+
+    name = "command"
+
+    def run(self, input: CommandInput, ctx: ExecutionContext) -> ScriptOutput:
+        if not input.command:
+            raise ScriptError("Command is empty")
+        shown = input.command if isinstance(input.command, str) else " ".join(input.command)
+        return run_process(input.command, settings=input, ctx=ctx, label=f"Command '{shown}'")
+
+
+def run_process(
+    command: str | list[str],
+    *,
+    settings: _ProcessInput,
+    ctx: ExecutionContext,
+    label: str,
+) -> ScriptOutput:
+    """Run ``command`` with the data-exchange protocol and return its output.
+
+    A string runs through the shell. Raises :class:`ScriptError` when the
+    process cannot be run, times out, exits nonzero with ``check`` set, or
+    writes invalid JSON to ``TASKMAESTRO_OUTPUT``. The exchange directory is
+    deleted before returning, also on failure: everything it held is returned in
+    :class:`ScriptOutput` or :class:`ScriptError`, and inputs may be sensitive.
+    """
+    ctx.scratch_dir.mkdir(parents=True, exist_ok=True)
+    exchange_dir = Path(tempfile.mkdtemp(prefix="script-", dir=ctx.scratch_dir))
+    try:
+        return _run_in_exchange_dir(command, exchange_dir, settings=settings, ctx=ctx, label=label)
+    finally:
+        shutil.rmtree(exchange_dir, ignore_errors=True)
+
+
+def _run_in_exchange_dir(
+    command: str | list[str],
+    exchange_dir: Path,
+    *,
+    settings: _ProcessInput,
+    ctx: ExecutionContext,
+    label: str,
+) -> ScriptOutput:
+    output_path = exchange_dir / "output.json"
+    env = {**os.environ, **settings.env, OUTPUT_ENV: str(output_path)}
+    env.pop(INPUT_ENV, None)
+    if settings.inputs:
+        input_path = exchange_dir / "input.json"
+        try:
+            input_path.write_bytes(to_json(settings.inputs))
+        except PydanticSerializationError as exc:
+            raise ScriptError(
+                f"{label}: inputs cannot be converted to JSON ({exc}). Runtime-only "
+                "objects cannot be passed to a subprocess; use a regular Task instead."
+            ) from exc
+        env[INPUT_ENV] = str(input_path)
+
+    ctx.logger.debug("%s: running in %s", label, exchange_dir)
+    completed = _run(command, settings=settings, env=env, label=label)
+
+    if settings.check and completed.returncode != 0:
+        tail = "\n".join(completed.stderr.splitlines()[-_STDERR_TAIL_LINES:])
+        message = f"{label} failed with exit code {completed.returncode}"
+        raise ScriptError(
+            f"{message}: {tail}" if tail else message,
+            returncode=completed.returncode,
+            stdout=completed.stdout,
+            stderr=completed.stderr,
+        )
+
+    return ScriptOutput(
+        result=_read_result(output_path, label, completed),
+        returncode=completed.returncode,
+        stdout=completed.stdout,
+        stderr=completed.stderr,
+    )
+
+
+def _read_result(
+    output_path: Path, label: str, completed: subprocess.CompletedProcess[str]
+) -> Any:
+    """Return the JSON value written to the output file, or ``None`` if none was written."""
+    try:
+        text = output_path.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return None
+    if not text.strip():
+        return None
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError as exc:
+        raise ScriptError(
+            f"{label} wrote invalid JSON to ${OUTPUT_ENV}: {exc}",
+            returncode=completed.returncode,
+            stdout=completed.stdout,
+            stderr=completed.stderr,
+        ) from exc
+
+
+def _run(
+    command: str | list[str],
+    *,
+    settings: _ProcessInput,
+    env: dict[str, str],
+    label: str,
+) -> subprocess.CompletedProcess[str]:
+    """Run ``command`` to completion, killing its whole process tree on timeout or error.
+
+    On POSIX the process leads a new session, so a timeout, a runner timeout
+    (``SIGALRM``) or ``KeyboardInterrupt`` kills every process it started, not
+    just the shell. ``subprocess.run`` would kill only the direct child.
+    """
+    try:
+        process = subprocess.Popen(
+            command,
+            shell=isinstance(command, str),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            env=env,
+            cwd=settings.cwd,
+            start_new_session=sys.platform != "win32",
+        )
+    except OSError as exc:
+        raise ScriptError(f"{label} could not be started: {exc}") from exc
+
+    with process:  # Closes the pipes and reaps the process on exit.
+        try:
+            stdout, stderr = process.communicate(timeout=settings.timeout)
+        except subprocess.TimeoutExpired as exc:
+            _kill_process_tree(process)
+            # ``communicate`` stored the output read so far on the exception
+            # (POSIX). Not calling it again avoids blocking on pipes held open
+            # by a process that escaped the process group.
+            raise ScriptError(
+                f"{label} timed out after {settings.timeout}s",
+                stdout=_text(exc.stdout),
+                stderr=_text(exc.stderr),
+            ) from exc
+        except BaseException:
+            _kill_process_tree(process)
+            raise
+    return subprocess.CompletedProcess(command, process.returncode, stdout, stderr)
+
+
+def _kill_process_tree(process: subprocess.Popen[str]) -> None:
+    """Kill ``process`` and, on POSIX, every process in its process group."""
+    if sys.platform != "win32":
+        try:
+            os.killpg(process.pid, signal.SIGKILL)
+        except ProcessLookupError:  # Group already gone.
+            pass
+        except PermissionError:  # A group member changed user; kill what we can.
+            process.kill()
+    else:
+        process.kill()
+    process.wait()
+
+
+def _python_interpreter() -> str:
+    """Return the interpreter for script tasks, or raise if there is no usable one.
+
+    When Python is embedded in another application, ``sys.executable`` can be
+    empty or point to the host application rather than a Python interpreter.
+    """
+    executable = sys.executable
+    name = Path(executable).name.lower()
+    if not executable or not Path(executable).is_file() or not name.startswith(_PYTHON_NAMES):
+        raise ScriptError(
+            f"No Python interpreter available to run the script (sys.executable is "
+            f"{executable!r}). This happens when Python is embedded in another "
+            "application. Use taskmaestro.command with an explicit interpreter instead, "
+            "e.g. command: [/path/to/python, script.py]."
+        )
+    return executable
+
+
+def _text(value: str | bytes | None) -> str:
+    if value is None:
+        return ""
+    return value.decode(errors="replace") if isinstance(value, bytes) else value
diff --git a/taskmaestro/workflow.py b/taskmaestro/workflow.py
index e49cbb2..6c107d1 100644
--- a/taskmaestro/workflow.py
+++ b/taskmaestro/workflow.py
@@ -465,6 +465,14 @@ def _validate_types(self) -> None:
                 if cf:
                     downstream_input = get_input_type(task_cls)
                     model_fields = downstream_input.model_fields
+                    # Unknown names first: a misspelled field also leaves the
+                    # intended one uncovered, and the typo is the real error.
+                    for field_name in cf:
+                        if field_name not in model_fields:
+                            raise WorkflowDefinitionError(
+                                f"Config field '{field_name}' not found on "
+                                f"{downstream_input.__name__} (input of '{name}')"
+                            )
                     # Validate config_fields cover all required input fields
                     task_map = self.get_task_map(name)
                     map_fields = (
@@ -477,13 +485,6 @@ def _validate_types(self) -> None:
                                 f"{downstream_input.__name__} is not covered by "
                                 f"config_fields for root task '{name}'"
                             )
-                    # Validate config field names exist on the model
-                    for field_name in cf:
-                        if field_name not in model_fields:
-                            raise WorkflowDefinitionError(
-                                f"Config field '{field_name}' not found on "
-                                f"{downstream_input.__name__} (input of '{name}')"
-                            )
                 continue
             elif isinstance(deps, str):
                 # Single dependency (whole output)
diff --git a/tests/test_condition_tasks.py b/tests/test_condition_tasks.py
index c1c3b9a..fa6fce8 100644
--- a/tests/test_condition_tasks.py
+++ b/tests/test_condition_tasks.py
@@ -78,7 +78,10 @@ def test_condition_tasks_are_registered() -> None:
     for name, target in registered.items():
         module, _, attribute = target.partition(":")
         loaded[name] = getattr(import_module(module), attribute)
-    assert loaded == {
+    condition_entries = {
+        name: obj for name, obj in loaded.items() if name.startswith("condition.")
+    }
+    assert condition_entries == {
         "condition.compare": Compare,
         "condition.is_true": IsTrue,
         "condition.path_exists": PathExists,
diff --git a/tests/test_scripts.py b/tests/test_scripts.py
new file mode 100644
index 0000000..a1afe6d
--- /dev/null
+++ b/tests/test_scripts.py
@@ -0,0 +1,491 @@
+"""Tests for the built-in ``taskmaestro.python`` and ``taskmaestro.command`` tasks."""
+
+from __future__ import annotations
+
+import json
+import os
+import signal
+import subprocess
+import sys
+import textwrap
+import time
+from pathlib import Path
+from typing import Any
+
+import pytest
+from pydantic import BaseModel
+
+from taskmaestro import (
+    Command,
+    CommandInput,
+    ConfigLoadError,
+    EmptyConfig,
+    ExecutionContext,
+    JobStatus,
+    ObjectModel,
+    PythonScript,
+    PythonScriptInput,
+    ScriptError,
+    WorkflowBuilder,
+    load_workflow_from_yaml,
+    registered_task_names,
+)
+from taskmaestro.scripts import _kill_process_tree
+
+ECHO_SCRIPT = """\
+import json, os, sys
+data = None
+if "TASKMAESTRO_INPUT" in os.environ:
+    with open(os.environ["TASKMAESTRO_INPUT"]) as f:
+        data = json.load(f)
+with open(os.environ["TASKMAESTRO_OUTPUT"], "w") as f:
+    json.dump({"args": sys.argv[1:], "inputs": data, "cwd": os.getcwd(),
+               "flag": os.environ.get("FLAG")}, f)
+print("done")
+"""
+
+
+@pytest.fixture
+def ctx(tmp_path: Path) -> ExecutionContext:
+    return ExecutionContext(scratch_dir=tmp_path / "scratch")
+
+
+def _script(directory: Path, text: str, name: str = "step.py") -> Path:
+    path = directory / name
+    path.write_text(textwrap.dedent(text))
+    return path
+
+
+class Point(BaseModel):
+    x: int
+
+
+def test_python_script_exchanges_json_and_captures_output(
+    tmp_path: Path, ctx: ExecutionContext
+) -> None:
+    script = _script(tmp_path, ECHO_SCRIPT)
+    workdir = tmp_path / "work"
+    workdir.mkdir()
+
+    output = PythonScript().run(
+        PythonScriptInput(
+            script=script,
+            args=["a", "b"],
+            inputs={"point": Point(x=1), "n": 2},
+            env={"FLAG": "on"},
+            cwd=workdir,
+        ),
+        ctx,
+    )
+
+    assert output.result == {
+        "args": ["a", "b"],
+        "inputs": {"point": {"x": 1}, "n": 2},
+        "cwd": str(workdir),
+        "flag": "on",
+    }
+    assert (output.returncode, output.stdout, output.stderr) == (0, "done\n", "")
+    assert list(ctx.scratch_dir.iterdir()) == []  # Exchange directory removed.
+
+
+def test_relative_script_resolves_against_the_process_cwd(
+    tmp_path: Path, ctx: ExecutionContext, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _script(tmp_path, ECHO_SCRIPT)
+    elsewhere = tmp_path / "elsewhere"
+    elsewhere.mkdir()
+    monkeypatch.chdir(tmp_path)
+
+    output = PythonScript().run(PythonScriptInput(script=Path("step.py"), cwd=elsewhere), ctx)
+
+    assert output.result["inputs"] is None
+    assert output.result["cwd"] == str(elsewhere)
+
+
+def test_missing_output_file_gives_null_result(tmp_path: Path, ctx: ExecutionContext) -> None:
+    script = _script(tmp_path, "print('no result')\n")
+
+    output = PythonScript().run(PythonScriptInput(script=script), ctx)
+
+    assert output.result is None
+    assert output.stdout == "no result\n"
+
+
+def test_empty_output_file_gives_null_result(tmp_path: Path, ctx: ExecutionContext) -> None:
+    script = _script(
+        tmp_path, "import os\nopen(os.environ['TASKMAESTRO_OUTPUT'], 'w').write(' \\n')\n"
+    )
+
+    output = PythonScript().run(PythonScriptInput(script=script), ctx)
+
+    assert output.result is None
+
+
+def test_output_file_is_read_as_utf8(tmp_path: Path, ctx: ExecutionContext) -> None:
+    script = _script(
+        tmp_path,
+        "import os\n"
+        "open(os.environ['TASKMAESTRO_OUTPUT'], 'wb').write('\"Ærøskøbing\"'.encode('utf-8'))\n",
+    )
+
+    output = PythonScript().run(PythonScriptInput(script=script), ctx)
+
+    assert output.result == "Ærøskøbing"
+
+
+@pytest.mark.parametrize("executable", ["", "host"])
+def test_python_script_fails_without_a_python_interpreter(
+    tmp_path: Path, ctx: ExecutionContext, monkeypatch: pytest.MonkeyPatch, executable: str
+) -> None:
+    # Embedded Python: sys.executable is empty or the host application.
+    if executable:
+        host = tmp_path / "ResInsight"
+        host.write_text("")
+        executable = str(host)
+    monkeypatch.setattr(sys, "executable", executable)
+    script = _script(tmp_path, "print('never runs')\n")
+
+    with pytest.raises(ScriptError, match="No Python interpreter") as info:
+        PythonScript().run(PythonScriptInput(script=script), ctx)
+    assert info.value.returncode is None
+
+
+def test_nonzero_exit_fails_with_exit_code_and_stderr(
+    tmp_path: Path, ctx: ExecutionContext
+) -> None:
+    script = _script(tmp_path, "import sys\nprint('bad things', file=sys.stderr)\nsys.exit(3)\n")
+
+    with pytest.raises(ScriptError, match="failed with exit code 3: bad things") as info:
+        PythonScript().run(PythonScriptInput(script=script), ctx)
+
+    assert info.value.returncode == 3
+    assert info.value.stderr == "bad things\n"
+
+
+def test_nonzero_exit_without_check_is_returned(tmp_path: Path, ctx: ExecutionContext) -> None:
+    script = _script(tmp_path, "import sys\nsys.exit(3)\n")
+
+    output = PythonScript().run(PythonScriptInput(script=script, check=False), ctx)
+
+    assert output.returncode == 3
+
+
+def test_missing_script_fails_before_running(tmp_path: Path, ctx: ExecutionContext) -> None:
+    with pytest.raises(ScriptError, match=r"Script not found: .*missing\.py"):
+        PythonScript().run(PythonScriptInput(script=tmp_path / "missing.py"), ctx)
+
+
+def test_invalid_output_json_fails(tmp_path: Path, ctx: ExecutionContext) -> None:
+    script = _script(
+        tmp_path, "import os\nopen(os.environ['TASKMAESTRO_OUTPUT'], 'w').write('{oops')\n"
+    )
+
+    with pytest.raises(ScriptError, match=r"wrote invalid JSON to \$TASKMAESTRO_OUTPUT"):
+        PythonScript().run(PythonScriptInput(script=script), ctx)
+
+
+def test_runtime_only_inputs_cannot_be_sent_to_a_process(
+    tmp_path: Path, ctx: ExecutionContext
+) -> None:
+    script = _script(tmp_path, ECHO_SCRIPT)
+    opaque = ObjectModel[object](value=object())
+
+    with pytest.raises(ScriptError, match="cannot be converted to JSON"):
+        PythonScript().run(PythonScriptInput(script=script, inputs={"case": opaque}), ctx)
+
+
+def test_timeout_kills_the_process_and_keeps_partial_output(
+    tmp_path: Path, ctx: ExecutionContext
+) -> None:
+    script = _script(tmp_path, "import time\nprint('working', flush=True)\ntime.sleep(30)\n")
+
+    started = time.monotonic()
+    with pytest.raises(ScriptError, match=r"timed out after 1\.0s") as info:
+        PythonScript().run(PythonScriptInput(script=script, timeout=1.0), ctx)
+    assert time.monotonic() - started < 10
+    assert info.value.stdout == "working\n"
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell")
+def test_command_string_runs_through_the_shell(ctx: ExecutionContext) -> None:
+    output = Command().run(
+        CommandInput(
+            command=(
+                'printf "%s" "$GREETING" | tr a-z A-Z; printf \'[1, 2]\' > "$TASKMAESTRO_OUTPUT"'
+            ),
+            env={"GREETING": "hello"},
+        ),
+        ctx,
+    )
+
+    assert output.stdout == "HELLO"
+    assert output.result == [1, 2]
+
+
+def test_command_list_runs_without_a_shell(ctx: ExecutionContext) -> None:
+    output = Command().run(
+        CommandInput(command=[sys.executable, "-c", "import sys; print(sys.argv[1])", "$HOME"]),
+        ctx,
+    )
+
+    assert output.stdout == "$HOME\n"
+
+
+@pytest.mark.parametrize(
+    ("body", "settings"),
+    [
+        ("import sys; sys.exit(3)", {}),
+        ("import os; open(os.environ['TASKMAESTRO_OUTPUT'], 'w').write('{')", {}),
+        ("import time; time.sleep(30)", {"timeout": 0.5}),
+        ("pass", {"inputs": {"case": ObjectModel[object](value=object())}}),
+    ],
+    ids=["nonzero-exit", "invalid-json", "timeout", "unserializable-inputs"],
+)
+def test_exchange_directory_is_removed_on_failure(
+    tmp_path: Path, ctx: ExecutionContext, body: str, settings: dict[str, Any]
+) -> None:
+    script = _script(tmp_path, body + "\n")
+
+    with pytest.raises(ScriptError):
+        PythonScript().run(PythonScriptInput(script=script, **settings), ctx)
+
+    assert list(ctx.scratch_dir.iterdir()) == []
+
+
+class _FakeProcess:
+    """Stands in for ``Popen`` in ``_kill_process_tree`` tests."""
+
+    pid = 4242
+
+    def __init__(self) -> None:
+        self.calls: list[str] = []
+
+    def kill(self) -> None:
+        self.calls.append("kill")
+
+    def wait(self) -> int:
+        self.calls.append("wait")
+        return -9
+
+
+def _patch_killpg(monkeypatch: pytest.MonkeyPatch, error: type[OSError] | None) -> list[int]:
+    groups: list[int] = []
+
+    def killpg(pgid: int, sig: int) -> None:
+        groups.append(pgid)
+        if error is not None:
+            raise error()
+
+    monkeypatch.setattr(os, "killpg", killpg, raising=False)
+    monkeypatch.setattr(signal, "SIGKILL", 9, raising=False)
+    return groups
+
+
+@pytest.mark.parametrize(
+    ("error", "expected_calls"),
+    [
+        (None, ["wait"]),
+        (ProcessLookupError, ["wait"]),  # Group already gone: nothing left to kill.
+        (PermissionError, ["kill", "wait"]),  # Fall back to the direct child.
+    ],
+)
+def test_kill_process_tree_on_posix(
+    monkeypatch: pytest.MonkeyPatch, error: type[OSError] | None, expected_calls: list[str]
+) -> None:
+    monkeypatch.setattr(sys, "platform", "linux")
+    groups = _patch_killpg(monkeypatch, error)
+    process = _FakeProcess()
+
+    _kill_process_tree(process)  # type: ignore[arg-type]
+
+    assert groups == [process.pid]
+    assert process.calls == expected_calls
+
+
+def test_kill_process_tree_on_windows_kills_the_process(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(sys, "platform", "win32")
+    groups = _patch_killpg(monkeypatch, None)
+    process = _FakeProcess()
+
+    _kill_process_tree(process)  # type: ignore[arg-type]
+
+    assert groups == []
+    assert process.calls == ["kill", "wait"]
+
+
+def _assert_process_gone(pid_file: Path) -> None:
+    pid = int(pid_file.read_text())
+    for _ in range(50):  # The killed process may need a moment to be reaped.
+        try:
+            os.kill(pid, 0)
+        except ProcessLookupError:
+            return
+        time.sleep(0.05)
+    os.kill(pid, signal.SIGKILL)
+    pytest.fail(f"Process {pid} survived the timeout")
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups")
+def test_timeout_kills_processes_started_by_the_shell(
+    tmp_path: Path, ctx: ExecutionContext
+) -> None:
+    pid_file = tmp_path / "pid"
+    # No ``exec``: sleep is a child of the shell, not the shell itself.
+    command = f'sleep 30 & echo $! > "{pid_file}"; wait'
+
+    started = time.monotonic()
+    with pytest.raises(ScriptError, match="timed out"):
+        Command().run(CommandInput(command=command, timeout=1.0), ctx)
+
+    assert time.monotonic() - started < 10
+    _assert_process_gone(pid_file)
+
+
+def test_command_that_cannot_start_fails(ctx: ExecutionContext) -> None:
+    with pytest.raises(ScriptError, match="could not be started"):
+        Command().run(CommandInput(command=["/nonexistent/taskmaestro-tool"]), ctx)
+
+
+def test_empty_command_fails(ctx: ExecutionContext) -> None:
+    with pytest.raises(ScriptError, match="Command is empty"):
+        Command().run(CommandInput(command=[]), ctx)
+
+
+@pytest.mark.skipif(not hasattr(signal, "SIGALRM"), reason="Runner timeouts need SIGALRM")
+def test_runner_timeout_kills_a_running_process(tmp_path: Path) -> None:
+    pid_file = tmp_path / "pid"
+    builder = WorkflowBuilder("slow")
+    builder.task(Command, config_fields=["command"])
+    command = ["sh", "-c", f'echo $$ > "{pid_file}"; exec sleep 30']
+
+    started = time.monotonic()
+    job = builder.build().run(
+        EmptyConfig(),
+        task_config={"command": {"command": command}},
+        ctx=ExecutionContext(scratch_dir=tmp_path),
+        timeout_seconds=1,
+    )
+
+    assert job.status == JobStatus.FAILED
+    assert time.monotonic() - started < 10
+    with pytest.raises(ProcessLookupError):
+        os.kill(int(pid_file.read_text()), 0)
+
+
+@pytest.mark.skipif(not hasattr(signal, "SIGALRM"), reason="Runner timeouts need SIGALRM")
+def test_runner_timeout_kills_processes_started_by_the_shell(tmp_path: Path) -> None:
+    pid_file = tmp_path / "pid"
+    builder = WorkflowBuilder("slow")
+    builder.task(Command, config_fields=["command"])
+    command = f'sleep 30 & echo $! > "{pid_file}"; wait'
+
+    started = time.monotonic()
+    job = builder.build().run(
+        EmptyConfig(),
+        task_config={"command": {"command": command}},
+        ctx=ExecutionContext(scratch_dir=tmp_path),
+        timeout_seconds=1,
+    )
+
+    assert job.status == JobStatus.FAILED
+    assert time.monotonic() - started < 10
+    _assert_process_gone(pid_file)
+
+
+def test_script_tasks_are_registered() -> None:
+    assert {"taskmaestro.python", "taskmaestro.command"} <= registered_task_names()
+
+
+def _yaml_files(tmp_path: Path, workflow: str, inputs: str = "{}\n") -> tuple[Path, Path]:
+    workflow_path = tmp_path / "workflow.yaml"
+    input_path = tmp_path / "input.yaml"
+    workflow_path.write_text(textwrap.dedent(workflow))
+    input_path.write_text(inputs)
+    return workflow_path, input_path
+
+
+def test_yaml_workflow_connects_script_results_with_collect(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _script(tmp_path, ECHO_SCRIPT)
+    monkeypatch.chdir(tmp_path)
+    workflow_path, input_path = _yaml_files(
+        tmp_path,
+        """\
+        workflow:
+          name: scripts
+          tasks:
+            - task: taskmaestro.python
+              name: first
+            - task: taskmaestro.command
+              name: second
+              depends_on:
+                inputs:
+                  collect: {args: [first, result], code: [first, returncode]}
+        """,
+        f"first: {{script: step.py, args: [two]}}\n"
+        f"second: {{command: [{json.dumps(sys.executable)}, step.py]}}\n",
+    )
+
+    job = load_workflow_from_yaml(workflow_path, input_path).run()
+
+    assert job.status == JobStatus.COMPLETED, job.error
+    assert job.result is not None
+    assert job.result.result["inputs"] == {
+        "args": {"args": ["two"], "inputs": None, "cwd": str(tmp_path), "flag": None},
+        "code": 0,
+    }
+
+
+def test_whole_output_cannot_be_connected_to_inputs(tmp_path: Path) -> None:
+    workflow_path, input_path = _yaml_files(
+        tmp_path,
+        """\
+        workflow:
+          name: scripts
+          tasks:
+            - task: taskmaestro.command
+              name: first
+            - task: taskmaestro.command
+              name: second
+              depends_on: {inputs: first}
+        """,
+        'first: {command: "true"}\nsecond: {command: "true"}\n',
+    )
+
+    with pytest.raises(ConfigLoadError, match=r"expects dict\[str, Any\]"):
+        load_workflow_from_yaml(workflow_path, input_path)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="Example uses sh and wc")
+def test_scripts_example_runs(tmp_path: Path) -> None:
+    root = Path(__file__).resolve().parents[1]
+    input_path = tmp_path / "input.yaml"
+    example_input = (root / "examples/scripts/input.yaml").read_text()
+    input_path.write_text(
+        example_input.replace("/tmp/taskmaestro-script-tasks", json.dumps(str(tmp_path)))
+    )
+
+    run = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "taskmaestro",
+            "run",
+            "examples/scripts/workflow.yaml",
+            "--input",
+            str(input_path),
+            "--json",
+        ],
+        cwd=root,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert run.returncode == 0, run.stderr
+    report = tmp_path / "word_count.json"
+    assert json.loads(run.stdout)["result"]["result"] == {"report": str(report)}
+    saved = json.loads(report.read_text())
+    assert saved["counts"]["words"] == 32
+    assert saved["wc_output"].split()[0] == "172"
diff --git a/tests/test_workflow.py b/tests/test_workflow.py
index a1dced7..488637b 100644
--- a/tests/test_workflow.py
+++ b/tests/test_workflow.py
@@ -1051,6 +1051,12 @@ def test_root_task_config_field_not_on_model_raises(self) -> None:
                 .build()
             )
 
+    def test_root_task_misspelled_config_field_is_reported_before_missing_ones(self) -> None:
+        with pytest.raises(WorkflowDefinitionError, match="Config field 'cuont'"):
+            Workflow.builder("bad").add_task(
+                ConfigOnlyTask, config_fields=["path", "cuont"]
+            ).build()
+
     def test_single_dep_with_config_fields(self) -> None:
         """Single dep + config_fields: upstream fields + config cover all required."""
         wf = (