From 303450f0d5ce4156b22500a9a96cd90c3b51430c Mon Sep 17 00:00:00 2001 From: Kristian Bendiksen Date: Wed, 21 Oct 2026 13:45:58 +0200 Subject: [PATCH 1/4] Add public workflow definition and introspection API Move generic workflow-definition logic from ResInsight's taskmaestro helper into taskmaestro so editors can use public API instead of private internals: - definition: convert workflow.yaml/input.yaml to and from an editable graph definition, from registered workflows, with atomic writes, backups and a value filter for saved inputs - introspection: describe workflows and task schemas, with a schema generator hook for extra type annotations; the CLI now uses it - discovery.load_registry: load all entry points, reporting broken ones - yaml_config: public yaml_load, load_workflow_only, resolve_task and workflow_imports - DefinitionError exception Bump version to 0.4.0. --- AGENTS.md | 4 +- pyproject.toml | 2 +- taskmaestro/__init__.py | 4 +- taskmaestro/cli.py | 261 ++---------- taskmaestro/definition.py | 781 +++++++++++++++++++++++++++++++++++ taskmaestro/discovery.py | 58 +++ taskmaestro/exceptions.py | 16 + taskmaestro/introspection.py | 317 ++++++++++++++ taskmaestro/yaml_config.py | 54 ++- tests/test_definition.py | 617 +++++++++++++++++++++++++++ tests/test_discovery.py | 47 +++ 11 files changed, 1910 insertions(+), 251 deletions(-) create mode 100644 taskmaestro/definition.py create mode 100644 taskmaestro/introspection.py create mode 100644 tests/test_definition.py diff --git a/AGENTS.md b/AGENTS.md index f8b0969..ac45350 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -26,7 +26,9 @@ mypy taskmaestro # type check (strict mode) | `taskmaestro/dependencies.py`, `taskmaestro/mapping.py` | Task handles, output references, `collect()`, mapped-task configuration | | `taskmaestro/workflow_task.py` | Nested workflows wrapped as tasks | | `taskmaestro/yaml_config.py` | YAML parsing, task imports, workflow and input validation | -| `taskmaestro/cli.py`, `taskmaestro/discovery.py` | CLI (`run`, `validate`, `graph`, `tasks list/describe`, `workflow describe`), plugin entry-point discovery | +| `taskmaestro/cli.py`, `taskmaestro/discovery.py` | CLI (`run`, `validate`, `graph`, `tasks list/describe`, `workflow describe`), plugin entry-point discovery (`load_registry` collects broken entry points instead of raising) | +| `taskmaestro/introspection.py` | Public JSON helpers: `describe_workflow`, `describe_yaml_workflow`, `task_descriptor`, `TaskSchemaGenerator` (subclass and override `type_annotations` to add schema keys), `jsonable`, `error_payload` | +| `taskmaestro/definition.py` | Editable graph form of workflow YAML: `load_definition`, `definition_from_workflow`, `write_definition` (atomic, optional `.bak`, `value_filter`); raises `DefinitionError` | | `taskmaestro/job.py` | `Job[C]`, `JobStatus`, `TaskStatus`, `TaskResult` dataclass | | `taskmaestro/runner.py` | `Runner` — topological execution, timeout via `signal.alarm`, hook dispatch | | `taskmaestro/hooks/base.py` | `Event` StrEnum, `Hook` protocol, `BaseHook` no-op base | diff --git a/pyproject.toml b/pyproject.toml index f74a383..d65fa83 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "taskmaestro" -version = "0.3.0" +version = "0.4.0" description = "A Python 3.12+ library for defining and executing typed DAG task workflows" readme = "README.md" requires-python = ">=3.12" diff --git a/taskmaestro/__init__.py b/taskmaestro/__init__.py index efd0471..d15923b 100644 --- a/taskmaestro/__init__.py +++ b/taskmaestro/__init__.py @@ -1,6 +1,6 @@ """Workflow Runner: typed DAG task workflows with Pydantic models.""" -__version__ = "0.3.0" +__version__ = "0.4.0" from taskmaestro.context import ExecutionContext from taskmaestro.dependencies import OutputHandle, TaskHandle, collect @@ -17,6 +17,7 @@ from taskmaestro.exceptions import ( ConfigLoadError, CycleDetectedError, + DefinitionError, IncompleteInputError, JobStateError, MappedTaskExecutionError, @@ -47,6 +48,7 @@ "WORKFLOW_ENTRY_POINT_GROUP", "ConfigLoadError", "CycleDetectedError", + "DefinitionError", "EmptyConfig", "ExecutionContext", "HookError", diff --git a/taskmaestro/cli.py b/taskmaestro/cli.py index 5093eec..fd3e493 100644 --- a/taskmaestro/cli.py +++ b/taskmaestro/cli.py @@ -5,27 +5,30 @@ import argparse import json import logging -import math import sys from collections.abc import Iterator, Sequence from contextlib import contextmanager, redirect_stdout -from dataclasses import asdict -from datetime import date -from pathlib import Path, PurePath +from pathlib import Path from typing import Any -from pydantic import ValidationError from pydantic.errors import PydanticInvalidForJsonSchema, PydanticSchemaGenerationError -from pydantic.json_schema import GenerateJsonSchema, JsonSchemaValue -from pydantic_core import core_schema -from taskmaestro.dependencies import CollectionRef, OutputRef from taskmaestro.discovery import get_registered_task, registered_task_names -from taskmaestro.exceptions import ConfigLoadError, PluginLoadError, WorkflowDefinitionError -from taskmaestro.job import EmptyConfig, Job, JobConfiguration, JobStatus +from taskmaestro.exceptions import ConfigLoadError, PluginLoadError +from taskmaestro.introspection import ( + TaskSchemaGenerator, + dependency_spec, + describe_yaml_workflow, + error_payload, + jsonable, +) +from taskmaestro.job import JobStatus from taskmaestro.task import get_input_type, get_output_type -from taskmaestro.workflow import Workflow -from taskmaestro.yaml_config import LoadedWorkflow, _load_workflow_only, load_workflow_from_yaml +from taskmaestro.yaml_config import ( + LoadedWorkflow, + load_workflow_from_yaml, + workflow_imports, +) def _add_workflow_arguments(parser: argparse.ArgumentParser) -> None: @@ -36,12 +39,8 @@ def _add_workflow_arguments(parser: argparse.ArgumentParser) -> None: @contextmanager def _workflow_imports(path: str) -> Iterator[None]: """Make task modules next to a YAML file importable for the duration of loading.""" - original_path = sys.path.copy() - sys.path.insert(0, str(Path(path).resolve().parent)) - try: + with workflow_imports(Path(path).resolve().parent): yield - finally: - sys.path[:] = original_path def _load(args: argparse.Namespace) -> LoadedWorkflow: @@ -49,31 +48,10 @@ def _load(args: argparse.Namespace) -> LoadedWorkflow: return load_workflow_from_yaml(args.workflow, args.input) -def _error(code: str, exc: Exception, message: str, *, task: str | None = None) -> dict[str, Any]: - """Return safe diagnostics from an exception and its causes.""" - issues: list[dict[str, str]] = [] - cause: BaseException | None = exc - while cause is not None: - if isinstance(cause, ValidationError): - issues = [ - {"field": ".".join(map(str, issue["loc"])), "code": issue["type"]} - for issue in cause.errors(include_input=False, include_context=False) - ] - break - if isinstance(cause, WorkflowDefinitionError): - task = task or cause.task_name - issues = [{"field": field, "code": "missing"} for field in cause.fields] - if issues: - break - cause = cause.__cause__ - return { - "code": code, - "type": type(exc).__name__, - "message": message, - "task": task, - "field": issues[0]["field"] if issues else None, - "issues": issues, - } +_error = error_payload +_jsonable = jsonable +_dependency_spec = dependency_spec +_TaskSchemaGenerator = TaskSchemaGenerator def _run(args: argparse.Namespace) -> int: @@ -171,203 +149,10 @@ def _tasks_list(args: argparse.Namespace) -> int: return 0 -class _TaskSchemaGenerator(GenerateJsonSchema): - """Describe runtime-only Python objects without pretending they accept JSON.""" - - def is_instance_schema(self, schema: core_schema.IsInstanceSchema) -> JsonSchemaValue: - cls = schema["cls"] - return { - "not": {}, # No JSON value can satisfy an isinstance check for this object. - "x-taskmaestro-opaque": True, - "x-taskmaestro-python-type": f"{cls.__module__}.{cls.__qualname__}", - } - - -def _output_ref(ref: OutputRef) -> dict[str, str | None]: - return {"task": ref.task_name, "field": ref.output_field} - - -def _dependency_spec(deps: Any) -> Any: - """Render validated dependencies, including field routing and collections.""" - if deps is None: - return None - if isinstance(deps, str): - return {"task": deps, "field": None} - if isinstance(deps, tuple): - return {"task": deps[0], "field": deps[1]} - result: dict[str, Any] = {} - for field, ref in deps.items(): - if isinstance(ref, CollectionRef): - members: list[Any] | dict[str, Any] - if ref.kind == "keyed": - members = {key: _output_ref(item) for key, item in ref.keyed_members} - else: - members = [_output_ref(item) for item in ref.positional_members] - result[field] = {"collect": {"kind": ref.kind, "members": members}} - elif isinstance(ref, tuple): - result[field] = {"task": ref[0], "field": ref[1]} - else: - result[field] = {"task": ref, "field": None} - return result - - -def _jsonable_key(key: object, path: str) -> str: - """Convert a YAML mapping key to the JSON object key it would get.""" - value = _jsonable(key, path) - if isinstance(value, str): - return value - if isinstance(value, bool | int | float) or value is None: - return json.dumps(value) - raise TypeError(f"Configuration value '{path}' has an unsupported mapping key type") - - -def _jsonable(value: object, path: str) -> Any: - """Convert a raw configuration value (as loaded from YAML) to plain JSON data. - - Dates and datetimes become ISO 8601 strings and paths become strings; - everything else must already be JSON-compatible. ``path`` names the value - in error messages, which deliberately never include the value itself. - """ - if value is None or isinstance(value, bool | int | str): - return value - if isinstance(value, float): - if not math.isfinite(value): - raise ValueError(f"Configuration value '{path}' is not a finite number") - return value - if isinstance(value, date): # Includes datetime. - return value.isoformat() - if isinstance(value, PurePath): - return str(value) - if isinstance(value, dict): - result: dict[str, Any] = {} - for key, item in value.items(): - json_key = _jsonable_key(key, path) - if json_key in result: - raise ValueError( - f"Configuration value '{path}' has keys that collide as JSON key '{json_key}'" - ) - result[json_key] = _jsonable(item, f"{path}.{json_key}") - return result - if isinstance(value, list | tuple): - return [_jsonable(item, f"{path}[{index}]") for index, item in enumerate(value)] - if isinstance(value, set | frozenset): - items = [_jsonable(item, f"{path}[]") for item in value] - try: - return sorted(items) - except TypeError: - # Mixed types: order by canonical JSON text so output is deterministic. - return sorted(items, key=lambda item: json.dumps(item, sort_keys=True)) - raise TypeError(f"Configuration value '{path}' has unsupported type {type(value).__name__}") - - -def _missing_config_fields(workflow: Workflow, config: JobConfiguration) -> dict[str, list[str]]: - """Return the declared config fields that the input does not supply, per task.""" - return { - name: sorted(workflow.get_config_fields(name) - config.config_fields_for_task(name)) - for name, _task in workflow.topological_order() - } - - -def _fill_missing( - workflow: Workflow, config: JobConfiguration, missing: dict[str, list[str]] -) -> JobConfiguration: - """Return a copy of ``config`` with placeholder values for missing config fields. - - ``Job`` only checks that these fields are present, so placeholders let it run - its remaining validation. Map sources get no placeholder, so a missing one is - still rejected with its own error. - """ - filled: dict[str, dict[str, Any]] = {} - for name, fields in missing.items(): - task_map = workflow.get_task_map(name) - map_source = task_map.over if task_map is not None else None - placeholders = {field: None for field in fields if field != map_source} - filled[name] = {**placeholders, **config.get_config_for_task(name)} - return JobConfiguration(filled) - - -def _workflow_description( - workflow: Workflow, - *, - configured: dict[str, list[str]] | None, - missing: dict[str, list[str]] | None = None, - values: dict[str, dict[str, Any]] | None = None, -) -> dict[str, Any]: - tasks: list[dict[str, Any]] = [] - for name, task in workflow.topological_order(): - input_type = get_input_type(task) - output_type = workflow.get_output_annotation(name) - task_map = workflow.get_task_map(name) - tasks.append( - { - "name": name, - "python_type": f"{task.__module__}.{task.__qualname__}", - "depends_on": _dependency_spec(workflow.get_dependencies(name)), - "config_fields": sorted(workflow.get_config_fields(name)), - "provided_config_fields": configured[name] if configured is not None else None, - "missing_config_fields": missing[name] if missing is not None else None, - "config_values": values[name] if values is not None else None, - "required_input_fields": sorted( - field for field, info in input_type.model_fields.items() if info.is_required() - ), - "map": asdict(task_map) if task_map is not None else None, - "input_schema": input_type.model_json_schema( - schema_generator=_TaskSchemaGenerator - ), - "output_schema": output_type.model_json_schema( - schema_generator=_TaskSchemaGenerator - ), - } - ) - return {"workflow": workflow.name, "result_task": workflow.result_task_name, "tasks": tasks} - - def _workflow_describe(args: argparse.Namespace) -> int: - try: - with redirect_stdout(sys.stderr), _workflow_imports(args.workflow): - workflow, config = _load_workflow_only( - Path(args.workflow), Path(args.input) if args.input else None - ) - configured = None - missing = None - values = None - if args.input is not None: - assert config is not None - missing = _missing_config_fields(workflow, config) - # Check the rest of the supplied config (map sources, root inputs) - # without constructing hooks or executing any tasks. Missing config - # fields are reported rather than rejected: the input may be a - # template whose remaining fields the caller fills in. - Job( - workflow, - EmptyConfig(), - job_configuration=_fill_missing(workflow, config, missing), - ) - configured = { - name: sorted(config.config_fields_for_task(name)) - for name, _task in workflow.topological_order() - } - # Raw input values, before Pydantic validation, so opaque or - # runtime-only markers pass through unchanged. - values = { - name: { - field: _jsonable(value, f"{name}.{field}") - for field, value in sorted(config.get_config_for_task(name).items()) - } - for name, _task in workflow.topological_order() - } - description = _workflow_description( - workflow, configured=configured, missing=missing, values=values - ) - except WorkflowDefinitionError as exc: - raise ConfigLoadError(f"Job validation failed: {exc}") from exc - except ( - TypeError, - ValueError, - PydanticInvalidForJsonSchema, - PydanticSchemaGenerationError, - ) as exc: - raise ConfigLoadError(f"Cannot describe workflow: {exc}") from exc + description = describe_yaml_workflow( + Path(args.workflow), Path(args.input) if args.input else None + ) print(json.dumps(description, indent=None if args.json else 2)) return 0 diff --git a/taskmaestro/definition.py b/taskmaestro/definition.py new file mode 100644 index 0000000..caf31e1 --- /dev/null +++ b/taskmaestro/definition.py @@ -0,0 +1,781 @@ +"""Workflow definitions: an editable graph form of workflow YAML files. + +A *definition* is a JSON-compatible document that describes a workflow as a +plain graph: task instances (``nodes``), connections between task outputs and +inputs (``edges``) and per-task configuration values (``inputs``). Editors +modify definitions; this module converts them from and to the standard +``workflow.yaml``/``input.yaml`` pair and from registered workflows. + +Problems are reported as ``DefinitionError``. +""" + +from __future__ import annotations + +import os +import shutil +import tempfile +from collections.abc import Callable +from itertools import pairwise +from pathlib import Path +from typing import Any + +import yaml +from pydantic.json_schema import GenerateJsonSchema + +from taskmaestro.discovery import registered_task_names +from taskmaestro.exceptions import ConfigLoadError, DefinitionError +from taskmaestro.introspection import ( + TaskSchemaGenerator, + describe_workflow, + describe_yaml_workflow, + error_payload, + jsonable, + task_descriptor, +) +from taskmaestro.yaml_config import import_class, resolve_task, workflow_imports, yaml_load + +SchemaGenerator = type[GenerateJsonSchema] +ValueFilter = Callable[[Any], Any] +DEFINITION_FORMAT = 1 +WORKFLOW_FILE = "workflow.yaml" +INPUT_FILE = "input.yaml" +BACKUP_SUFFIX = ".bak" + + +# --- Small helpers --- + + +def _edge( + upstream: str, + output: str | None, + downstream: str, + input_field: str | None, + *, + collect: str | None = None, + key: str | None = None, +) -> dict[str, Any]: + edge = {"from": upstream, "output": output, "to": downstream, "input": input_field} + if collect is not None: + # A member of a collected input: "list" members keep their order, "dict" members a key + edge["collect"] = collect + edge["key"] = key + return edge + + +def _map_json(task_map: Any) -> dict[str, Any]: + return { + "over": task_map.over, + "key_as": task_map.key_as, + "value_as": task_map.value_as, + "error_mode": task_map.error_mode, + } + + +def _empty_definition(name: str) -> dict[str, Any]: + return { + "format": DEFINITION_FORMAT, + "name": name, + "result_task": None, + "editable": True, + "readonly_reasons": [], + "warnings": [], + "source": {"workflow_yaml": None, "input_yaml": None, "registered_id": None}, + "header_comment": "", + "passthrough": {}, + "nodes": [], + "edges": [], + "inputs": {}, + "task_types": {}, + "describe": None, + "describe_error": None, + } + + +def header_comment(text: str) -> str: + """Return the comment block at the top of a YAML file.""" + lines: list[str] = [] + for line in text.splitlines(): + stripped = line.strip() + if stripped.startswith("#"): + lines.append(line.rstrip()) + elif stripped == "" and lines: + lines.append("") + elif stripped != "": + break + while lines and lines[-1] == "": + lines.pop() + return "\n".join(lines) + + +def _jsonable(value: Any, path: str) -> Any: + try: + return jsonable(value, path) + except (TypeError, ValueError) as exc: + raise DefinitionError(str(exc)) from exc + + +def _yaml_load(text: str, what: str) -> Any: + try: + return yaml_load(text) + except yaml.YAMLError as exc: + raise DefinitionError(f"{what} parse error: {exc}") from exc + + +def _is_identifier(name: Any) -> bool: + return isinstance(name, str) and name.isidentifier() + + +# --- Loading YAML --- + + +class TaskResolver: + """Resolve the ``task:`` strings of a YAML file like taskmaestro's loader.""" + + def __init__(self) -> None: + self.installed = registered_task_names() + + def resolve(self, task_id: str) -> type[Any]: + try: + return resolve_task(task_id, self.installed) + except ConfigLoadError as exc: + raise DefinitionError(str(exc)) from exc + + +def _inner_workflow_name(base_dir: Path, workflow_path: str) -> str: + try: + raw = yaml.safe_load((base_dir / workflow_path).read_text()) + return str(raw["workflow"]["name"]) + except Exception: + return Path(workflow_path).stem + + +def _add_task_type( + definition: dict[str, Any], + task_id: str, + cls: type[Any], + schema_generator: SchemaGenerator, +) -> None: + if task_id in definition["task_types"]: + return + try: + definition["task_types"][task_id] = task_descriptor(task_id, cls, schema_generator) + except Exception as exc: + definition["readonly_reasons"].append(f"Cannot describe task '{task_id}': {exc}") + + +def _load_nodes( + definition: dict[str, Any], config: Any, base_dir: Path, schema_generator: SchemaGenerator +) -> dict[str, set[str]]: + """Create one node per YAML entry and return the reference lookup.""" + resolver = TaskResolver() + candidates: dict[str, set[str]] = {} + for entry in config.workflow.tasks: + node: dict[str, Any] = {"name": "", "task": entry.task, "config_fields": []} + if entry.workflow: + key = entry.workflow + name = entry.name or _inner_workflow_name(base_dir, entry.workflow) + node["workflow"] = entry.workflow + definition["readonly_reasons"].append( + f"Task '{name}' is a nested workflow ('{entry.workflow}')" + ) + else: + key = entry.task + try: + cls = resolver.resolve(entry.task) + except DefinitionError as exc: + cls = None + definition["readonly_reasons"].append( + f"Cannot load task '{entry.task}': {exc.message}" + ) + name = entry.name or (cls.name if cls is not None else entry.task) + if cls is not None: + _add_task_type(definition, entry.task, cls, schema_generator) + if entry.map is not None: + node["map"] = _map_json(entry.map) + if any(existing["name"] == name for existing in definition["nodes"]): + raise DefinitionError(f"Duplicate task name '{name}'", task=name) + node["name"] = name + node["config_fields"] = list(entry.config_fields or []) + node["_declared_config"] = entry.config_fields is not None + definition["nodes"].append(node) + candidates.setdefault(key, set()).add(name) + if entry.name: + candidates.setdefault(entry.name, set()).add(name) + return candidates + + +def _resolve_reference(candidates: dict[str, set[str]], ref: str, context: str) -> str: + names = candidates.get(ref) + where = f" for task '{context}'" if context else "" + if names is None: + raise DefinitionError(f"Dependency '{ref}'{where} not found", task=context or None) + if len(names) > 1: + raise DefinitionError( + f"Dependency '{ref}'{where} is ambiguous; it matches {sorted(names)}. " + "Use the instance name.", + task=context or None, + ) + return next(iter(names)) + + +def _field_reference(raw: Any, task: str, field: str | None) -> tuple[str, str | None]: + if isinstance(raw, str): + return raw, None + if isinstance(raw, list) and len(raw) == 2 and all(isinstance(item, str) for item in raw): + return raw[0], raw[1] + where = f" for field '{field}'" if field else "" + raise DefinitionError( + f"Invalid dependency {raw!r}{where} on task '{task}'", task=task, field=field + ) + + +def _load_edges(definition: dict[str, Any], config: Any, candidates: dict[str, set[str]]) -> None: + entries = config.workflow.tasks + nodes = definition["nodes"] + linear = all(entry.depends_on is None and entry.map is None for entry in entries) + if linear: + for previous, node in pairwise(nodes): + definition["edges"].append(_edge(previous["name"], None, node["name"], None)) + if len(nodes) > 1: + definition["warnings"].append( + "The workflow has no 'depends_on' entries; tasks were chained in " + "file order and the dependencies will be saved explicitly" + ) + return + + for entry, node in zip(entries, nodes, strict=True): + name = node["name"] + deps = entry.depends_on + if deps is None: + continue + if isinstance(deps, dict): + for field, raw in deps.items(): + if isinstance(raw, dict) and set(raw) == {"collect"}: + _load_collect(definition, candidates, raw["collect"], name, field) + continue + ref, output = _field_reference(raw, name, field) + upstream = _resolve_reference(candidates, ref, name) + definition["edges"].append(_edge(upstream, output, name, field)) + continue + ref, output = _field_reference(deps, name, None) + upstream = _resolve_reference(candidates, ref, name) + definition["edges"].append(_edge(upstream, output, name, None)) + + +def _load_collect( + definition: dict[str, Any], + candidates: dict[str, set[str]], + members: Any, + name: str, + field: str, +) -> None: + if isinstance(members, list): + keyed: list[tuple[str | None, Any]] = [(None, member) for member in members] + kind = "list" + elif isinstance(members, dict): + keyed = [(str(key), member) for key, member in members.items()] + kind = "dict" + else: + raise DefinitionError( + f"'collect' for field '{field}' on task '{name}' must be a list or mapping", + task=name, + field=field, + ) + for key, member in keyed: + ref, output = _field_reference(member, name, field) + upstream = _resolve_reference(candidates, ref, name) + definition["edges"].append(_edge(upstream, output, name, field, collect=kind, key=key)) + + +def _load_inputs(definition: dict[str, Any], input_path: Path | None) -> None: + raw_input: Any = {} + if input_path is not None and input_path.exists(): + raw_input = _yaml_load(input_path.read_text(), "Input YAML") or {} + if not isinstance(raw_input, dict): + raise DefinitionError("Input YAML file must contain a mapping") + + nodes = {node["name"]: node for node in definition["nodes"]} + for key, values in raw_input.items(): + if key not in nodes: + definition["warnings"].append(f"Dropped input values for unknown task '{key}'") + continue + if values is None: + values = {} + if not isinstance(values, dict): + raise DefinitionError(f"Input value for task '{key}' must be a mapping", task=key) + definition["inputs"][key] = _jsonable(values, str(key)) + + for node in nodes.values(): + values = definition["inputs"].setdefault(node["name"], {}) + map_source = node.get("map", {}).get("over") if node.get("map") else None + if not node.pop("_declared_config"): + # taskmaestro uses the input keys when config_fields is omitted. + node["config_fields"] = [field for field in values if field != map_source] + wired = { + edge["input"] + for edge in definition["edges"] + if edge["to"] == node["name"] and edge["input"] is not None + } + for field in sorted(wired & set(node["config_fields"])): + node["config_fields"].remove(field) + if field in values: + definition["warnings"].append( + f"Input '{node['name']}.{field}' is connected to an upstream task; " + "its configured value was dropped" + ) + del values[field] + if definition["editable"]: + for field in [f for f in values if f not in node["config_fields"]]: + if field == map_source: + continue + definition["warnings"].append( + f"Dropped value for '{node['name']}.{field}', which is not a " + "configuration field" + ) + del values[field] + definition["inputs"] = {name: definition["inputs"][name] for name in nodes} + + +def describe_yaml( + workflow_path: Path, + input_path: Path | None = None, + schema_generator: SchemaGenerator = TaskSchemaGenerator, +) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: + """Describe a YAML workflow. Returns ``(description, error)``.""" + try: + return describe_yaml_workflow(workflow_path, input_path, schema_generator), None + except Exception as exc: # Errors raised by task modules at import time too. + return None, error_payload("configuration_error", exc, str(exc)) + + +def load_definition( + workflow_dir: Path, + *, + describe: bool = False, + schema_generator: SchemaGenerator = TaskSchemaGenerator, +) -> dict[str, Any]: + """Load ``workflow.yaml`` (and ``input.yaml``) from a workflow folder. + + ``schema_generator`` produces the JSON schemas of task inputs and outputs. + """ + from pydantic import ValidationError + + from taskmaestro.yaml_config import YamlWorkflowConfig + + workflow_dir = workflow_dir.resolve() + workflow_path = workflow_dir / WORKFLOW_FILE + input_path = workflow_dir / INPUT_FILE + try: + text = workflow_path.read_text() + except OSError as exc: + raise DefinitionError(f"Cannot read '{workflow_path}': {exc}") from exc + raw = _yaml_load(text, "YAML") + if not isinstance(raw, dict): + raise DefinitionError("YAML file must contain a mapping at top level") + try: + config = YamlWorkflowConfig.model_validate(raw) + except ValidationError as exc: + raise DefinitionError(f"YAML schema validation error: {exc}") from exc + + definition = _empty_definition(config.workflow.name) + definition["source"] = { + "workflow_yaml": str(workflow_path), + "input_yaml": str(input_path) if input_path.exists() else None, + "registered_id": None, + } + definition["header_comment"] = header_comment(text) + definition["passthrough"] = { + key: _jsonable(value, str(key)) for key, value in raw.items() if key != "workflow" + } + + with workflow_imports(workflow_dir): + candidates = _load_nodes(definition, config, workflow_dir, schema_generator) + _load_edges(definition, config, candidates) + if config.workflow.result_task: + definition["result_task"] = _resolve_reference( + candidates, config.workflow.result_task, "" + ) + definition["editable"] = not definition["readonly_reasons"] + _load_inputs(definition, input_path) + + if describe: + definition["describe"], definition["describe_error"] = describe_yaml( + workflow_path, + input_path if input_path.exists() else None, + schema_generator, + ) + return definition + + +# --- Registered workflows --- + + +def _class_task_id(cls: type[Any], class_ids: dict[type[Any], str]) -> str | None: + """Return a task identifier that resolves back to ``cls``, if any.""" + if cls in class_ids: + return class_ids[cls] + if "" in cls.__qualname__: + return None + path = f"{cls.__module__}.{cls.__qualname__}" + try: + return path if import_class(path) is cls else None + except ConfigLoadError: + return None + + +def definition_from_workflow( + workflow: Any, + registered_id: str | None, + class_ids: dict[type[Any], str], + *, + describe: bool = False, + schema_generator: SchemaGenerator = TaskSchemaGenerator, +) -> dict[str, Any]: + """Convert an in-memory workflow to a definition. + + ``class_ids`` maps registered task classes to their entry-point names. + """ + from taskmaestro.dependencies import CollectionRef + + definition = _empty_definition(workflow.name) + definition["source"]["registered_id"] = registered_id + reasons = definition["readonly_reasons"] + + has_dependents: set[str] = set() + for name, cls in workflow.topological_order(): + node: dict[str, Any] = { + "name": name, + "task": None, + "config_fields": sorted(workflow.get_config_fields(name)), + } + if hasattr(cls, "_inner_workflow"): + reasons.append(f"Task '{name}' is a nested workflow") + else: + task_id = _class_task_id(cls, class_ids) + if task_id is None: + reasons.append( + f"Task '{name}' uses class '{cls.__module__}.{cls.__qualname__}', " + "which cannot be referenced from YAML" + ) + else: + node["task"] = task_id + _add_task_type(definition, task_id, cls, schema_generator) + task_map = workflow.get_task_map(name) + if task_map is not None: + node["map"] = _map_json(task_map) + definition["nodes"].append(node) + definition["inputs"][name] = {} + + deps = workflow.get_dependencies(name) + if deps is None: + continue + if isinstance(deps, str): + definition["edges"].append(_edge(deps, None, name, None)) + elif isinstance(deps, tuple): + definition["edges"].append(_edge(deps[0], deps[1], name, None)) + else: + for field, ref in deps.items(): + if isinstance(ref, CollectionRef): + kind = "list" if ref.kind == "positional" else "dict" + members = ( + [(None, member) for member in ref.positional_members] + if ref.kind == "positional" + else list(ref.keyed_members) + ) + for key, member in members: + definition["edges"].append( + _edge( + member.task_name, + member.output_field, + name, + field, + collect=kind, + key=key, + ) + ) + elif isinstance(ref, tuple): + definition["edges"].append(_edge(ref[0], ref[1], name, field)) + else: + definition["edges"].append(_edge(ref, None, name, field)) + + has_dependents.update(edge["from"] for edge in definition["edges"]) + sinks = [node["name"] for node in definition["nodes"] if node["name"] not in has_dependents] + if sinks != [workflow.result_task_name]: + definition["result_task"] = workflow.result_task_name + definition["editable"] = not reasons + if describe: + try: + definition["describe"] = describe_workflow(workflow, schema_generator=schema_generator) + except Exception as exc: + definition["describe_error"] = { + "code": "describe_error", + "message": str(exc), + } + return definition + + +# --- Saving YAML --- + + +class _FlowList(list[Any]): + """A list written in YAML flow style (``[a, b]``).""" + + +class _Dumper(yaml.SafeDumper): + def increase_indent(self, flow: bool = False, indentless: bool = False) -> Any: + return super().increase_indent(flow, False) + + +def _represent_flow_list(dumper: yaml.SafeDumper, data: _FlowList) -> Any: + return dumper.represent_sequence("tag:yaml.org,2002:seq", data, flow_style=True) + + +_Dumper.add_representer(_FlowList, _represent_flow_list) + + +def _dump_yaml(data: Any) -> str: + return yaml.dump( + data, + Dumper=_Dumper, + sort_keys=False, + default_flow_style=False, + allow_unicode=True, + width=100, + ) + + +def _reference(upstream: str, output: str | None) -> Any: + return upstream if output is None else _FlowList([upstream, output]) + + +def _check_saveable(definition: dict[str, Any]) -> None: + if not definition.get("editable", True): + reasons = "; ".join(definition.get("readonly_reasons", [])) + raise DefinitionError(f"The workflow is read-only and cannot be saved: {reasons}") + if not _is_identifier(definition.get("name", "")): + raise DefinitionError(f"Invalid workflow name '{definition.get('name', '')}'") + nodes = definition.get("nodes", []) + if not nodes: + raise DefinitionError("The workflow has no tasks") + names: set[str] = set() + for node in nodes: + name = node.get("name") + if not _is_identifier(name): + raise DefinitionError(f"Invalid task name '{name}'", task=name) + if name in names: + raise DefinitionError(f"Duplicate task name '{name}'", task=name) + if not node.get("task"): + raise DefinitionError(f"Task '{name}' has no task type", task=name) + names.add(name) + for edge in definition.get("edges", []): + for end in ("from", "to"): + if edge.get(end) not in names: + raise DefinitionError(f"Connection refers to unknown task '{edge.get(end)}'") + result_task = definition.get("result_task") + if result_task is not None and result_task not in names: + raise DefinitionError(f"Unknown result task '{result_task}'") + + +def _dependencies(definition: dict[str, Any], node_name: str) -> Any: + incoming = [edge for edge in definition.get("edges", []) if edge["to"] == node_name] + whole = [edge for edge in incoming if edge.get("input") is None] + fields = [edge for edge in incoming if edge.get("input") is not None] + if whole and fields: + raise DefinitionError( + f"Task '{node_name}' cannot take its whole input from one task and " + "individual input fields from others", + task=node_name, + ) + if len(whole) > 1: + raise DefinitionError( + f"Task '{node_name}' has more than one connection to its whole input", + task=node_name, + ) + if whole: + return _reference(whole[0]["from"], whole[0].get("output")) + if not fields: + return None + deps: dict[str, Any] = {} + for edge in fields: + if edge.get("collect"): + _add_collect_member(deps, edge, node_name) + continue + if edge["input"] in deps: + raise DefinitionError( + f"Input '{node_name}.{edge['input']}' has more than one connection", + task=node_name, + field=edge["input"], + ) + deps[edge["input"]] = _reference(edge["from"], edge.get("output")) + return deps + + +def _add_collect_member(deps: dict[str, Any], edge: dict[str, Any], node_name: str) -> None: + field = edge["input"] + kind = edge["collect"] + if kind not in ("list", "dict"): + raise DefinitionError( + f"Unknown collect kind '{kind}' for input '{node_name}.{field}'", + task=node_name, + field=field, + ) + existing = deps.get(field) + if existing is None: + existing = {"collect": [] if kind == "list" else {}} + deps[field] = existing + found = existing.get("collect") if isinstance(existing, dict) else None + if not isinstance(found, list if kind == "list" else dict): + raise DefinitionError( + f"Input '{node_name}.{field}' mixes collected and direct connections, " + "or list and keyed members", + task=node_name, + field=field, + ) + members: Any = found + reference = _reference(edge["from"], edge.get("output")) + if kind == "list": + members.append(reference) + return + key = edge.get("key") + if not isinstance(key, str) or not key: + raise DefinitionError( + f"A keyed member of '{node_name}.{field}' has no key", + task=node_name, + field=field, + ) + if key in members: + raise DefinitionError( + f"Duplicate key '{key}' in '{node_name}.{field}'", + task=node_name, + field=field, + ) + members[key] = reference + + +def _map_document(task_map: Any, node_name: str) -> dict[str, Any]: + if not isinstance(task_map, dict): + raise DefinitionError(f"Invalid map for task '{node_name}'", task=node_name) + document: dict[str, Any] = {} + for key in ("over", "key_as", "value_as"): + value = task_map.get(key) + if not _is_identifier(value): + raise DefinitionError( + f"Invalid map '{key}' for task '{node_name}': {value!r}", task=node_name + ) + document[key] = value + if document["key_as"] == document["value_as"]: + raise DefinitionError( + f"The map of task '{node_name}' uses '{document['key_as']}' for both the key " + "and the value", + task=node_name, + ) + error_mode = task_map.get("error_mode") or "fail_fast" + if error_mode not in ("fail_fast", "collect_all"): + raise DefinitionError( + f"Invalid map error mode '{error_mode}' for task '{node_name}'", + task=node_name, + ) + if error_mode != "fail_fast": + document["error_mode"] = error_mode + return document + + +def workflow_document(definition: dict[str, Any]) -> dict[str, Any]: + """Build the ``workflow.yaml`` document for a definition.""" + _check_saveable(definition) + nodes = definition["nodes"] + has_edges = bool(definition.get("edges")) + tasks: list[dict[str, Any]] = [] + for index, node in enumerate(nodes): + entry: dict[str, Any] = {"task": node["task"], "name": node["name"]} + deps = _dependencies(definition, node["name"]) + if deps is None and index == 0 and len(nodes) > 1 and not has_edges: + # Without any depends_on taskmaestro chains the tasks in file order. + deps = {} + if deps is not None: + entry["depends_on"] = deps + if node.get("config_fields"): + entry["config_fields"] = _FlowList(node["config_fields"]) + if node.get("map"): + entry["map"] = _map_document(node["map"], node["name"]) + tasks.append(entry) + + section: dict[str, Any] = {"name": definition["name"]} + if definition.get("result_task"): + section["result_task"] = definition["result_task"] + section["tasks"] = tasks + document: dict[str, Any] = {"workflow": section} + for key, value in (definition.get("passthrough") or {}).items(): + if key != "workflow": + document[key] = value + return document + + +def input_document( + definition: dict[str, Any], value_filter: ValueFilter | None = None +) -> dict[str, Any]: + """Build the ``input.yaml`` document: one key per task, literal values only. + + ``value_filter`` is applied to each task's values, e.g. to drop values that + must not be persisted. + """ + inputs = definition.get("inputs") or {} + document: dict[str, Any] = {} + for node in definition["nodes"]: + values = inputs.get(node["name"]) or {} + fields = list(node.get("config_fields") or []) + if node.get("map"): + fields.append(node["map"].get("over")) + kept = {field: value for field, value in values.items() if field in fields} + document[node["name"]] = value_filter(kept) if value_filter is not None else kept + return document + + +def _write_atomically(path: Path, text: str) -> None: + fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(text) + os.replace(temp_name, path) + except BaseException: + Path(temp_name).unlink(missing_ok=True) + raise + + +def write_definition( + definition: dict[str, Any], + out_dir: Path, + *, + workflow_only: bool = False, + backup: bool = False, + value_filter: ValueFilter | None = None, +) -> dict[str, Any]: + """Write ``workflow.yaml`` (and ``input.yaml``) for a definition.""" + workflow_text = "\n".join( + _dump_yaml({key: value}) for key, value in workflow_document(definition).items() + ) + comment = (definition.get("header_comment") or "").strip("\n") + if comment: + workflow_text = f"{comment}\n\n{workflow_text}" + input_text = None if workflow_only else _dump_yaml(input_document(definition, value_filter)) + + out_dir.mkdir(parents=True, exist_ok=True) + workflow_path = out_dir / WORKFLOW_FILE + input_path = out_dir / INPUT_FILE + backups: list[str] = [] + if backup: + for path in (workflow_path, input_path): + backup_path = path.with_name(path.name + BACKUP_SUFFIX) + if path.exists() and not backup_path.exists(): + shutil.copy2(path, backup_path) + backups.append(str(backup_path)) + + _write_atomically(workflow_path, workflow_text) + if input_text is not None: + _write_atomically(input_path, input_text) + return { + "status": "ok", + "workflow_yaml": str(workflow_path), + "input_yaml": str(input_path) if input_text is not None else None, + "backups": backups, + } diff --git a/taskmaestro/discovery.py b/taskmaestro/discovery.py index 7546686..3b76368 100644 --- a/taskmaestro/discovery.py +++ b/taskmaestro/discovery.py @@ -2,6 +2,7 @@ from __future__ import annotations +from dataclasses import dataclass from importlib.metadata import EntryPoint, entry_points from typing import Any @@ -94,3 +95,60 @@ def get_registered_workflow(name: str) -> Workflow: if entry_point is None: raise PluginLoadError(f"No workflow entry point named '{name}' is registered") return _load(entry_point, Workflow, "workflow") + + +@dataclass(frozen=True) +class RegistryProblem: + """A registered entry point that could not be loaded.""" + + kind: str # "task" or "workflow" + name: str + entry_point: str + message: str + + +@dataclass(frozen=True) +class Registry: + """Everything that loaded from the entry-point groups, plus what did not.""" + + tasks: dict[str, type[Task[Any, Any]]] + workflows: dict[str, Workflow] + problems: list[RegistryProblem] + + +def load_registry(*, include_workflows: bool = True) -> Registry: + """Load all registered tasks and workflows, collecting failures instead of raising. + + Unlike ``registered_tasks``/``registered_workflows``, one broken or duplicate + entry point does not hide the others. Entry points are visited in name + order and the first of a duplicated name wins. + """ + problems: list[RegistryProblem] = [] + + def load_group[T](group: str, kind: str, expected: type[T]) -> dict[str, T]: + loaded: dict[str, T] = {} + for entry_point in sorted(entry_points(group=group), key=lambda ep: ep.name): + name = entry_point.name + if name in loaded: + problems.append( + RegistryProblem(kind, name, entry_point.value, f"Duplicate {kind} id") + ) + continue + try: + value: Any = entry_point.load() + if kind == "task": + if not (isinstance(value, type) and issubclass(value, Task)): + raise TypeError("The entry point is not a Task subclass") + elif not isinstance(value, expected): + raise TypeError("The entry point is not a Workflow") + except Exception as exc: + problems.append(RegistryProblem(kind, name, entry_point.value, str(exc))) + continue + loaded[name] = value + return loaded + + tasks = load_group(TASK_ENTRY_POINT_GROUP, "task", type) + workflows = ( + load_group(WORKFLOW_ENTRY_POINT_GROUP, "workflow", Workflow) if include_workflows else {} + ) + return Registry(tasks=tasks, workflows=workflows, problems=problems) diff --git a/taskmaestro/exceptions.py b/taskmaestro/exceptions.py index d60bf15..43e72bb 100644 --- a/taskmaestro/exceptions.py +++ b/taskmaestro/exceptions.py @@ -80,3 +80,19 @@ class ConfigLoadError(WorkflowRunnerError): class PluginLoadError(WorkflowRunnerError): """Raised when an installed task or workflow entry point is invalid.""" + + +class DefinitionError(WorkflowRunnerError): + """Raised when a workflow definition cannot be loaded from or saved to YAML. + + ``task`` and ``field`` name the offending task and input field, if known. + """ + + def __init__(self, message: str, *, task: str | None = None, field: str | None = None) -> None: + super().__init__(message) + self.message = message + self.task = task + self.field = field + + def to_json(self) -> dict[str, str | None]: + return {"message": self.message, "task": self.task, "field": self.field} diff --git a/taskmaestro/introspection.py b/taskmaestro/introspection.py new file mode 100644 index 0000000..235db12 --- /dev/null +++ b/taskmaestro/introspection.py @@ -0,0 +1,317 @@ +"""Public helpers for describing workflows, tasks and configuration values as JSON. + +The CLI's ``describe`` commands are built on these functions. Other front ends +(editors, services) can reuse them, and can extend the generated JSON schemas +by subclassing ``TaskSchemaGenerator`` and overriding ``type_annotations``. +""" + +from __future__ import annotations + +import inspect +import json +import math +import sys +from contextlib import redirect_stdout +from dataclasses import asdict +from datetime import date +from pathlib import Path, PurePath +from typing import Any + +from pydantic import BaseModel, TypeAdapter, ValidationError +from pydantic.errors import PydanticInvalidForJsonSchema, PydanticSchemaGenerationError +from pydantic.json_schema import GenerateJsonSchema, JsonSchemaValue +from pydantic_core import core_schema + +from taskmaestro.dependencies import CollectionRef, OutputRef +from taskmaestro.exceptions import ConfigLoadError, WorkflowDefinitionError +from taskmaestro.job import EmptyConfig, Job, JobConfiguration +from taskmaestro.task import get_input_type, get_output_type +from taskmaestro.workflow import Workflow +from taskmaestro.yaml_config import load_workflow_only, workflow_imports + + +def python_type_name(cls: type[Any]) -> str: + """Return the dotted ``module.QualifiedName`` of a class.""" + return f"{cls.__module__}.{cls.__qualname__}" + + +def error_payload( + code: str, exc: Exception, message: str, *, task: str | None = None +) -> dict[str, Any]: + """Return safe diagnostics from an exception and its causes.""" + issues: list[dict[str, str]] = [] + cause: BaseException | None = exc + while cause is not None: + if isinstance(cause, ValidationError): + issues = [ + {"field": ".".join(map(str, issue["loc"])), "code": issue["type"]} + for issue in cause.errors(include_input=False, include_context=False) + ] + break + if isinstance(cause, WorkflowDefinitionError): + task = task or cause.task_name + issues = [{"field": field, "code": "missing"} for field in cause.fields] + if issues: + break + cause = cause.__cause__ + return { + "code": code, + "type": type(exc).__name__, + "message": message, + "task": task, + "field": issues[0]["field"] if issues else None, + "issues": issues, + } + + +class TaskSchemaGenerator(GenerateJsonSchema): + """Describe runtime-only Python objects without pretending they accept JSON. + + Override ``type_annotations`` to add extra keys to the schema of every + opaque Python type and every Pydantic model. + """ + + @classmethod + def type_annotations(cls, python_type: type[Any]) -> dict[str, Any]: + return {} + + def is_instance_schema(self, schema: core_schema.IsInstanceSchema) -> JsonSchemaValue: + python_type = schema["cls"] + return { + "not": {}, # No JSON value can satisfy an isinstance check for this object. + "x-taskmaestro-opaque": True, + "x-taskmaestro-python-type": python_type_name(python_type), + **self.type_annotations(python_type), + } + + def model_schema(self, schema: core_schema.ModelSchema) -> JsonSchemaValue: + json_schema = super().model_schema(schema) + json_schema.update(self.type_annotations(schema["cls"])) + return json_schema + + +def model_json_schema( + model: type[Any], schema_generator: type[GenerateJsonSchema] = TaskSchemaGenerator +) -> dict[str, Any]: + """Return the JSON schema of a Pydantic model or any type Pydantic can adapt.""" + if isinstance(model, type) and issubclass(model, BaseModel): + return model.model_json_schema(schema_generator=schema_generator) + return TypeAdapter(model).json_schema(schema_generator=schema_generator) + + +def _output_ref(ref: OutputRef) -> dict[str, str | None]: + return {"task": ref.task_name, "field": ref.output_field} + + +def dependency_spec(deps: Any) -> Any: + """Render validated dependencies, including field routing and collections.""" + if deps is None: + return None + if isinstance(deps, str): + return {"task": deps, "field": None} + if isinstance(deps, tuple): + return {"task": deps[0], "field": deps[1]} + result: dict[str, Any] = {} + for field, ref in deps.items(): + if isinstance(ref, CollectionRef): + members: list[Any] | dict[str, Any] + if ref.kind == "keyed": + members = {key: _output_ref(item) for key, item in ref.keyed_members} + else: + members = [_output_ref(item) for item in ref.positional_members] + result[field] = {"collect": {"kind": ref.kind, "members": members}} + elif isinstance(ref, tuple): + result[field] = {"task": ref[0], "field": ref[1]} + else: + result[field] = {"task": ref, "field": None} + return result + + +def _jsonable_key(key: object, path: str) -> str: + """Convert a YAML mapping key to the JSON object key it would get.""" + value = jsonable(key, path) + if isinstance(value, str): + return value + if isinstance(value, bool | int | float) or value is None: + return json.dumps(value) + raise TypeError(f"Configuration value '{path}' has an unsupported mapping key type") + + +def jsonable(value: object, path: str) -> Any: + """Convert a raw configuration value (as loaded from YAML) to plain JSON data. + + Dates and datetimes become ISO 8601 strings and paths become strings; + everything else must already be JSON-compatible. ``path`` names the value + in error messages, which deliberately never include the value itself. + """ + if value is None or isinstance(value, bool | int | str): + return value + if isinstance(value, float): + if not math.isfinite(value): + raise ValueError(f"Configuration value '{path}' is not a finite number") + return value + if isinstance(value, date): # Includes datetime. + return value.isoformat() + if isinstance(value, PurePath): + return str(value) + if isinstance(value, dict): + result: dict[str, Any] = {} + for key, item in value.items(): + json_key = _jsonable_key(key, path) + if json_key in result: + raise ValueError( + f"Configuration value '{path}' has keys that collide as JSON key '{json_key}'" + ) + result[json_key] = jsonable(item, f"{path}.{json_key}") + return result + if isinstance(value, list | tuple): + return [jsonable(item, f"{path}[{index}]") for index, item in enumerate(value)] + if isinstance(value, set | frozenset): + items = [jsonable(item, f"{path}[]") for item in value] + try: + return sorted(items) + except TypeError: + # Mixed types: order by canonical JSON text so output is deterministic. + return sorted(items, key=lambda item: json.dumps(item, sort_keys=True)) + raise TypeError(f"Configuration value '{path}' has unsupported type {type(value).__name__}") + + +def missing_config_fields(workflow: Workflow, config: JobConfiguration) -> dict[str, list[str]]: + """Return the declared config fields that the input does not supply, per task.""" + return { + name: sorted(workflow.get_config_fields(name) - config.config_fields_for_task(name)) + for name, _task in workflow.topological_order() + } + + +def _fill_missing( + workflow: Workflow, config: JobConfiguration, missing: dict[str, list[str]] +) -> JobConfiguration: + """Return a copy of ``config`` with placeholder values for missing config fields. + + ``Job`` only checks that these fields are present, so placeholders let it run + its remaining validation. Map sources get no placeholder, so a missing one is + still rejected with its own error. + """ + filled: dict[str, dict[str, Any]] = {} + for name, fields in missing.items(): + task_map = workflow.get_task_map(name) + map_source = task_map.over if task_map is not None else None + placeholders = {field: None for field in fields if field != map_source} + filled[name] = {**placeholders, **config.get_config_for_task(name)} + return JobConfiguration(filled) + + +def describe_yaml_workflow( + workflow_path: Path, + input_path: Path | None = None, + schema_generator: type[GenerateJsonSchema] = TaskSchemaGenerator, +) -> dict[str, Any]: + """Describe a YAML workflow without running any task. + + With ``input_path`` the description also reports provided, missing and + configured values per task. Raises ConfigLoadError if the workflow or input + is invalid; output printed while importing task modules goes to stderr. + """ + try: + with redirect_stdout(sys.stderr), workflow_imports(Path(workflow_path).resolve().parent): + workflow, config = load_workflow_only(Path(workflow_path), input_path) + configured = None + missing = None + values = None + if input_path is not None: + assert config is not None + missing = missing_config_fields(workflow, config) + # Check the rest of the supplied config (map sources, root inputs) + # without constructing hooks or executing any tasks. Missing config + # fields are reported rather than rejected: the input may be a + # template whose remaining fields the caller fills in. + Job( + workflow, + EmptyConfig(), + job_configuration=_fill_missing(workflow, config, missing), + ) + configured = { + name: sorted(config.config_fields_for_task(name)) + for name, _task in workflow.topological_order() + } + # Raw input values, before Pydantic validation, so opaque or + # runtime-only markers pass through unchanged. + values = { + name: { + field: jsonable(value, f"{name}.{field}") + for field, value in sorted(config.get_config_for_task(name).items()) + } + for name, _task in workflow.topological_order() + } + return describe_workflow( + workflow, + configured=configured, + missing=missing, + values=values, + schema_generator=schema_generator, + ) + except WorkflowDefinitionError as exc: + raise ConfigLoadError(f"Job validation failed: {exc}") from exc + except ( + TypeError, + ValueError, + PydanticInvalidForJsonSchema, + PydanticSchemaGenerationError, + ) as exc: + raise ConfigLoadError(f"Cannot describe workflow: {exc}") from exc + + +def task_descriptor( + task_id: str, + task: type[Any], + schema_generator: type[GenerateJsonSchema] = TaskSchemaGenerator, +) -> dict[str, Any]: + """Describe a task class: identity, timeout and input/output JSON schemas.""" + return { + "id": task_id, + "name": getattr(task, "name", task.__name__), + "python_type": python_type_name(task), + "description": inspect.getdoc(task) or "", + "timeout_seconds": getattr(task, "timeout_seconds", None), + "input_schema": model_json_schema(get_input_type(task), schema_generator), + "output_schema": model_json_schema(get_output_type(task), schema_generator), + } + + +def describe_workflow( + workflow: Workflow, + *, + configured: dict[str, list[str]] | None = None, + missing: dict[str, list[str]] | None = None, + values: dict[str, dict[str, Any]] | None = None, + schema_generator: type[GenerateJsonSchema] = TaskSchemaGenerator, +) -> dict[str, Any]: + """Describe a workflow's tasks, dependencies and schemas without running it. + + ``configured``, ``missing`` and ``values`` report per-task input state and + are ``None`` in the output when not supplied. + """ + tasks: list[dict[str, Any]] = [] + for name, task in workflow.topological_order(): + input_type = get_input_type(task) + output_type = workflow.get_output_annotation(name) + task_map = workflow.get_task_map(name) + tasks.append( + { + "name": name, + "python_type": python_type_name(task), + "depends_on": dependency_spec(workflow.get_dependencies(name)), + "config_fields": sorted(workflow.get_config_fields(name)), + "provided_config_fields": configured[name] if configured is not None else None, + "missing_config_fields": missing[name] if missing is not None else None, + "config_values": values[name] if values is not None else None, + "required_input_fields": sorted( + field for field, info in input_type.model_fields.items() if info.is_required() + ), + "map": asdict(task_map) if task_map is not None else None, + "input_schema": model_json_schema(input_type, schema_generator), + "output_schema": model_json_schema(output_type, schema_generator), + } + ) + return {"workflow": workflow.name, "result_task": workflow.result_task_name, "tasks": tasks} diff --git a/taskmaestro/yaml_config.py b/taskmaestro/yaml_config.py index c18d631..f5f5602 100644 --- a/taskmaestro/yaml_config.py +++ b/taskmaestro/yaml_config.py @@ -3,7 +3,10 @@ from __future__ import annotations import importlib +import sys import typing +from collections.abc import Iterator +from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Any @@ -140,11 +143,36 @@ def flatten_mapping(self, node: yaml.nodes.MappingNode) -> None: super().flatten_mapping(node) -def _yaml_load(text: str) -> Any: +def yaml_load(text: str) -> Any: """Safely parse YAML while rejecting duplicate mapping keys.""" return yaml.load(text, Loader=_UniqueKeyLoader) +_yaml_load = yaml_load + + +def resolve_task( + task_id: str, installed_task_names: set[str] | None = None +) -> type[Task[Any, Any]]: + """Resolve the ``task:`` value of a YAML file to a Task class. + + A registered entry-point name wins; otherwise ``task_id`` is imported as a + dotted class path. Raises ConfigLoadError if it cannot be resolved. + """ + if installed_task_names is None: + installed_task_names = registered_task_names() + try: + if task_id in installed_task_names: + cls: type[Task[Any, Any]] = get_registered_task(task_id) + else: + cls = import_class(task_id) + except PluginLoadError as exc: + raise ConfigLoadError(str(exc)) from exc + if not (isinstance(cls, type) and issubclass(cls, Task)): + raise ConfigLoadError(f"'{task_id}' is not a Task subclass") + return cls + + def import_class(dotted_path: str) -> type[Any]: """Import a class from a dotted path like 'pkg.mod.ClassName'. @@ -316,15 +344,7 @@ def _load_workflow_only( key = task_config.workflow else: assert task_config.task is not None - try: - if task_config.task in installed_task_names: - cls = get_registered_task(task_config.task) - else: - cls = import_class(task_config.task) - except PluginLoadError as exc: - raise ConfigLoadError(str(exc)) from exc - if not (isinstance(cls, type) and issubclass(cls, Task)): - raise ConfigLoadError(f"'{task_config.task}' is not a Task subclass") + cls = resolve_task(task_config.task, installed_task_names) key = task_config.task registered_name = task_config.name if task_config.name else cls.name entries.append(_Entry(task_config, cls, key, registered_name)) @@ -624,3 +644,17 @@ def run_workflow_from_yaml(workflow_path: str | Path, input_path: str | Path) -> """Load and run a workflow from a workflow YAML file and an input YAML file in one step.""" loaded = load_workflow_from_yaml(workflow_path, input_path) return loaded.run() + + +load_workflow_only = _load_workflow_only + + +@contextmanager +def workflow_imports(directory: str | Path) -> Iterator[None]: + """Make task modules in ``directory`` importable for the duration of loading.""" + original_path = sys.path.copy() + sys.path.insert(0, str(Path(directory).resolve())) + try: + yield + finally: + sys.path[:] = original_path diff --git a/tests/test_definition.py b/tests/test_definition.py new file mode 100644 index 0000000..72b6029 --- /dev/null +++ b/tests/test_definition.py @@ -0,0 +1,617 @@ +"""Tests for editable workflow definitions and the public introspection helpers.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from taskmaestro import ( + ConfigLoadError, + ExecutionContext, + JobStatus, + Task, + TaskMap, + Workflow, + collect, + load_workflow_from_yaml, + workflow_task, +) +from taskmaestro.definition import ( + definition_from_workflow, + describe_yaml, + load_definition, + write_definition, +) +from taskmaestro.exceptions import DefinitionError +from taskmaestro.introspection import ( + TaskSchemaGenerator, + describe_yaml_workflow, + error_payload, + jsonable, + model_json_schema, +) +from taskmaestro.yaml_config import load_workflow_only, resolve_task +from tests.conftest import AddOne, Double, FanInTask, NumberInput, NumberOutput + +WORKFLOW = """\ +# A pipeline. +workflow: + name: pipeline + tasks: + - task: tests.conftest.AddOne + name: first + config_fields: [value] + - task: tests.conftest.Double + name: second + depends_on: first +""" + + +def _write(directory: Path, workflow: str = WORKFLOW, inputs: str | None = None) -> Path: + directory.mkdir(parents=True, exist_ok=True) + (directory / "workflow.yaml").write_text(workflow) + if inputs is not None: + (directory / "input.yaml").write_text(inputs) + return directory + + +class OpaqueAnnotator(TaskSchemaGenerator): + @classmethod + def type_annotations(cls, python_type: type[Any]) -> dict[str, Any]: + return {"x-test-type": python_type.__name__} + + +def test_load_definition_builds_nodes_edges_and_inputs(tmp_path: Path) -> None: + folder = _write(tmp_path / "wf", inputs="first: {value: 3}\n") + + definition = load_definition(folder) + + assert [node["name"] for node in definition["nodes"]] == ["first", "second"] + assert definition["edges"] == [ + {"from": "first", "output": None, "to": "second", "input": None} + ] + assert definition["inputs"] == {"first": {"value": 3}, "second": {}} + assert definition["header_comment"] == "# A pipeline." + assert definition["editable"] is True + + +def test_written_definition_loads_and_runs_like_the_original(tmp_path: Path) -> None: + source = _write(tmp_path / "src", inputs="first: {value: 3}\n") + definition = load_definition(source) + + result = write_definition(definition, tmp_path / "out") + + assert result["backups"] == [] + reloaded = load_definition(tmp_path / "out") + for key in ("nodes", "edges", "inputs", "result_task"): + assert reloaded[key] == definition[key] + loaded = load_workflow_from_yaml(result["workflow_yaml"], result["input_yaml"]) + assert loaded.run().status == JobStatus.COMPLETED + + +def test_value_filter_is_applied_to_saved_inputs(tmp_path: Path) -> None: + definition = load_definition(_write(tmp_path / "src", inputs="first: {value: 3}\n")) + + write_definition( + definition, + tmp_path / "out", + value_filter=lambda values: {k: v for k, v in values.items() if k != "value"}, + ) + + assert load_definition(tmp_path / "out")["inputs"]["first"] == {} + + +def test_backup_keeps_the_first_version_of_existing_files(tmp_path: Path) -> None: + folder = _write(tmp_path / "wf", inputs="first: {value: 3}\n") + definition = load_definition(folder) + + result = write_definition(definition, folder, backup=True) + + assert sorted(Path(path).name for path in result["backups"]) == [ + "input.yaml.bak", + "workflow.yaml.bak", + ] + + +def test_schema_generator_annotates_task_types_and_description(tmp_path: Path) -> None: + folder = _write(tmp_path / "wf", inputs="first: {value: 3}\n") + + definition = load_definition(folder, describe=True, schema_generator=OpaqueAnnotator) + + descriptor = definition["task_types"]["tests.conftest.AddOne"] + assert descriptor["input_schema"]["x-test-type"] == "NumberInput" + first = definition["describe"]["tasks"][0] + assert first["output_schema"]["x-test-type"] == "NumberOutput" + + +def test_unresolvable_task_makes_definition_read_only(tmp_path: Path) -> None: + workflow = WORKFLOW.replace("tests.conftest.Double", "tests.conftest.Missing") + definition = load_definition(_write(tmp_path / "wf", workflow)) + + assert definition["editable"] is False + with pytest.raises(DefinitionError, match="read-only"): + write_definition(definition, tmp_path / "out") + + +def test_invalid_yaml_raises_definition_error(tmp_path: Path) -> None: + with pytest.raises(DefinitionError, match="parse error"): + load_definition(_write(tmp_path / "wf", "workflow: [\n")) + + +def test_definition_from_registered_style_workflow(tmp_path: Path) -> None: + folder = _write(tmp_path / "wf", inputs="first: {value: 3}\n") + workflow, _config = load_workflow_only(folder / "workflow.yaml") + + definition = definition_from_workflow(workflow, None, {AddOne: "add_one"}) + + first, second = definition["nodes"] + assert first["task"] == "add_one" + assert second["task"] == "tests.conftest.Double" + assert definition["edges"] == [ + {"from": "first", "output": None, "to": "second", "input": None} + ] + + +def test_describe_yaml_reports_errors_instead_of_raising(tmp_path: Path) -> None: + folder = _write(tmp_path / "wf", inputs="first: {value: 3}\n") + + description, error = describe_yaml(folder / "workflow.yaml", folder / "input.yaml") + assert error is None and description is not None + assert description["tasks"][0]["provided_config_fields"] == ["value"] + + description, error = describe_yaml(tmp_path / "missing.yaml", None) + assert description is None + assert error is not None and error["code"] == "configuration_error" + + +def test_describe_yaml_workflow_raises_config_load_error(tmp_path: Path) -> None: + with pytest.raises(ConfigLoadError): + describe_yaml_workflow(tmp_path / "missing.yaml") + + +def test_resolve_task_accepts_class_paths_and_rejects_non_tasks() -> None: + assert resolve_task("tests.conftest.Double") is Double + with pytest.raises(ConfigLoadError, match="not a Task subclass"): + resolve_task("tests.conftest.NumberInput") + + +def test_public_json_helpers() -> None: + assert jsonable({"path": Path("/tmp/x")}, "task") == {"path": "/tmp/x"} + payload = error_payload("code", ValueError("boom"), "message", task="t") + assert payload["task"] == "t" and payload["issues"] == [] + + +def test_model_json_schema_adapts_non_model_types() -> None: + assert model_json_schema(list[int]) == {"items": {"type": "integer"}, "type": "array"} + + +def test_definition_error_to_json() -> None: + error = DefinitionError("bad", task="t", field="f") + assert error.to_json() == {"message": "bad", "task": "t", "field": "f"} + + +# --- Loading YAML: graph features --- + +COLLECTIONS = "tests.test_collections" + +COLLECT_WORKFLOW = f"""\ +# Collections. + +workflow: + name: collections + result_task: listed + tasks: + - task: {COLLECTIONS}.ProduceSurface + name: first + - task: {COLLECTIONS}.ProduceEnvelope + name: second + - task: {COLLECTIONS}.CollectSurfaceList + name: listed + depends_on: + surfaces: + collect: [first, [second, surface]] + - task: {COLLECTIONS}.CollectSurfaceDict + name: keyed + depends_on: + surfaces: + collect: {{top: first, base: [second, surface]}} +extra: + note: kept +""" + + +def test_collections_result_task_and_passthrough_round_trip(tmp_path: Path) -> None: + inputs = "first: {value: 1}\nsecond: {value: 2}\nunknown: {value: 3}\n" + definition = load_definition(_write(tmp_path / "src", COLLECT_WORKFLOW, inputs)) + + assert definition["result_task"] == "listed" + assert definition["passthrough"] == {"extra": {"note": "kept"}} + assert "Dropped input values for unknown task 'unknown'" in definition["warnings"] + assert definition["edges"] == [ + {"from": "first", "output": None, "to": "listed", "input": "surfaces", + "collect": "list", "key": None}, + {"from": "second", "output": "surface", "to": "listed", "input": "surfaces", + "collect": "list", "key": None}, + {"from": "first", "output": None, "to": "keyed", "input": "surfaces", + "collect": "dict", "key": "top"}, + {"from": "second", "output": "surface", "to": "keyed", "input": "surfaces", + "collect": "dict", "key": "base"}, + ] # fmt: skip + + result = write_definition(definition, tmp_path / "out") + + reloaded = load_definition(tmp_path / "out") + for key in ("nodes", "edges", "inputs", "result_task", "passthrough", "header_comment"): + assert reloaded[key] == definition[key] + loaded = load_workflow_from_yaml(result["workflow_yaml"], result["input_yaml"]) + assert loaded.run().status == JobStatus.COMPLETED + + +MAP_WORKFLOW = """\ +workflow: + name: mapped + tasks: + - task: tests.conftest.AddOne + - task: tests.test_mapping.MappedNumber + depends_on: {base: tests.conftest.AddOne} + config_fields: [multiplier] + map: {over: items, key_as: item_name, value_as: amount, error_mode: collect_all} + - task: tests.test_mapping.SumAggregate + depends_on: {values: [tests.test_mapping.MappedNumber, root]} +""" + + +def test_mapped_task_round_trip(tmp_path: Path) -> None: + inputs = ( + "add_one: {value: 1}\n" + "mapped_number: {multiplier: 2, items: {a: 1, b: 2}, stray: 5}\n" + "sum_aggregate:\n" + ) + definition = load_definition(_write(tmp_path / "src", MAP_WORKFLOW, inputs)) + + mapped = definition["nodes"][1] + assert mapped["map"] == { + "over": "items", + "key_as": "item_name", + "value_as": "amount", + "error_mode": "collect_all", + } + assert definition["inputs"]["mapped_number"] == {"multiplier": 2, "items": {"a": 1, "b": 2}} + assert definition["inputs"]["sum_aggregate"] == {} + assert any("mapped_number.stray" in warning for warning in definition["warnings"]) + + result = write_definition(definition, tmp_path / "out") + + assert "error_mode: collect_all" in Path(result["workflow_yaml"]).read_text() + assert load_definition(tmp_path / "out")["inputs"] == definition["inputs"] + loaded = load_workflow_from_yaml(result["workflow_yaml"], result["input_yaml"]) + assert loaded.run().status == JobStatus.COMPLETED + + +LINEAR_WORKFLOW = """\ +workflow: + name: linear + tasks: + - task: tests.conftest.AddOne + - task: tests.conftest.Double + name: twice + - task: tests.conftest.Double + name: again +""" + + +def test_linear_workflow_is_chained_and_saved_explicitly(tmp_path: Path) -> None: + definition = load_definition(_write(tmp_path / "src", LINEAR_WORKFLOW)) + + assert [(edge["from"], edge["to"]) for edge in definition["edges"]] == [ + ("add_one", "twice"), + ("twice", "again"), + ] + assert any("chained in file order" in warning for warning in definition["warnings"]) + assert list(definition["task_types"]) == ["tests.conftest.AddOne", "tests.conftest.Double"] + + definition["edges"] = [] + result = write_definition(definition, tmp_path / "out", workflow_only=True) + + assert result["input_yaml"] is None + assert "depends_on: {}" in Path(result["workflow_yaml"]).read_text() + + +def test_wired_config_fields_are_dropped(tmp_path: Path) -> None: + workflow = """\ +workflow: + name: wired + tasks: + - task: tests.conftest.AddOne + name: first + - task: tests.conftest.Double + name: second + config_fields: [value] + depends_on: {value: [first, value]} +""" + definition = load_definition(_write(tmp_path / "wf", workflow, "second: {value: 5}\n")) + + assert definition["nodes"][1]["config_fields"] == [] + assert definition["inputs"]["second"] == {} + assert any("connected to an upstream task" in w for w in definition["warnings"]) + + +def test_nested_workflows_make_definition_read_only(tmp_path: Path) -> None: + folder = tmp_path / "wf" + _write(folder, "workflow:\n name: outer\n tasks:\n" + " - workflow: inner.yaml\n - workflow: missing.yaml\n") # fmt: skip + (folder / "inner.yaml").write_text(WORKFLOW) + + definition = load_definition(folder) + + assert [node["name"] for node in definition["nodes"]] == ["pipeline", "missing"] + assert definition["nodes"][0]["workflow"] == "inner.yaml" + assert definition["editable"] is False + # Values of read-only definitions are kept even when they are not config fields. + assert definition["inputs"] == {"pipeline": {}, "missing": {}} + + +def test_undescribable_task_makes_definition_read_only( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + def fail(*args: Any) -> dict[str, Any]: + raise ValueError("no schema") + + monkeypatch.setattr("taskmaestro.definition.task_descriptor", fail) + + definition = load_definition(_write(tmp_path / "wf")) + + assert ( + "Cannot describe task 'tests.conftest.AddOne': no schema" + in (definition["readonly_reasons"]) + ) + + +def _two_add_ones(depends_on: str) -> str: + return f"""\ +workflow: + name: refs + tasks: + - task: tests.conftest.AddOne + name: a + - task: tests.conftest.AddOne + name: b + - task: tests.conftest.Double + depends_on: {depends_on} +""" + + +@pytest.mark.parametrize( + ("workflow", "inputs", "match"), + [ + ("workflow: [\n", None, "YAML parse error"), + ("- a\n", None, "mapping at top level"), + ("workflow: {}\n", None, "schema validation error"), + (WORKFLOW + "extra: .nan\n", None, "not a finite number"), + (WORKFLOW, "- a\n", "Input YAML file must contain a mapping"), + (WORKFLOW, "first: 3\n", "Input value for task 'first' must be a mapping"), + (WORKFLOW, "first: [\n", "Input YAML parse error"), + (_two_add_ones("missing"), None, "Dependency 'missing' for task 'double' not found"), + (_two_add_ones("tests.conftest.AddOne"), None, "is ambiguous"), + (_two_add_ones("[a, b, c]"), None, "Invalid dependency \\['a', 'b', 'c'\\] on"), + (_two_add_ones("{value: [a, b, c]}"), None, "for field 'value'"), + (_two_add_ones("{value: {collect: a}}"), None, "must be a list or mapping"), + ( + LINEAR_WORKFLOW.replace(" name: again\n", "").replace(" name: twice\n", ""), + None, + "Duplicate task name", + ), + (WORKFLOW.replace("tasks:", "result_task: missing\n tasks:"), None, "not found"), + ], +) +def test_load_definition_errors( + tmp_path: Path, workflow: str, inputs: str | None, match: str +) -> None: + with pytest.raises(DefinitionError, match=match): + load_definition(_write(tmp_path / "wf", workflow, inputs)) + + +def test_load_definition_reports_unreadable_folder(tmp_path: Path) -> None: + with pytest.raises(DefinitionError, match="Cannot read"): + load_definition(tmp_path / "missing") + + +# --- Registered workflows --- + + +def test_definition_from_builder_workflow_with_collections_and_map() -> None: + from tests import test_collections as coll + from tests import test_mapping as mapping + + workflow = ( + Workflow.builder("built", result_task="first") + .add_task(coll.ProduceSurface, name="first") + .add_task(coll.ProduceEnvelope, name="second") + .add_task( + coll.CollectSurfaceList, + name="listed", + depends_on={"surfaces": collect("first", ("second", "surface"))}, + ) + .add_task( + coll.CollectSurfaceDict, + name="keyed", + depends_on={"surfaces": collect(top="first", base=("second", "surface"))}, + ) + .add_task(AddOne, name="seed") + .add_task(mapping.ProduceEnvelope, name="envelope") + .add_task(Double, name="whole", depends_on=("envelope", "number")) + .add_task( + mapping.MappedNumber, + depends_on={"base": "seed"}, + config_fields=["multiplier"], + mapped_over=TaskMap(over="items", key_as="item_name", value_as="amount"), + ) + .add_task(mapping.SumAggregate, depends_on={"values": ("mapped_number", "root")}) + .add_task(FanInTask, name="fan_in", depends_on={"a": ("envelope", "number"), "b": "whole"}) + .build() + ) + + definition = definition_from_workflow( + workflow, "registered", {}, describe=True, schema_generator=OpaqueAnnotator + ) + + assert definition["editable"] is True + assert definition["source"]["registered_id"] == "registered" + assert definition["result_task"] == "first" + edges = {(e["from"], e["to"], e["input"], e.get("key")) for e in definition["edges"]} + assert ("second", "keyed", "surfaces", "base") in edges + assert ("first", "listed", "surfaces", None) in edges + assert ("envelope", "whole", None, None) in edges + assert ("envelope", "fan_in", "a", None) in edges + assert ("whole", "fan_in", "b", None) in edges + mapped = next(n for n in definition["nodes"] if n["name"] == "mapped_number") + assert mapped["map"]["over"] == "items" + assert definition["describe"]["tasks"][0]["input_schema"]["x-test-type"] == "NumberInput" + + +def test_definition_from_workflow_with_unreferenceable_tasks( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class LocalTask(Task[NumberInput, NumberOutput]): + name = "local" + + def run(self, input: NumberInput, ctx: ExecutionContext) -> NumberOutput: + return NumberOutput(value=input.value) + + Renamed = type("Renamed", (LocalTask,), {"name": "renamed", "__qualname__": "Renamed"}) + Renamed.__module__ = "tests.no_such_module" + Shadowed = type("AddOne", (LocalTask,), {"name": "shadowed", "__qualname__": "AddOne"}) + Shadowed.__module__ = "tests.conftest" + inner = Workflow.builder("inner").add_task(AddOne).build() + + workflow = ( + Workflow.builder("outer", result_task="nested") + .add_task(LocalTask) + .add_task(Renamed) + .add_task(Shadowed) + .add_task(workflow_task(inner, name="nested")) + .build() + ) + + def fail(*args: Any, **kwargs: Any) -> dict[str, Any]: + raise ValueError("cannot describe") + + monkeypatch.setattr("taskmaestro.definition.describe_workflow", fail) + + definition = definition_from_workflow(workflow, None, {}, describe=True) + + assert definition["editable"] is False + reasons = definition["readonly_reasons"] + assert sum("cannot be referenced from YAML" in reason for reason in reasons) == 3 + assert "Task 'nested' is a nested workflow" in reasons + assert definition["describe"] is None + assert definition["describe_error"] == {"code": "describe_error", "message": "cannot describe"} + + +# --- Saving: validation --- + + +def _saveable(tmp_path: Path) -> dict[str, Any]: + return load_definition(_write(tmp_path / "src", COLLECT_WORKFLOW)) + + +def _edit(definition: dict[str, Any], change: str) -> None: + nodes = definition["nodes"] + edges = definition["edges"] + if change == "name": + definition["name"] = "not valid" + elif change == "no_nodes": + definition["nodes"] = [] + elif change == "node_name": + nodes[0]["name"] = "1st" + elif change == "duplicate_node": + nodes[1]["name"] = "first" + elif change == "no_task": + nodes[0]["task"] = None + elif change == "unknown_edge": + edges[0]["from"] = "ghost" + elif change == "result_task": + definition["result_task"] = "ghost" + elif change == "mixed_whole_and_fields": + edges.append({"from": "second", "output": None, "to": "listed", "input": None}) + elif change == "two_whole": + edges[:] = [ + {"from": "first", "output": None, "to": "second", "input": None}, + {"from": "listed", "output": None, "to": "second", "input": None}, + ] + elif change == "double_field": + edges[:] = [ + {"from": "first", "output": None, "to": "second", "input": "value"}, + {"from": "listed", "output": None, "to": "second", "input": "value"}, + ] + elif change == "collect_kind": + edges[0]["collect"] = "set" + elif change == "collect_mixed": + edges[1]["collect"] = "dict" + edges[1]["key"] = "k" + elif change == "collect_and_direct": + edges.insert(0, {"from": "second", "output": None, "to": "listed", "input": "surfaces"}) + elif change == "collect_no_key": + edges[2]["key"] = "" + elif change == "collect_duplicate_key": + edges[3]["key"] = "top" + else: + nodes[0]["map"] = { + "map_type": "oops", + "map_name": {"over": "1", "key_as": "k", "value_as": "v"}, + "map_same": {"over": "i", "key_as": "k", "value_as": "k"}, + "map_mode": {"over": "i", "key_as": "k", "value_as": "v", "error_mode": "x"}, + }[change] + + +@pytest.mark.parametrize( + ("change", "match"), + [ + ("name", "Invalid workflow name"), + ("no_nodes", "no tasks"), + ("node_name", "Invalid task name '1st'"), + ("duplicate_node", "Duplicate task name 'first'"), + ("no_task", "has no task type"), + ("unknown_edge", "unknown task 'ghost'"), + ("result_task", "Unknown result task 'ghost'"), + ("mixed_whole_and_fields", "whole input"), + ("two_whole", "more than one connection to its whole input"), + ("double_field", "'second.value' has more than one connection"), + ("collect_kind", "Unknown collect kind 'set'"), + ("collect_mixed", "mixes collected and direct"), + ("collect_and_direct", "mixes collected and direct"), + ("collect_no_key", "has no key"), + ("collect_duplicate_key", "Duplicate key 'top'"), + ("map_type", "Invalid map for task"), + ("map_name", "Invalid map 'over'"), + ("map_same", "for both the key and the value"), + ("map_mode", "Invalid map error mode 'x'"), + ], +) +def test_write_definition_rejects_invalid_definitions( + tmp_path: Path, change: str, match: str +) -> None: + definition = _saveable(tmp_path) + _edit(definition, change) + + with pytest.raises(DefinitionError, match=match): + write_definition(definition, tmp_path / "out") + + +def test_failed_write_keeps_the_original_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + folder = _write(tmp_path / "wf") + definition = load_definition(folder) + + def fail(*args: Any) -> None: + raise OSError("disk full") + + monkeypatch.setattr("taskmaestro.definition.os.replace", fail) + + with pytest.raises(OSError, match="disk full"): + write_definition(definition, folder) + assert (folder / "workflow.yaml").read_text() == WORKFLOW + assert sorted(path.name for path in folder.iterdir()) == ["workflow.yaml"] diff --git a/tests/test_discovery.py b/tests/test_discovery.py index a0f4f9f..67f0c20 100644 --- a/tests/test_discovery.py +++ b/tests/test_discovery.py @@ -23,6 +23,7 @@ registered_workflow_names, registered_workflows, ) +from taskmaestro.discovery import load_registry class Input(BaseModel): @@ -107,6 +108,52 @@ def test_rejects_duplicate_names(monkeypatch: pytest.MonkeyPatch) -> None: registered_tasks() +def test_load_registry_collects_problems_and_keeps_good_entries( + monkeypatch: pytest.MonkeyPatch, +) -> None: + entries: list[Any] = [ + _entry_point("good", "tests.test_discovery:ExampleTask", "taskmaestro.tasks"), + _entry_point("good", "tests.test_discovery:ExampleTask", "taskmaestro.tasks"), + _entry_point("broken", "tests.test_discovery:missing", "taskmaestro.tasks"), + ] + monkeypatch.setattr( + "taskmaestro.discovery.entry_points", + lambda *, group: entries if group == "taskmaestro.tasks" else [], + ) + + registry = load_registry() + + assert set(registry.tasks) == {"good"} + assert registry.workflows == {} + assert [(p.kind, p.name) for p in registry.problems] == [("task", "broken"), ("task", "good")] + assert registry.problems[1].message == "Duplicate task id" + + +def test_load_registry_checks_entry_point_types(monkeypatch: pytest.MonkeyPatch) -> None: + entries: dict[str, list[Any]] = { + "taskmaestro.tasks": [ + _entry_point("not_a_task", "tests.test_discovery:Input", "taskmaestro.tasks") + ], + "taskmaestro.workflows": [ + _entry_point("good", "tests.test_discovery:example_workflow", "taskmaestro.workflows"), + _entry_point("wrong", "tests.test_discovery:ExampleTask", "taskmaestro.workflows"), + ], + } + monkeypatch.setattr( + "taskmaestro.discovery.entry_points", lambda *, group: entries.get(group, []) + ) + + registry = load_registry() + + assert registry.tasks == {} + assert registry.workflows == {"good": example_workflow} + assert [(p.kind, p.name, p.message) for p in registry.problems] == [ + ("task", "not_a_task", "The entry point is not a Task subclass"), + ("workflow", "wrong", "The entry point is not a Workflow"), + ] + assert load_registry(include_workflows=False).workflows == {} + + 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]) From 50016ea49cd1ecf773b1a654391405e35b7a99ab Mon Sep 17 00:00:00 2001 From: Kristian Bendiksen Date: Wed, 21 Oct 2026 14:23:43 +0200 Subject: [PATCH 2/4] Support registered workflows in the CLI Accept a registered workflow name wherever a workflow YAML path is accepted (run, validate, graph, workflow describe), add 'workflow list', and report the workflow 'source' in 'workflow describe' output. New public API: load_registered_workflow, load_input_configuration, registered_workflow, describe_registered_workflow and WorkflowLookupError, whose message lists the registered workflows. Closes #14 --- AGENTS.md | 6 +- README.md | 29 +++++- taskmaestro/__init__.py | 4 + taskmaestro/cli.py | 61 +++++++++--- taskmaestro/exceptions.py | 7 ++ taskmaestro/introspection.py | 149 ++++++++++++++++++---------- taskmaestro/yaml_config.py | 114 ++++++++++++++++------ tests/test_cli.py | 181 ++++++++++++++++++++++++++++++++++- 8 files changed, 453 insertions(+), 98 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index ac45350..1b0b407 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -25,9 +25,9 @@ mypy taskmaestro # type check (strict mode) | `taskmaestro/workflow.py` | `Workflow` (linear + DAG), `WorkflowBuilder`, validation (cycles, types, fan-in) | | `taskmaestro/dependencies.py`, `taskmaestro/mapping.py` | Task handles, output references, `collect()`, mapped-task configuration | | `taskmaestro/workflow_task.py` | Nested workflows wrapped as tasks | -| `taskmaestro/yaml_config.py` | YAML parsing, task imports, workflow and input validation | -| `taskmaestro/cli.py`, `taskmaestro/discovery.py` | CLI (`run`, `validate`, `graph`, `tasks list/describe`, `workflow describe`), plugin entry-point discovery (`load_registry` collects broken entry points instead of raising) | -| `taskmaestro/introspection.py` | Public JSON helpers: `describe_workflow`, `describe_yaml_workflow`, `task_descriptor`, `TaskSchemaGenerator` (subclass and override `type_annotations` to add schema keys), `jsonable`, `error_payload` | +| `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) | +| `taskmaestro/introspection.py` | Public JSON helpers: `describe_workflow`, `describe_yaml_workflow`, `describe_registered_workflow`, `task_descriptor`, `TaskSchemaGenerator` (subclass and override `type_annotations` to add schema keys), `jsonable`, `error_payload` | | `taskmaestro/definition.py` | Editable graph form of workflow YAML: `load_definition`, `definition_from_workflow`, `write_definition` (atomic, optional `.bak`, `value_filter`); raises `DefinitionError` | | `taskmaestro/job.py` | `Job[C]`, `JobStatus`, `TaskStatus`, `TaskResult` dataclass | | `taskmaestro/runner.py` | `Runner` — topological execution, timeout via `signal.alarm`, hook dispatch | diff --git a/README.md b/README.md index aa59723..99f2521 100644 --- a/README.md +++ b/README.md @@ -627,6 +627,7 @@ WorkflowRunnerError (base) │ └── IncompleteInputError # Missing fan-in field mappings ├── JobStateError # e.g., re-running a completed job ├── ConfigLoadError # YAML config loading failure +│ └── WorkflowLookupError # Unknown or broken registered workflow name └── TaskExecutionError # Runtime task failure ├── MappedTaskExecutionError # One or more mapped items failed ├── TaskOutputTypeError # Output type mismatch @@ -635,15 +636,35 @@ WorkflowRunnerError (base) ## Command-Line Interface -Installed packages provide a `taskmaestro` command for YAML workflows: +Installed packages provide a `taskmaestro` command for YAML and registered workflows: ```bash taskmaestro validate workflow.yaml --input input.yaml taskmaestro graph workflow.yaml --input input.yaml taskmaestro run workflow.yaml --input input.yaml --log-level INFO taskmaestro workflow describe workflow.yaml --json +taskmaestro workflow list --json ``` +Wherever a workflow path is accepted, you can also give the name of a workflow +registered in the `taskmaestro.workflows` entry-point group (see +[Plugin discovery](#plugin-discovery)): + +```bash +taskmaestro workflow describe resinsight.completions --input input.yaml --json +taskmaestro run resinsight.completions --input input.yaml --json +``` + +The argument is a YAML path if it ends in `.yaml`/`.yml` or names an existing +file; otherwise it is looked up as a registered name. An unknown or broken name +is a configuration error whose message lists the registered workflows. Input +files work the same for both kinds, but registered workflows always run with a +default `Runner` and `ExecutionContext`; their root tasks take input values only +for declared `config_fields`. `workflow list` prints the registered names (one +per line, or `{"workflows": [...]}` with `--json`) without importing the +plugins, so one broken plugin does not break the list. In Python, use +`load_registered_workflow(name, input_path)`. + By default, `run` prints the final output as JSON and reports errors on stderr; `validate` prints a human-readable confirmation. `graph` prints Mermaid markup. @@ -669,7 +690,9 @@ failures have status `failed` and `code: "task_failed"`. If a completed result cannot be encoded as JSON (e.g. a Python-only object), `run --json` returns `code: "serialization_error"`. Missing configuration fields produce `issues` with code `missing`; when no structured field metadata is available, `field` -is `null`. Use text mode when you need the original exception message. +is `null`. Use text mode when you need the original exception message. The +exception is an unknown or broken registered workflow name: its `message` +names the registered workflows. Exit codes: `0` success, `1` task or result-serialization failure, `2` workflow configuration failure. In JSON mode logs and ordinary Python `print()` output @@ -686,6 +709,8 @@ instance includes its `name`, `python_type`, `depends_on` references (`task` / and Pydantic input/output schemas. Collection dependencies include their kind (`positional` or `keyed`) and member references. `python_type` identifies the loaded class; it may differ from the plugin entry-point identifier used in YAML. +`source` tells where the workflow came from: `{"kind": "yaml", "path": "/abs/workflow.yaml"}` +or `{"kind": "entry_point", "name": "resinsight.completions"}`. Pass `--input input.yaml` to check the input against the workflow. The input may be incomplete (for example a template whose runtime-picker fields a UI diff --git a/taskmaestro/__init__.py b/taskmaestro/__init__.py index d15923b..305d178 100644 --- a/taskmaestro/__init__.py +++ b/taskmaestro/__init__.py @@ -26,6 +26,7 @@ TaskOutputTypeError, TaskTimeoutError, WorkflowDefinitionError, + WorkflowLookupError, WorkflowRunnerError, WorkflowTaskError, ) @@ -39,6 +40,7 @@ from taskmaestro.workflow_task import workflow_task from taskmaestro.yaml_config import ( LoadedWorkflow, + load_registered_workflow, load_workflow_from_yaml, run_workflow_from_yaml, ) @@ -75,11 +77,13 @@ "Workflow", "WorkflowBuilder", "WorkflowDefinitionError", + "WorkflowLookupError", "WorkflowRunnerError", "WorkflowTaskError", "collect", "get_registered_task", "get_registered_workflow", + "load_registered_workflow", "load_workflow_from_yaml", "registered_task_names", "registered_tasks", diff --git a/taskmaestro/cli.py b/taskmaestro/cli.py index fd3e493..a740af0 100644 --- a/taskmaestro/cli.py +++ b/taskmaestro/cli.py @@ -13,11 +13,16 @@ from pydantic.errors import PydanticInvalidForJsonSchema, PydanticSchemaGenerationError -from taskmaestro.discovery import get_registered_task, registered_task_names -from taskmaestro.exceptions import ConfigLoadError, PluginLoadError +from taskmaestro.discovery import ( + get_registered_task, + registered_task_names, + registered_workflow_names, +) +from taskmaestro.exceptions import ConfigLoadError, PluginLoadError, WorkflowLookupError from taskmaestro.introspection import ( TaskSchemaGenerator, dependency_spec, + describe_registered_workflow, describe_yaml_workflow, error_payload, jsonable, @@ -26,16 +31,24 @@ from taskmaestro.task import get_input_type, get_output_type from taskmaestro.yaml_config import ( LoadedWorkflow, + load_registered_workflow, load_workflow_from_yaml, workflow_imports, ) +WORKFLOW_ARGUMENT_HELP = "Workflow YAML file, or the name of a registered workflow" + def _add_workflow_arguments(parser: argparse.ArgumentParser) -> None: - parser.add_argument("workflow", help="Path to the workflow YAML file") + parser.add_argument("workflow", help=WORKFLOW_ARGUMENT_HELP) parser.add_argument("--input", required=True, help="Path to the input YAML file") +def _is_workflow_path(workflow: str) -> bool: + """Return True if a workflow argument is a YAML path rather than a registered name.""" + return Path(workflow).suffix.lower() in (".yaml", ".yml") or Path(workflow).is_file() + + @contextmanager def _workflow_imports(path: str) -> Iterator[None]: """Make task modules next to a YAML file importable for the duration of loading.""" @@ -44,6 +57,8 @@ def _workflow_imports(path: str) -> Iterator[None]: def _load(args: argparse.Namespace) -> LoadedWorkflow: + if not _is_workflow_path(args.workflow): + return load_registered_workflow(args.workflow, args.input) with _workflow_imports(args.workflow): return load_workflow_from_yaml(args.workflow, args.input) @@ -149,10 +164,21 @@ def _tasks_list(args: argparse.Namespace) -> int: return 0 +def _workflow_list(args: argparse.Namespace) -> int: + names = sorted(registered_workflow_names()) + if args.json: + print(json.dumps({"workflows": names})) + else: + print("\n".join(names)) + return 0 + + def _workflow_describe(args: argparse.Namespace) -> int: - description = describe_yaml_workflow( - Path(args.workflow), Path(args.input) if args.input else None - ) + input_path = Path(args.input) if args.input else None + if _is_workflow_path(args.workflow): + description = describe_yaml_workflow(Path(args.workflow), input_path) + else: + description = describe_registered_workflow(args.workflow, input_path) print(json.dumps(description, indent=None if args.json else 2)) return 0 @@ -218,12 +244,21 @@ def build_parser() -> argparse.ArgumentParser: describe_parser.add_argument("--json", action="store_true", help="Print single-line JSON") describe_parser.set_defaults(handler=_tasks_describe) - workflow_parser = subparsers.add_parser("workflow", help="Inspect YAML workflows") + workflow_parser = subparsers.add_parser( + "workflow", help="Inspect YAML and registered workflows" + ) workflow_subparsers = workflow_parser.add_subparsers(dest="workflow_command", required=True) + workflow_list_parser = workflow_subparsers.add_parser( + "list", help="List registered workflow names" + ) + workflow_list_parser.add_argument( + "--json", action="store_true", help="Print machine-readable JSON" + ) + workflow_list_parser.set_defaults(handler=_workflow_list) workflow_describe_parser = workflow_subparsers.add_parser( "describe", help="Describe a workflow without running tasks" ) - workflow_describe_parser.add_argument("workflow", help="Path to the workflow YAML file") + workflow_describe_parser.add_argument("workflow", help=WORKFLOW_ARGUMENT_HELP) workflow_describe_parser.add_argument( "--input", help="Optional input YAML; reports provided, missing and configured values per task", @@ -247,13 +282,17 @@ def main(argv: Sequence[str] | None = None) -> int: args.command in ("run", "validate") or (args.command == "workflow" and args.workflow_command == "describe") ): + # A lookup message only names workflows, so it is safe to report. + message = ( + str(exc) + if isinstance(exc, WorkflowLookupError) + else "Workflow configuration is invalid" + ) print( json.dumps( { "status": "invalid", - "error": _error( - "configuration_error", exc, "Workflow configuration is invalid" - ), + "error": _error("configuration_error", exc, message), } ) ) diff --git a/taskmaestro/exceptions.py b/taskmaestro/exceptions.py index 43e72bb..7b3a206 100644 --- a/taskmaestro/exceptions.py +++ b/taskmaestro/exceptions.py @@ -78,6 +78,13 @@ class ConfigLoadError(WorkflowRunnerError): """Raised when YAML config loading fails (parse errors, import failures, validation).""" +class WorkflowLookupError(ConfigLoadError): + """A registered workflow name is unknown or its entry point cannot be loaded. + + The message names the registered workflows and contains no input values. + """ + + class PluginLoadError(WorkflowRunnerError): """Raised when an installed task or workflow entry point is invalid.""" diff --git a/taskmaestro/introspection.py b/taskmaestro/introspection.py index 235db12..8623ee8 100644 --- a/taskmaestro/introspection.py +++ b/taskmaestro/introspection.py @@ -11,7 +11,8 @@ import json import math import sys -from contextlib import redirect_stdout +from collections.abc import Iterator +from contextlib import contextmanager, redirect_stdout from dataclasses import asdict from datetime import date from pathlib import Path, PurePath @@ -27,7 +28,12 @@ from taskmaestro.job import EmptyConfig, Job, JobConfiguration from taskmaestro.task import get_input_type, get_output_type from taskmaestro.workflow import Workflow -from taskmaestro.yaml_config import load_workflow_only, workflow_imports +from taskmaestro.yaml_config import ( + load_input_configuration, + load_workflow_only, + registered_workflow, + workflow_imports, +) def python_type_name(cls: type[Any]) -> str: @@ -202,55 +208,11 @@ def _fill_missing( return JobConfiguration(filled) -def describe_yaml_workflow( - workflow_path: Path, - input_path: Path | None = None, - schema_generator: type[GenerateJsonSchema] = TaskSchemaGenerator, -) -> dict[str, Any]: - """Describe a YAML workflow without running any task. - - With ``input_path`` the description also reports provided, missing and - configured values per task. Raises ConfigLoadError if the workflow or input - is invalid; output printed while importing task modules goes to stderr. - """ +@contextmanager +def _describe_errors() -> Iterator[None]: + """Report describe failures as ConfigLoadError.""" try: - with redirect_stdout(sys.stderr), workflow_imports(Path(workflow_path).resolve().parent): - workflow, config = load_workflow_only(Path(workflow_path), input_path) - configured = None - missing = None - values = None - if input_path is not None: - assert config is not None - missing = missing_config_fields(workflow, config) - # Check the rest of the supplied config (map sources, root inputs) - # without constructing hooks or executing any tasks. Missing config - # fields are reported rather than rejected: the input may be a - # template whose remaining fields the caller fills in. - Job( - workflow, - EmptyConfig(), - job_configuration=_fill_missing(workflow, config, missing), - ) - configured = { - name: sorted(config.config_fields_for_task(name)) - for name, _task in workflow.topological_order() - } - # Raw input values, before Pydantic validation, so opaque or - # runtime-only markers pass through unchanged. - values = { - name: { - field: jsonable(value, f"{name}.{field}") - for field, value in sorted(config.get_config_for_task(name).items()) - } - for name, _task in workflow.topological_order() - } - return describe_workflow( - workflow, - configured=configured, - missing=missing, - values=values, - schema_generator=schema_generator, - ) + yield except WorkflowDefinitionError as exc: raise ConfigLoadError(f"Job validation failed: {exc}") from exc except ( @@ -262,6 +224,93 @@ def describe_yaml_workflow( raise ConfigLoadError(f"Cannot describe workflow: {exc}") from exc +def _describe_loaded( + workflow: Workflow, + config: JobConfiguration | None, + schema_generator: type[GenerateJsonSchema], + source: dict[str, str], +) -> dict[str, Any]: + """Describe a loaded workflow and, if given, its input configuration.""" + configured = None + missing = None + values = None + if config is not None: + missing = missing_config_fields(workflow, config) + # Check the rest of the supplied config (map sources, root inputs) + # without constructing hooks or executing any tasks. Missing config + # fields are reported rather than rejected: the input may be a + # template whose remaining fields the caller fills in. + Job( + workflow, + EmptyConfig(), + job_configuration=_fill_missing(workflow, config, missing), + ) + configured = { + name: sorted(config.config_fields_for_task(name)) + for name, _task in workflow.topological_order() + } + # Raw input values, before Pydantic validation, so opaque or + # runtime-only markers pass through unchanged. + values = { + name: { + field: jsonable(value, f"{name}.{field}") + for field, value in sorted(config.get_config_for_task(name).items()) + } + for name, _task in workflow.topological_order() + } + description = describe_workflow( + workflow, + configured=configured, + missing=missing, + values=values, + schema_generator=schema_generator, + ) + description["source"] = source + return description + + +def describe_yaml_workflow( + workflow_path: Path, + input_path: Path | None = None, + schema_generator: type[GenerateJsonSchema] = TaskSchemaGenerator, +) -> dict[str, Any]: + """Describe a YAML workflow without running any task. + + With ``input_path`` the description also reports provided, missing and + configured values per task. ``source`` is ``{"kind": "yaml", "path": ...}``. + Raises ConfigLoadError if the workflow or input is invalid; output printed + while importing task modules goes to stderr. + """ + path = Path(workflow_path).resolve() + with _describe_errors(), redirect_stdout(sys.stderr), workflow_imports(path.parent): + workflow, config = load_workflow_only(Path(workflow_path), input_path) + return _describe_loaded( + workflow, + config if input_path is not None else None, + schema_generator, + {"kind": "yaml", "path": str(path)}, + ) + + +def describe_registered_workflow( + name: str, + input_path: Path | None = None, + schema_generator: type[GenerateJsonSchema] = TaskSchemaGenerator, +) -> dict[str, Any]: + """Describe a workflow registered in the ``taskmaestro.workflows`` entry-point group. + + Like ``describe_yaml_workflow``; ``source`` is + ``{"kind": "entry_point", "name": ...}``. Raises ConfigLoadError for an + unknown workflow or invalid input. + """ + with _describe_errors(), redirect_stdout(sys.stderr): + workflow = registered_workflow(name) + config = load_input_configuration(workflow, input_path) if input_path else None + return _describe_loaded( + workflow, config, schema_generator, {"kind": "entry_point", "name": name} + ) + + def task_descriptor( task_id: str, task: type[Any], diff --git a/taskmaestro/yaml_config.py b/taskmaestro/yaml_config.py index f5f5602..0fc2fd7 100644 --- a/taskmaestro/yaml_config.py +++ b/taskmaestro/yaml_config.py @@ -16,8 +16,18 @@ from taskmaestro.context import ExecutionContext from taskmaestro.dependencies import CollectionDependency, OutputReference, collect -from taskmaestro.discovery import get_registered_task, registered_task_names -from taskmaestro.exceptions import ConfigLoadError, PluginLoadError, WorkflowDefinitionError +from taskmaestro.discovery import ( + get_registered_task, + get_registered_workflow, + registered_task_names, + registered_workflow_names, +) +from taskmaestro.exceptions import ( + ConfigLoadError, + PluginLoadError, + WorkflowDefinitionError, + WorkflowLookupError, +) from taskmaestro.hooks.base import BaseHook from taskmaestro.job import EmptyConfig, Job, JobConfiguration from taskmaestro.mapping import TaskMap @@ -266,6 +276,34 @@ def _timeout_seconds(self) -> float | None: # --- Main loader --- +def _read_input_yaml(input_path: Path) -> dict[str, Any]: + """Parse an input YAML file, which must contain a mapping.""" + try: + raw_input = _yaml_load(input_path.read_text()) + except yaml.YAMLError as exc: + raise ConfigLoadError(f"Input YAML parse error: {exc}") from exc + except OSError as exc: + raise ConfigLoadError(f"Cannot read input file '{input_path}': {exc}") from exc + if not isinstance(raw_input, dict): + raise ConfigLoadError("Input YAML file must contain a mapping") + return raw_input + + +def _check_input_keys(raw_input: dict[str, Any], task_names: typing.Collection[str]) -> None: + """Check that input keys are task names and their values mappings or null.""" + for key, value in raw_input.items(): + if key not in task_names: + raise ConfigLoadError( + f"Input top-level key '{key}' is not a task name " + f"(known tasks: {sorted(task_names)})" + ) + if value is not None and not isinstance(value, dict): + raise ConfigLoadError( + f"Input value for task '{key}' must be a mapping or null " + f"(got {type(value).__name__})" + ) + + def _load_workflow_only( workflow_path: Path, input_path: Path | None = None, @@ -305,15 +343,7 @@ def _load_workflow_only( # 2. Parse input YAML (optional for inner workflows) raw_input: dict[str, Any] = {} if input_path is not None: - try: - raw_input = _yaml_load(input_path.read_text()) - except yaml.YAMLError as exc: - raise ConfigLoadError(f"Input YAML parse error: {exc}") from exc - except OSError as exc: - raise ConfigLoadError(f"Cannot read input file '{input_path}': {exc}") from exc - - if not isinstance(raw_input, dict): - raise ConfigLoadError("Input YAML file must contain a mapping") + raw_input = _read_input_yaml(input_path) # 3. Schema validation try: @@ -414,17 +444,7 @@ def _resolve_yaml_collection(raw_collection: Any, context_task: str) -> Collecti # 6b. Input YAML always uses per-task configuration. Top-level keys are # registered task instance names and each value is a mapping or null. entry_by_name = {entry.registered_name: entry for entry in entries} - for key, value in raw_input.items(): - if key not in entry_by_name: - raise ConfigLoadError( - f"Input top-level key '{key}' is not a task name " - f"(known tasks: {sorted(entry_by_name)})" - ) - if value is not None and not isinstance(value, dict): - raise ConfigLoadError( - f"Input value for task '{key}' must be a mapping or null " - f"(got {type(value).__name__})" - ) + _check_input_keys(raw_input, entry_by_name) per_task_data: dict[str, dict[str, Any]] = {} per_task_cfg_fields: dict[str, list[str]] = {} @@ -576,15 +596,7 @@ def load_workflow_from_yaml(workflow_path: str | Path, input_path: str | Path) - raise ConfigLoadError("YAML file must contain a mapping at top level") # 2. Parse input YAML - try: - raw_input = _yaml_load(input_path.read_text()) - except yaml.YAMLError as exc: - raise ConfigLoadError(f"Input YAML parse error: {exc}") from exc - except OSError as exc: - raise ConfigLoadError(f"Cannot read input file '{input_path}': {exc}") from exc - - if not isinstance(raw_input, dict): - raise ConfigLoadError("Input YAML file must contain a mapping") + _read_input_yaml(input_path) # 3. Schema validation (for runner/context sections) try: @@ -649,6 +661,46 @@ def run_workflow_from_yaml(workflow_path: str | Path, input_path: str | Path) -> load_workflow_only = _load_workflow_only +def registered_workflow(name: str) -> Workflow: + """Load a registered workflow, raising WorkflowLookupError if it is unknown or invalid. + + The error message lists the registered workflow names. + """ + try: + return get_registered_workflow(name) + except PluginLoadError as exc: + names = sorted(registered_workflow_names()) + available = ", ".join(names) if names else "none" + raise WorkflowLookupError(f"{exc} (registered workflows: {available})") from exc + + +def load_input_configuration(workflow: Workflow, input_path: str | Path) -> JobConfiguration: + """Read an input YAML file as per-task configuration for ``workflow``. + + The input is checked the same way as for YAML workflows: top-level keys + must be task names and each value a mapping or null. + """ + raw_input = _read_input_yaml(Path(input_path)) + _check_input_keys(raw_input, {name for name, _task in workflow.topological_order()}) + return JobConfiguration({name: dict(values or {}) for name, values in raw_input.items()}) + + +def load_registered_workflow(name: str, input_path: str | Path) -> LoadedWorkflow: + """Load a registered workflow with an input YAML file, ready to run. + + Registered workflows have no ``runner:`` or ``context:`` sections, so the + workflow runs with a default ``Runner`` and ``ExecutionContext``. + Raises ConfigLoadError for an unknown workflow or invalid input. + """ + workflow = registered_workflow(name) + job_configuration = load_input_configuration(workflow, input_path) + try: + job = Job(workflow, EmptyConfig(), job_configuration=job_configuration) + except WorkflowDefinitionError as exc: + raise ConfigLoadError(f"Job validation failed: {exc}") from exc + return LoadedWorkflow(workflow=workflow, runner=Runner(), job=job, context=ExecutionContext()) + + @contextmanager def workflow_imports(directory: str | Path) -> Iterator[None]: """Make task modules in ``directory`` importable for the duration of loading.""" diff --git a/tests/test_cli.py b/tests/test_cli.py index 8afd006..3adba9f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -12,9 +12,10 @@ from pydantic import BaseModel from pydantic_core import core_schema -from taskmaestro import ExecutionContext, ObjectModel, Task +from taskmaestro import ExecutionContext, ObjectModel, Task, Workflow from taskmaestro.cli import _dependency_spec, _jsonable, main from taskmaestro.dependencies import CollectionRef, OutputRef +from tests.test_discovery import ExampleTask class ExternalClient: @@ -835,3 +836,181 @@ def test_python_dash_m_exit_code(tmp_path: Path) -> None: ) assert completed.returncode == 1 assert "Workflow failed at fail: intentional failure" in completed.stderr + + +configured_workflow = ( + Workflow.builder("example").add_task(ExampleTask, config_fields=["value"]).build() +) + + +@pytest.fixture +def registered_workflows(monkeypatch: pytest.MonkeyPatch) -> None: + entries = [ + EntryPoint( + name="example", + value="tests.test_cli:configured_workflow", + group="taskmaestro.workflows", + ), + EntryPoint(name="broken", value="missing_module:workflow", group="taskmaestro.workflows"), + ] + monkeypatch.setattr( + "taskmaestro.discovery.entry_points", + lambda *, group: entries if group == "taskmaestro.workflows" else [], + ) + + +def _example_input(tmp_path: Path, text: str = "ExampleTask:\n value: 4\n") -> Path: + input_path = tmp_path / "input.yaml" + input_path.write_text(text, encoding="utf-8") + return input_path + + +def test_workflow_list(registered_workflows: None, capsys: pytest.CaptureFixture[str]) -> None: + assert main(["workflow", "list", "--json"]) == 0 + assert json.loads(capsys.readouterr().out) == {"workflows": ["broken", "example"]} + assert main(["workflow", "list"]) == 0 + assert capsys.readouterr().out == "broken\nexample\n" + + +def test_workflow_list_empty( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setattr("taskmaestro.discovery.entry_points", lambda *, group: []) + + assert main(["workflow", "list", "--json"]) == 0 + assert json.loads(capsys.readouterr().out) == {"workflows": []} + + +def test_registered_workflow_run_validate_and_graph( + registered_workflows: None, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + input_path = str(_example_input(tmp_path)) + + assert main(["run", "example", "--input", input_path, "--json"]) == 0 + assert json.loads(capsys.readouterr().out) == { + "status": "completed", + "workflow": "example", + "result": {"value": 5}, + } + assert main(["validate", "example", "--input", input_path, "--json"]) == 0 + assert json.loads(capsys.readouterr().out)["status"] == "valid" + assert main(["graph", "example", "--input", input_path]) == 0 + assert "graph TD" in capsys.readouterr().out + + +def test_registered_workflow_describe( + registered_workflows: None, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + assert main(["workflow", "describe", "example", "--json"]) == 0 + description = json.loads(capsys.readouterr().out) + assert description["workflow"] == "example" + assert description["source"] == {"kind": "entry_point", "name": "example"} + assert description["tasks"][0]["config_values"] is None + + input_path = _example_input(tmp_path, "ExampleTask:\n") + assert main(["workflow", "describe", "example", "--input", str(input_path), "--json"]) == 0 + task = json.loads(capsys.readouterr().out)["tasks"][0] + assert task["missing_config_fields"] == ["value"] + assert task["config_values"] == {} + + input_path = _example_input(tmp_path) + assert main(["workflow", "describe", "example", "--input", str(input_path), "--json"]) == 0 + task = json.loads(capsys.readouterr().out)["tasks"][0] + assert task["missing_config_fields"] == [] + assert task["config_values"] == {"value": 4} + + +@pytest.mark.parametrize( + "argv", + [ + ["run", "unknown", "--input", "input.yaml", "--json"], + ["validate", "unknown", "--input", "input.yaml", "--json"], + ["workflow", "describe", "unknown", "--json"], + ], +) +def test_unknown_registered_workflow_reports_configuration_error( + registered_workflows: None, argv: list[str], capsys: pytest.CaptureFixture[str] +) -> None: + assert main(argv) == 2 + error = json.loads(capsys.readouterr().out) + assert error["status"] == "invalid" + assert error["error"]["code"] == "configuration_error" + assert "'unknown'" in error["error"]["message"] + assert "registered workflows: broken, example" in error["error"]["message"] + + +def test_unknown_registered_workflow_without_json( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setattr("taskmaestro.discovery.entry_points", lambda *, group: []) + + assert main(["graph", "unknown", "--input", "input.yaml"]) == 2 + captured = capsys.readouterr() + assert captured.out == "" + assert "Configuration error:" in captured.err + assert "registered workflows: none" in captured.err + + +def test_broken_registered_workflow_reports_configuration_error( + registered_workflows: None, capsys: pytest.CaptureFixture[str] +) -> None: + assert main(["workflow", "describe", "broken", "--json"]) == 2 + error = json.loads(capsys.readouterr().out)["error"] + assert error["code"] == "configuration_error" + assert "broken" in error["message"] + + +@pytest.mark.parametrize( + ("text", "message"), + [ + ("other:\n value: 1\n", "'other' is not a task name"), + ("ExampleTask: 3\n", "must be a mapping or null"), + ("- 1\n", "must contain a mapping"), + ], +) +def test_registered_workflow_rejects_invalid_input( + registered_workflows: None, + tmp_path: Path, + text: str, + message: str, + capsys: pytest.CaptureFixture[str], +) -> None: + input_path = str(_example_input(tmp_path, text)) + + assert main(["validate", "example", "--input", input_path]) == 2 + assert message in capsys.readouterr().err + + +def test_registered_workflow_missing_config_fails_job_validation( + registered_workflows: None, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + input_path = str(_example_input(tmp_path, "ExampleTask:\n")) + + assert main(["validate", "example", "--input", input_path, "--json"]) == 2 + error = json.loads(capsys.readouterr().out)["error"] + assert error["message"] == "Workflow configuration is invalid" + assert error["issues"] == [{"field": "value", "code": "missing"}] + + +def test_workflow_argument_that_is_an_existing_file_is_a_path( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + workflow, input_path = _files(tmp_path) + named = tmp_path / "example" + named.write_text(workflow.read_text(), encoding="utf-8") + monkeypatch.chdir(tmp_path) + + assert main(["validate", "example", "--input", str(input_path), "--json"]) == 0 + assert json.loads(capsys.readouterr().out)["workflow"] == "cli_test" + + +def test_yaml_workflow_describe_reports_source( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + workflow, _input_path = _files(tmp_path) + + assert main(["workflow", "describe", str(workflow), "--json"]) == 0 + assert json.loads(capsys.readouterr().out)["source"] == { + "kind": "yaml", + "path": str(workflow.resolve()), + } From a90588019d2d61d5365f50ce40fd8b7040e35001 Mon Sep 17 00:00:00 2001 From: Kristian Bendiksen Date: Wed, 21 Oct 2026 15:33:24 +0200 Subject: [PATCH 3/4] Ignore non-identifier globals when resolving type aliases in graphs pydantic 2.14 also stores a module-level parametrized model such as ObjectModel[str] in the caller's globals under its bracketed name, so the alias lookup could return 'ObjectModel[str]' instead of the alias. --- taskmaestro/visualization.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/taskmaestro/visualization.py b/taskmaestro/visualization.py index 954c569..ce27d4f 100644 --- a/taskmaestro/visualization.py +++ b/taskmaestro/visualization.py @@ -33,7 +33,8 @@ def _safe_type_name(tp: Any, context_cls: type | None = None) -> str: if module_name and module_name in sys.modules: mod = sys.modules[module_name] for attr_name, attr_val in list(vars(mod).items()): - if attr_val is tp and not attr_name.startswith("_"): + # Pydantic may also register the model under its bracketed name. + if attr_val is tp and attr_name.isidentifier() and not attr_name.startswith("_"): return attr_name return name.replace("[", "‹").replace("]", "›") From b675d6fa3a40bb7ec9ab6ed355ce3777c4d89f33 Mon Sep 17 00:00:00 2001 From: Kristian Bendiksen Date: Wed, 21 Oct 2026 15:39:26 +0200 Subject: [PATCH 4/4] Require pydantic 2.4 or newer error_payload calls ValidationError.errors(include_input=False), which pydantic added in 2.4. The test suite fails on 2.0-2.3 and passes on 2.4 and later. --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index d65fa83..9409bb4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,7 +24,7 @@ classifiers = [ "Typing :: Typed", ] dependencies = [ - "pydantic>=2.0,<3.0", + "pydantic>=2.4,<3.0", "pyyaml>=6.0,<7.0", ]