Python and shell script

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

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:
 

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.

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}

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.

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"

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

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]

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"

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",

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