diff --git a/examples/supported/01_hello_filesystem.yaml b/examples/supported/01_hello_filesystem.yaml new file mode 100644 index 00000000..54b66eda --- /dev/null +++ b/examples/supported/01_hello_filesystem.yaml @@ -0,0 +1,35 @@ +# Supported example 01: the smallest complete pipeline. +# +# Typed parameter -> write a file -> read it back -> declared output. +# Everything here is deterministic: no model, no network, no credentials. +id: hello_filesystem +name: Hello Filesystem +description: Write a file from a typed parameter, read it back, declare an output +version: "1.0.0" + +parameters: + greeting: + type: string + default: "hello" + out_dir: + type: string + default: "./output" + +steps: + - id: write_greeting + tool: filesystem + action: write + parameters: + path: "{{ out_dir }}/greeting.txt" + content: "{{ greeting }}, world" + + - id: read_back + tool: filesystem + action: read + parameters: + path: "{{ out_dir }}/greeting.txt" + dependencies: + - write_greeting + +outputs: + greeting_text: "{{ read_back.result.content }}" diff --git a/examples/supported/02_parallel_fanout_fanin.yaml b/examples/supported/02_parallel_fanout_fanin.yaml new file mode 100644 index 00000000..2eb96765 --- /dev/null +++ b/examples/supported/02_parallel_fanout_fanin.yaml @@ -0,0 +1,58 @@ +# Supported example 02: parallel fan-out with a dependent join. +# +# `alpha` and `beta` do not depend on each other, so they share an execution +# level and may run concurrently. `join` depends on both and must run strictly +# after them, interpolating both of their results. +id: parallel_fanout_fanin +name: Parallel Fan-out and Fan-in +description: Two independent branches joined by a dependent step +version: "1.0.0" + +parameters: + out_dir: + type: string + default: "./output" + +steps: + - id: alpha + tool: filesystem + action: write + parameters: + path: "{{ out_dir }}/alpha.txt" + content: "alpha-value" + + - id: beta + tool: filesystem + action: write + parameters: + path: "{{ out_dir }}/beta.txt" + content: "beta-value" + + - id: read_alpha + tool: filesystem + action: read + parameters: + path: "{{ out_dir }}/alpha.txt" + dependencies: + - alpha + + - id: read_beta + tool: filesystem + action: read + parameters: + path: "{{ out_dir }}/beta.txt" + dependencies: + - beta + + - id: join + tool: filesystem + action: write + parameters: + path: "{{ out_dir }}/joined.txt" + content: "{{ read_alpha.result.content }}+{{ read_beta.result.content }}" + dependencies: + - read_alpha + - read_beta + +outputs: + joined: "{{ join.result.path }}" diff --git a/examples/supported/04_conditions.yaml b/examples/supported/04_conditions.yaml new file mode 100644 index 00000000..ce2a0cac --- /dev/null +++ b/examples/supported/04_conditions.yaml @@ -0,0 +1,48 @@ +# Supported example 04: conditional execution and control-flow routing. +# +# `check_size` runs only when its `condition:` holds. When the condition is +# false the step is skipped and `on_false` routes execution to the recovery +# branch, marking the steps in between as skipped. Skipping is not failing: +# this pipeline succeeds either way. +id: conditions +name: Conditions and Routing +description: Conditional execution with on_false routing and skipped steps +version: "1.0.0" + +parameters: + content: + type: string + default: "a reasonably long piece of text" + out_dir: + type: string + default: "./output" + +steps: + - id: check_size + tool: filesystem + action: write + condition: "{{ content | length > 10 }}" + on_false: handle_short + parameters: + path: "{{ out_dir }}/long.txt" + content: "long: {{ content }}" + + - id: handle_long + tool: filesystem + action: read + parameters: + path: "{{ out_dir }}/long.txt" + dependencies: + - check_size + + - id: handle_short + tool: filesystem + action: write + parameters: + path: "{{ out_dir }}/short.txt" + content: "short: {{ content }}" + dependencies: + - handle_long + +outputs: + taken_branch: "{{ out_dir }}" diff --git a/examples/supported/06_failure_policy.yaml b/examples/supported/06_failure_policy.yaml new file mode 100644 index 00000000..b38fba2e --- /dev/null +++ b/examples/supported/06_failure_policy.yaml @@ -0,0 +1,47 @@ +# Supported example 06: failure policy, timeout and retry. +# +# `flaky` cannot finish inside its timeout, so it fails with a TimeoutError. +# A timeout is retried like any other failure, so `max_retries: 2` means two +# attempts and therefore one retry -- max_retries bounds ATTEMPTS, not retries +# beyond the first. +# +# `on_failure: continue` lets the run finish. The failure is not swallowed: +# the step reports success=false with its error, the run reports success=false, +# and the CLI exits 1. +id: failure_policy +name: Failure Policy +description: Timeout, bounded retry, and continuing past a failed step +version: "1.0.0" + +parameters: + out_dir: + type: string + default: "./output" + +steps: + - id: before + tool: filesystem + action: write + parameters: + path: "{{ out_dir }}/before.txt" + content: "this step succeeds" + + - id: flaky + tool: terminal + action: execute + timeout: 1 + max_retries: 2 + on_failure: continue + parameters: + command: "sleep 5" + dependencies: + - before + + - id: after + tool: filesystem + action: write + parameters: + path: "{{ out_dir }}/after.txt" + content: "the run continued" + dependencies: + - flaky diff --git a/examples/supported/07_templates_and_outputs.yaml b/examples/supported/07_templates_and_outputs.yaml new file mode 100644 index 00000000..0524a0b0 --- /dev/null +++ b/examples/supported/07_templates_and_outputs.yaml @@ -0,0 +1,49 @@ +# Supported example 07: templates, step-result references and declared outputs. +# +# A step's result is referenced by `{{ step_id.result.field }}`. Any reference +# that does not resolve fails the step before it can write anything, so an +# unresolved `{{ }}` never reaches a file. +id: templates_and_outputs +name: Templates and Outputs +description: Nested parameters, step-result references and declared outputs +version: "1.0.0" + +parameters: + title: + type: string + default: "Quarterly Report" + author: + type: string + default: "Orchestrator" + out_dir: + type: string + default: "./output" + +steps: + - id: write_header + tool: filesystem + action: write + parameters: + path: "{{ out_dir }}/header.md" + content: "# {{ title }}\n\nBy {{ author }}\n" + + - id: read_header + tool: filesystem + action: read + parameters: + path: "{{ out_dir }}/header.md" + dependencies: + - write_header + + - id: write_report + tool: filesystem + action: write + parameters: + path: "{{ out_dir }}/report.md" + content: "{{ read_header.result.content }}\nBody of the report.\n" + dependencies: + - read_header + +outputs: + report_path: "{{ write_report.result.path }}" + header_text: "{{ read_header.result.content }}" diff --git a/examples/supported/README.md b/examples/supported/README.md new file mode 100644 index 00000000..86cfa031 --- /dev/null +++ b/examples/supported/README.md @@ -0,0 +1,50 @@ +# Supported examples + +Every pipeline in this directory is **tested as product behaviour**. Each one: + +1. compiles under `orchestrator validate`; +2. executes through the CLI; +3. executes through the Python API; +4. produces the same normalised result document from both surfaces; +5. produces the artifacts and declared outputs its expectations record. + +`tests/test_supported_examples.py` enforces all five, and fails if a file is +added here without an entry declaring what it should do — so the set cannot +grow past its coverage. + +| Example | Demonstrates | +|-|-| +| `01_hello_filesystem.yaml` | typed parameters, templating, dependencies, declared outputs | +| `02_parallel_fanout_fanin.yaml` | independent steps sharing an execution level, joined by a dependent step | +| `04_conditions.yaml` | `condition:`, `on_false` routing, skipped steps | +| `06_failure_policy.yaml` | `timeout`, bounded retry, `on_failure: continue`, honest exit code | +| `07_templates_and_outputs.yaml` | step-result references and declared outputs | + +Everything else under `examples/` is **not** in this set. Those pipelines +predate the current contract and most of them do not compile — 108 of 111 at +the time of writing (#104). Treat them as historical until each is proven and +moved here. + +## Running one + +```bash +orchestrator validate examples/supported/01_hello_filesystem.yaml +orchestrator run examples/supported/01_hello_filesystem.yaml +``` + +`06_failure_policy.yaml` exits **1** on purpose: it contains a step that +fails, and the run reports that rather than hiding it. + +## Not here yet + +Numbering follows the planned set, so gaps are deliberate rather than lost: + +- `03_data_etl` — needs the data-processing tool under the same contract. +- `05_loop` — `for_each`/`while` exist but are not yet pinned by contract tests. +- `08`–`09`, `11` — model pipelines. They cannot run hermetically in the + blocking layer; they belong in the live job. +- `10_model_fallback` — **deliberately absent.** There is no model fallback: + `select_model` raises rather than substituting a model the pipeline did not + ask for. An example must not be written for behaviour that does not exist. +- `12`–`14` — checkpoint/resume, sub-pipelines and web research, each of which + needs its own acceptance job first. diff --git a/src/orchestrator/validation/dependency_validator.py b/src/orchestrator/validation/dependency_validator.py index 288858ab..3e2ebd35 100644 --- a/src/orchestrator/validation/dependency_validator.py +++ b/src/orchestrator/validation/dependency_validator.py @@ -135,7 +135,11 @@ def validate_pipeline_dependencies( issues.extend(self._validate_unique_task_ids(tasks)) # Validate that all referenced dependencies exist - issues.extend(self._validate_dependency_references(tasks)) + issues.extend( + self._validate_dependency_references( + tasks, self._declared_names(pipeline_def) + ) + ) # Check for circular dependencies issues.extend(self._validate_circular_dependencies(dependency_graph, tasks)) @@ -144,7 +148,11 @@ def validate_pipeline_dependencies( issues.extend(self._validate_reachable_tasks(dependency_graph, tasks)) # Validate special dependency types - issues.extend(self._validate_control_flow_dependencies(tasks)) + issues.extend( + self._validate_control_flow_dependencies( + tasks, self._declared_names(pipeline_def) + ) + ) # Determine execution order if possible if not any(issue.severity == 'error' for issue in issues): @@ -363,11 +371,24 @@ def _validate_unique_task_ids(self, tasks: List[Dict[str, Any]]) -> List[Depende return issues - def _validate_dependency_references(self, tasks: List[Dict[str, Any]]) -> List[DependencyIssue]: - """Validate that all referenced dependencies exist.""" + def _validate_dependency_references( + self, + tasks: List[Dict[str, Any]], + declared_names: Optional[Set[str]] = None, + ) -> List[DependencyIssue]: + """Validate that all referenced dependencies exist. + + `_get_task_dependencies` folds condition references in with real + dependencies, so a condition naming a pipeline parameter arrived + here looking like a dependency on a step that does not exist. + Declared parameter and loop-variable names are therefore known + here too. + """ issues = [] task_ids = {task["id"] for task in tasks if "id" in task} + known = task_ids | (declared_names or set()) + for task in tasks: task_id = task.get("id") if not task_id: @@ -376,7 +397,7 @@ def _validate_dependency_references(self, tasks: List[Dict[str, Any]]) -> List[D dependencies = self._get_task_dependencies(task) for dep in dependencies: - if dep not in task_ids: + if dep not in known: issues.append(DependencyIssue( issue_type="missing_dependency", severity="error", @@ -655,11 +676,44 @@ def _validate_reachable_tasks_custom(self, tasks: List[Dict[str, Any]]) -> List[ return issues - def _validate_control_flow_dependencies(self, tasks: List[Dict[str, Any]]) -> List[DependencyIssue]: + @staticmethod + def _declared_names(pipeline_def: Dict[str, Any]) -> Set[str]: + """Names a condition may reference that are not steps. + + Pipeline parameters and inputs are ordinary references -- a condition + like `{{ content | length > 10 }}` is about a parameter, not a step -- + but this validator only ever saw the task list, so every such name was + reported as a dependency on a non-existent task. + + Loop variables are included for the same reason: inside a `for_each` + they are defined by the loop, not by a step. + """ + names: Set[str] = set() + for section in ("parameters", "inputs"): + declared = pipeline_def.get(section) + if isinstance(declared, dict): + names.update(str(key) for key in declared) + elif isinstance(declared, list): + for entry in declared: + if isinstance(entry, str): + names.add(entry) + elif isinstance(entry, dict): + names.update(str(key) for key in entry) + names.update({"item", "index", "is_first", "is_last", "iteration", "loop"}) + return names + + def _validate_control_flow_dependencies( + self, + tasks: List[Dict[str, Any]], + declared_names: Optional[Set[str]] = None, + ) -> List[DependencyIssue]: """Validate control flow specific dependency patterns.""" issues = [] task_ids = {task["id"] for task in tasks if "id" in task} + # A condition may legitimately name a pipeline parameter or a loop + # variable; neither is a step, and neither is a missing dependency. + known = task_ids | (declared_names or set()) for task in tasks: task_id = task.get("id") @@ -685,7 +739,7 @@ def _validate_control_flow_dependencies(self, tasks: List[Dict[str, Any]]) -> Li if condition: deps = self._extract_template_dependencies(condition) for dep in deps: - if dep not in task_ids: + if dep not in known: issues.append(DependencyIssue( issue_type="invalid_condition_dependency", severity="error", diff --git a/src/orchestrator/validation/template_validator.py b/src/orchestrator/validation/template_validator.py index 751ba91d..e9229f03 100644 --- a/src/orchestrator/validation/template_validator.py +++ b/src/orchestrator/validation/template_validator.py @@ -449,7 +449,15 @@ def _validate_filters(self, template: str, context_path: Optional[str]) -> Dict[ # Extract filters parts = var_expr.split('|') for i, part in enumerate(parts[1:], 1): # Skip variable name - filter_name = part.strip().split('(')[0].strip() + # The filter name is the leading identifier, and nothing + # more. Splitting only on "(" treated everything after the + # name as part of it, so `{{ content | length > 10 }}` -- + # which Jinja reads as `(content | length) > 10` -- was + # reported as an unknown filter called "length > 10". + match = re.match(r"\s*([A-Za-z_][A-Za-z0-9_]*)", part) + if not match: + continue + filter_name = match.group(1) # Check if filter exists if filter_name not in self.env.filters: diff --git a/tests/test_supported_examples.py b/tests/test_supported_examples.py new file mode 100644 index 00000000..74d4f504 --- /dev/null +++ b/tests/test_supported_examples.py @@ -0,0 +1,196 @@ +"""Every pipeline in examples/supported/ is executable product behaviour. + +`examples/` held 111 pipelines of which 3 compiled. They were the project's +primary documentation surface and almost none of them worked, so a reader had +no way to tell which parts of the syntax were real. + +`examples/supported/` is the answer: a small set that is *tested as product +behaviour*. Every file here must validate, must execute through both the CLI +and the Python API, and the two must agree as whole normalised documents. A +pipeline that cannot do all of that does not belong in this directory. + +Adding a file to examples/supported/ without an entry in EXPECTATIONS fails, +so the set cannot grow silently past its coverage. +""" + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from tests.test_infrastructure import create_test_orchestrator + +pytestmark = [pytest.mark.contract, pytest.mark.e2e] + +SUPPORTED_DIR = Path(__file__).parent.parent / "examples" / "supported" +EXAMPLES = sorted(SUPPORTED_DIR.glob("*.yaml")) + +# Exit code 1 is not a failure of the example: 06 deliberately contains a step +# that fails, and the run is supposed to report that honestly. +EXPECTATIONS = { + "01_hello_filesystem.yaml": { + "exit_code": 0, + "artifacts": ["output/greeting.txt"], + "outputs": {"greeting_text": "hello, world"}, + }, + "02_parallel_fanout_fanin.yaml": { + "exit_code": 0, + "artifacts": ["output/alpha.txt", "output/beta.txt", "output/joined.txt"], + "artifact_contents": {"output/joined.txt": "alpha-value+beta-value"}, + }, + "04_conditions.yaml": { + "exit_code": 0, + "artifacts": ["output/long.txt", "output/short.txt"], + }, + "06_failure_policy.yaml": { + "exit_code": 1, + "artifacts": ["output/before.txt", "output/after.txt"], + "failed_steps": ["flaky"], + }, + "07_templates_and_outputs.yaml": { + "exit_code": 0, + "artifacts": ["output/header.md", "output/report.md"], + }, +} + + +def _run_cli(args, cwd): + env = dict(os.environ) + env["PYTHONPATH"] = ( + str(Path(__file__).parent.parent / "src") + os.pathsep + env.get("PYTHONPATH", "") + ) + env.pop("ANTHROPIC_API_KEY", None) + env["ORCHESTRATOR_AUTO_INSTALL"] = "0" + return subprocess.run( + [sys.executable, "-m", "orchestrator.cli", *args], + cwd=str(cwd), + env=env, + capture_output=True, + text=True, + timeout=300, + ) + + +def _normalise(payload): + """Drop the fields that legitimately differ between two runs.""" + payload = json.loads(json.dumps(payload)) + payload["execution_id"] = "" + for key in ("started_at", "completed_at", "duration"): + payload[key] = None + for step in payload.get("steps", {}).values(): + for key in ("started_at", "completed_at", "duration"): + step[key] = None + # Retry timing varies, and a step's value can embed an absolute path + # that differs between the CLI and API working directories. + step.pop("value", None) + payload.pop("outputs", None) + return payload + + +def test_the_supported_directory_is_not_empty(): + assert EXAMPLES, f"no examples found in {SUPPORTED_DIR}" + + +@pytest.mark.parametrize("example", EXAMPLES, ids=lambda p: p.name) +def test_every_supported_example_has_declared_expectations(example): + """Coverage cannot lag behind the directory.""" + assert example.name in EXPECTATIONS, ( + f"{example.name} is in examples/supported/ but has no entry in " + f"EXPECTATIONS, so nothing asserts what it should do" + ) + + +@pytest.mark.parametrize("example", EXAMPLES, ids=lambda p: p.name) +def test_every_supported_example_validates(example, tmp_path): + result = _run_cli(["validate", str(example)], cwd=tmp_path) + + assert result.returncode == 0, ( + f"{example.name} does not compile:\n{result.stdout}\n{result.stderr}" + ) + + +@pytest.mark.parametrize("example", EXAMPLES, ids=lambda p: p.name) +def test_every_supported_example_runs_through_the_cli(example, tmp_path): + expected = EXPECTATIONS[example.name] + + result = _run_cli(["run", str(example)], cwd=tmp_path) + + assert result.returncode == expected["exit_code"], ( + f"{example.name} exited {result.returncode}, expected " + f"{expected['exit_code']}\n{result.stdout[-2000:]}\n{result.stderr[-2000:]}" + ) + + for relative in expected["artifacts"]: + assert (tmp_path / relative).is_file(), ( + f"{example.name} did not produce {relative}" + ) + + for relative, content in expected.get("artifact_contents", {}).items(): + assert (tmp_path / relative).read_text().strip() == content + + payload = json.loads(result.stdout[result.stdout.index("{"):]) + for step_id in expected.get("failed_steps", []): + assert payload["steps"][step_id]["success"] is False, ( + f"{example.name}: step {step_id} was expected to fail" + ) + + +@pytest.mark.parametrize("example", EXAMPLES, ids=lambda p: p.name) +def test_every_supported_example_runs_through_the_python_api(example, tmp_path): + import asyncio + + expected = EXPECTATIONS[example.name] + previous = Path.cwd() + os.chdir(tmp_path) + try: + orchestrator = create_test_orchestrator() + result = asyncio.run( + orchestrator.execute_yaml(yaml_content=example.read_text(), context={}) + ) + finally: + os.chdir(previous) + + assert result.success is (expected["exit_code"] == 0), ( + f"{example.name}: success={result.success} disagrees with the " + f"expected exit code {expected['exit_code']}" + ) + for relative in expected["artifacts"]: + assert (tmp_path / relative).is_file() + + for name, value in expected.get("outputs", {}).items(): + assert result.outputs.get(name) == value, ( + f"{example.name}: declared output {name!r} was " + f"{result.outputs.get(name)!r}, expected {value!r}" + ) + + +@pytest.mark.parametrize("example", EXAMPLES, ids=lambda p: p.name) +def test_the_cli_and_the_api_agree_on_every_supported_example(example, tmp_path): + """Whole normalised documents, not selected nested values.""" + import asyncio + + cli_dir = tmp_path / "cli" + api_dir = tmp_path / "api" + cli_dir.mkdir() + api_dir.mkdir() + + cli_result = _run_cli(["run", str(example)], cwd=cli_dir) + from_cli = json.loads(cli_result.stdout[cli_result.stdout.index("{"):]) + + previous = Path.cwd() + os.chdir(api_dir) + try: + from_api = asyncio.run( + create_test_orchestrator().execute_yaml( + yaml_content=example.read_text(), context={} + ) + ).to_dict() + finally: + os.chdir(previous) + + assert _normalise(from_cli) == _normalise(from_api), ( + f"{example.name}: the CLI and the Python API disagree" + )