From 724a5c34b847e0f495b706cb98abe52e0ccd8a92 Mon Sep 17 00:00:00 2001 From: Jeremy Manning Date: Tue, 4 Aug 2026 12:01:33 -0400 Subject: [PATCH] Derive loop scopes and bindings from the runtime, not from a table (#472) #473 replaced one "inside a loop" boolean with per-construct contracts. The direction was right and the tables were wrong, because they were written from reading the source and checked only against themselves: every test asked whether the validator agreed with `loop_contracts`, and both sides came from the same file. That proves consistency and says nothing about truth. So the tables are now read off running pipelines, and `tests/test_loop_runtime_parity.py` re-derives them the same way on every run: a name a contract declares must render in a real execution, and a name it withholds must fail to. The suite is hermetic -- filesystem tool only, no model, no network. What that turned up: * `foreach` is not an alias of `for_each`. `for_each` writes its two files; `foreach` gives `Schema validation failed: 325 errors`, and in a shape that does compile the body runs once with nothing bound. #473 declared it supported on the strength of a comment. Removed (#475). * An `action_loop`'s body was validated outside the loop it is the body of. The `action_loop` key holds the actions, and `_build_iteration_context` builds iteration state before they run. * A `while` condition is not a `for_each` iterable. It is re-evaluated every iteration against `iteration` and `loop_state`, so `while: "{{ iteration < 3 }}"` -- the ordinary way to write a bounded loop -- was being rejected. * `create_parallel_queue` mixes phases inside one object: `on` generates the queue, the nested actions run per item. A flat tuple of source fields cannot say that, so scope is now keyed by field path. * A step declaring two loop constructs was silently resolved by declaration order in this module, which no engine agreed to. It is now an error. * The `$` spelling is an observed list, not `"$" + every binding`: `{{ $position }}` raises `unexpected char '$'` at run time while `{{ $item }}` resolves (#474). Four runtime defects found while probing, filed rather than fixed here: #475 (dead `foreach`), #476 (`action_loop`'s required `until` is never evaluated, so the loop always runs once), #477 (constructs leak each other's names as stale values -- a debug dict became the template surface), #478 (`create_parallel_queue` reports success while skipping an unresolved write). Where the runtime leaks a name with no meaning for the construct, this withholds it rather than blessing it, so validation is deliberately stricter than the runtime in exactly the cases #477 covers. That is a language decision, recorded in the module docstring. `docs/loop_variables.md` documented `{{ $position }}` (a compile error), an `as:` key nothing reads, and "both formats work identically". Rewritten from the verified tables, with the two known defects marked inline. Co-Authored-By: Claude Opus 5 (1M context) --- docs/loop_variables.md | 218 +++++++++++------ src/orchestrator/core/loop_contracts.py | 228 ++++++++++------- src/orchestrator/core/template_globals.py | 2 +- .../validation/template_validator.py | 73 ++++-- tests/test_loop_contracts.py | 161 +++++++++--- tests/test_loop_runtime_parity.py | 231 ++++++++++++++++++ 6 files changed, 704 insertions(+), 209 deletions(-) create mode 100644 tests/test_loop_runtime_parity.py diff --git a/docs/loop_variables.md b/docs/loop_variables.md index 771aee74..c9a59e3f 100644 --- a/docs/loop_variables.md +++ b/docs/loop_variables.md @@ -1,29 +1,17 @@ -# Loop Variables in Orchestrator +# Loop Variables -## Overview +Each loop construct binds its own names, in its own places. There is no single +list that applies to all of them, and the same name can be bound in a step's +body while being unavailable in the expression that starts the loop. -When using `for_each` loops in Orchestrator pipelines, several variables are automatically available within the loop context for use in templates. +The tables here are the ones in `src/orchestrator/core/loop_contracts.py`, and +`tests/test_loop_runtime_parity.py` checks them by *executing* a pipeline for +every name: a name listed here renders in a real run, and a name absent from a +construct's table does not. -## Available Loop Variables +## `for_each` -### Core Variables - -- `{{ item }}` or `{{ $item }}` - The current item being processed -- `{{ index }}` or `{{ $index }}` - The zero-based index of the current iteration -- `{{ is_first }}` or `{{ $is_first }}` - Boolean indicating if this is the first iteration -- `{{ is_last }}` or `{{ $is_last }}` - Boolean indicating if this is the last iteration - -### Additional Variables - -- `{{ length }}` or `{{ $length }}` - Total number of items in the loop -- `{{ position }}` or `{{ $position }}` - One-based position (index + 1) -- `{{ remaining }}` or `{{ $remaining }}` - Number of items remaining after current -- `{{ has_next }}` or `{{ $has_next }}` - Boolean indicating if there's a next item -- `{{ has_prev }}` or `{{ $has_prev }}` - Boolean indicating if there's a previous item - -## Examples - -### Simple Loop +Iterates a collection. ```yaml steps: @@ -36,98 +24,174 @@ steps: parameters: path: "output/{{ item }}_{{ index }}.txt" content: | - Processing item {{ item }} at position {{ position }} + Processing {{ item }} at position {{ position }} This is item {{ index }} of {{ length }} - First item: {{ is_first }} - Last item: {{ is_last }} + First: {{ is_first }} Last: {{ is_last }} ``` -### Loop with Dependencies +Bound in the body: + +|name|meaning| +|-|-| +|`item`|the current item| +|`index`|zero-based iteration number| +|`is_first`, `is_last`|whether this is the first or last item| +|`position`|one-based position (`index + 1`)| +|`length`|number of items| +|`remaining`|items after this one| +|`has_next`, `has_prev`|whether a next or previous item exists| +|`loop_id`|the loop's identifier| +|`$loop_name`|the loop's name — this one has no bare spelling| + +**Not bound in the iterable.** `for_each: "{{ item.children }}"` cannot work: +the collection has to be evaluated before there is an item to bind. Validation +rejects it. + +## `while` + +Repeats until a condition goes false. ```yaml steps: - - id: process - for_each: "{{ data_items }}" + - id: retry + while: "{{ iteration < 3 }}" + max_iterations: 10 steps: - - id: transform - action: generate_text + - id: attempt parameters: - prompt: "Transform: {{ item }}" - - - id: save - tool: filesystem - action: write - parameters: - path: "results/{{ item }}.txt" - content: | - Original: {{ item }} - Transformed: {{ transform }} - Index: {{ index }} - dependencies: - - transform + note: "attempt {{ position }} of loop {{ loop_id }}" ``` -### Named Loops (Advanced) +Bound in the body: `iteration`, `index`, `is_first`, `position`, `loop_id`, +`loop_name`, `loop_state`. + +Bound in the `while:` and `until:` conditions: **`iteration` and `loop_state` +only**. Unlike a `for_each` iterable, a condition is re-evaluated every +iteration, so it can see the counter — but it sees only what the loop handler +puts in scope at that moment, which is less than the body gets. + +A `while` loop walks no collection, so it binds no `item`, `length` or +`is_last`. -When loops are nested or need explicit naming: +## `action_loop` + +Repeats a list of actions rather than walking a collection. ```yaml steps: - - id: outer - for_each: "{{ categories }}" - as: category_loop # Optional: give the loop a name - steps: - - id: process - action: generate_text + - id: poll + action_loop: + - action: filesystem parameters: - prompt: "Process category {{ $category_loop.item }}" + action: write + path: "out/{{ iteration }}.txt" + content: "attempt {{ iteration }}" + until: "{{ iteration >= 3 }}" + max_iterations: 5 ``` -## Template Formats +Bound in the body: `iteration`, `is_first`, `loop_id`, `has_previous`, +`total_duration`, `termination_reason`. No `item`, `index` or `position` — +there is no collection and no position in one. -Loop variables support both formats: -- Dollar prefix: `{{ $item }}`, `{{ $index }}` -- Without prefix: `{{ item }}`, `{{ index }}` +The `action_loop` key holds the body, so those names are available inside it. -Both formats work identically within loop contexts. +> **Known defect:** the `until:` condition is required but never evaluated, so +> the loop always runs a single iteration. See issue #476. -## Nested Loops +## `create_parallel_queue` -For nested loops, you can access parent loop variables using the loop name: +Generates a queue and runs actions across it in parallel. + +```yaml +steps: + - id: fan_out + action: create_parallel_queue + create_parallel_queue: + "on": "{{ work_items }}" + action_loop: + - action: filesystem + parameters: + action: write + path: "out/{{ index }}.txt" + content: "{{ item }} of {{ queue_size }}" +``` + +Bound in the actions: `item`, `index`, `is_first`, `is_last`, `queue`, +`queue_size`, `parallel_queue_id`, `parent_task`. + +**Not bound in `on:`** — that expression generates the queue, so nothing +per-item exists while it runs. + +Two things about this construct differ from steps elsewhere: `on` must be +quoted (YAML 1.1 reads a bare `on` as the boolean `true`), and the nested +actions use `action:` rather than `tool:`. + +## The `$` spelling + +`{{ $item }}` works. `{{ $position }}` is a compile error: + +``` +unexpected char '$' at 3 +``` + +`$` is not Jinja syntax. It works for some names only because a preprocessing +step rewrites them before rendering, and that rewrite covers a fixed list: +`$item`, `$index`, `$is_first`, `$is_last`, `$iteration`, `$loop_id`, +`$loop_name`, `$loop_state`. + +**Prefer the bare spelling.** It is what the runtime actually resolves and what +every table above is written in. The one exception is `$loop_name` in a +`for_each` body, where the bare form is not bound. The inconsistency is issue +#474. + +## Nested loops + +An inner loop sees its own bindings and the enclosing loop's, including in the +inner loop's own iterable: ```yaml steps: - id: outer for_each: "{{ categories }}" - as: outer_loop + loop_name: outer_loop steps: - id: inner - for_each: "{{ items }}" + for_each: "{{ item.entries }}" # the outer loop's item steps: - id: process parameters: - # Access both loops category: "{{ $outer_loop.item }}" - item: "{{ item }}" # Current (inner) loop item + entry: "{{ item }}" # the inner loop's item ``` -## Troubleshooting +The key for naming a loop is `loop_name:`. A named loop's variables are +reached as `{{ $. }}`. + +> **Known defect:** that spelling runs — the pipeline above writes `A` and `B` +> — but `orchestrator validate` rejects it with `unexpected char '$' at 3`, +> because validation parses the raw text while the runtime rewrites `$` first. +> See issue #474. + +## One loop per step -### Common Issues +A step declares one loop construct. A step carrying two — `for_each` and +`while` together, say — is rejected: which one would win is decided by +declaration order inside the validator, and no engine agrees to that order. -1. **Unrendered Templates**: If you see `{{ item }}` in your output instead of the actual value, ensure: - - You're within a `for_each` loop context - - The variable name is spelled correctly - - You're using the correct template syntax +## `foreach` is not `for_each` -2. **Index Starting at 0**: Remember that `index` is zero-based. Use `position` for one-based numbering. +`foreach:` is recognised by the declarative engine's spec objects but is not +expanded by the control-flow compiler, so a `foreach` step either fails schema +validation or runs its body exactly once with nothing bound. Use `for_each`. +See issue #475. + +## Troubleshooting -3. **Filesystem Paths**: When using loop variables in file paths, ensure the values don't contain invalid characters for filenames. +**A template appears unrendered in the output.** The name is not bound where +you used it. Check the construct's table above, and check whether you are in a +source expression (`for_each:`, `create_parallel_queue.on`) rather than a body. -## Implementation Notes +**`index` starts at 0.** Use `position` for one-based numbering. -As of the latest update, loop variables are properly injected into the execution context during both compile-time loop expansion and runtime ForEachTask expansion. The variables are available in: -- Task parameters -- Filesystem operations -- Template rendering -- Nested dependencies \ No newline at end of file +**`{{ $something }}` fails to compile.** Use the bare spelling; see above. diff --git a/src/orchestrator/core/loop_contracts.py b/src/orchestrator/core/loop_contracts.py index eb4e3801..c1734690 100644 --- a/src/orchestrator/core/loop_contracts.py +++ b/src/orchestrator/core/loop_contracts.py @@ -1,145 +1,203 @@ """What each loop construct binds, and where those bindings reach. -Validation treated "inside a loop" as one boolean and one union of names. Two -things follow from that, both wrong. +Every table here was read off a *running pipeline*, not off the source and not +off `docs/loop_variables.md`. `tests/test_loop_runtime_parity.py` re-derives it +the same way on every run: each declared binding must render in a real +execution, and a name this module withholds must fail to render. A table that +cannot be reproduced by running the thing is a claim, not a contract -- which +is how the previous version of this file came to declare `foreach` a supported +alias of `for_each` when no engine expands it (#475). -**The source expression was validated in loop scope.** `_validate_object_templates` -computed `is_loop` from the step dictionary and passed it down to every child, -including the iterable itself:: +Two axes, both of which the first version got wrong. + +**Scope is per field, not per step.** `_validate_object_templates` computed one +boolean for the whole step, so the iterable was validated in the scope it +introduces:: - id: process - for_each: "{{ item.children }}" # accepted -- but no item exists yet + for_each: "{{ item.children }}" # no item exists yet parameters: text: "{{ item.name }}" # correct -The iterable must resolve before there is an item to bind. A reference to a -loop variable in it can never work, and validation said nothing. - -**Every construct got every name.** `while` does not bind `item`; only -`create_parallel_queue` binds `queue`. Accepting the union meant -``{{ is_last }}`` inside a `while` loop passed validation and then failed to -render -- the false negative #470 introduced while removing a false positive. - -The tables below were read from the runtime rather than from -`docs/loop_variables.md`, which claims names the runtime does not bind in -every construct: - -`for_each` / `foreach` - `ControlSystem._render_task_templates` registers `item`, `index`, - `is_first` and `is_last` from `metadata["loop_context"]`; - `ContextManager.loop_context` supplies `item`, `index` and `loop_id`. -`while` - `WhileLoopHandler` builds `iteration`, `index`, `is_first`, `position`, - `loop_state` and `loop_id`. -`create_parallel_queue` - `ParallelQueueTask.get_template_variables` adds `queue`, `queue_size`, - `parallel_queue_id` and `parent_task` alongside the per-item names. -`action_loop` - `ActionLoopContext` exposes iteration state and previous-result metadata, - but no item: an action loop is not iterating a collection. - -Each construct's `source_fields` are evaluated before the loop exists, so they -are validated in the enclosing scope. Everything else in the step is body, and -sees that construct's bindings. +#473 fixed that with a flat `source_fields` tuple, which then made the opposite +error: it declared `action_loop` and `until` sources, so the *body* of an +action loop -- the one place iteration state certainly exists -- was validated +outside the loop. `scopes` below is keyed by field path for that reason: +`create_parallel_queue.on` is evaluated before there is a queue while the rest +of that same object is per-item. + +**Bindings are per construct.** `while` binds no `item`; only +`create_parallel_queue` binds `queue`. + +Where the runtime leaks a name that has no meaning for the construct -- a +`for_each:` iterable can read `{{ index }}` and get `0` from a context that is +not its own -- this file withholds it and #477 tracks the leak. Rendering a +stale zero is not a binding; blessing it here would turn a runtime bug into a +documented feature. """ from __future__ import annotations -from dataclasses import dataclass -from typing import Any, Dict, FrozenSet, Optional, Tuple +from dataclasses import dataclass, field +from typing import Any, Dict, FrozenSet, Mapping, Optional, Tuple -#: Jinja's own loop object, bound by `{% for %}` inside a template rather than -#: by a pipeline construct. Jinja's undeclared-name analysis already accounts -#: for it; it is listed so that a step-level loop does not report it either. +#: Jinja's own loop object, bound by `{% for %}` *inside* a template rather +#: than by a pipeline construct. It belongs to no contract -- putting it in one +#: would make that construct's empty scopes non-empty, and a scope that is +#: never empty reads downstream as "some loop is in scope". It is in the union +#: below because the data-flow validator consults that to know `{{ loop.index }}` +#: names Jinja's counter and not a missing task. JINJA_LOOP: FrozenSet[str] = frozenset({"loop"}) +#: `{{ $name }}` is not Jinja -- `$` is a syntax error. It works only because +#: `UnifiedTemplateResolver._preprocess_dollar_variables` rewrites it first, +#: and that rewrite does not reach every render path: `{{ $item }}` resolves +#: while `{{ $position }}` raises `unexpected char '$'`. So the `$` spelling is +#: an observed list, not `"$" + every binding` (#474). +PRESTRIPPED_DOLLAR_NAMES: FrozenSet[str] = frozenset({ + "$item", "$index", "$is_first", "$is_last", + "$iteration", "$loop_id", "$loop_name", "$loop_state", +}) + @dataclass(frozen=True) class LoopContract: """One loop construct: what it binds, and which fields see the bindings.""" key: str - #: Fields evaluated before the loop exists. Validated in the outer scope. - source_fields: Tuple[str, ...] - #: Names available to the body, in the bare spelling. The runtime also - #: registers a `$`-prefixed alias for each. + #: Bare names the body can render, confirmed by execution. bindings: FrozenSet[str] - - def dollar_bindings(self) -> FrozenSet[str]: - return frozenset(f"${name}" for name in self.bindings) + #: Field paths relative to the loop step whose scope is *not* the body's. + #: Matched by longest prefix, so an entry for `create_parallel_queue.on` + #: narrows that one field while its siblings keep the body scope. + scopes: Mapping[str, FrozenSet[str]] = field(default_factory=dict) + #: Names available only in the `$` spelling. `$loop_name` renders inside a + #: `for_each` body where bare `loop_name` does not; the asymmetry is real + #: and is part of #474. + dollar_only: FrozenSet[str] = frozenset() + + def _spell(self, bare: FrozenSet[str]) -> FrozenSet[str]: + """Both spellings of `bare`. An empty scope stays empty: a set that is + never empty reads as "some loop is in scope" downstream, which turns + every out-of-scope reference into the wrong diagnostic.""" + if not bare: + return frozenset() + dollar = frozenset(f"${name}" for name in bare | self.dollar_only) + return bare | (dollar & PRESTRIPPED_DOLLAR_NAMES) def all_bindings(self) -> FrozenSet[str]: - return self.bindings | self.dollar_bindings() | JINJA_LOOP - - + """Everything the body may name, in both spellings.""" + return self._spell(self.bindings) + + def bindings_for(self, path: str) -> FrozenSet[str]: + """What `path` -- relative to the loop step -- may name. + + Unlisted paths are body. Longest prefix wins so that a narrowed field + does not narrow the object around it. + """ + best: Optional[str] = None + for candidate in self.scopes: + if path == candidate or path.startswith(f"{candidate}."): + if best is None or len(candidate) > len(best): + best = candidate + if best is None: + return self.all_bindings() + return self._spell(self.scopes[best]) + + +#: `for_each` body, observed: `item`, `index`, `is_first`, `is_last`, +#: `position`, `length`, `remaining`, `has_next`, `has_prev`, `loop_id` all +#: render real values, and `$loop_name` renders where bare `loop_name` errors. +#: `iteration` renders `None` -- a while-loop name leaking into a construct +#: that has no iteration count (#477) -- so it is not a binding. FOR_EACH = LoopContract( key="for_each", - source_fields=("for_each", "foreach"), - bindings=frozenset({"item", "index", "is_first", "is_last"}), -) - -#: `foreach` is the same construct under the spelling the declarative engine -#: accepts. One contract, two keys -- the compiler treats them identically, so -#: validation must too. -FOREACH = LoopContract( - key="foreach", - source_fields=FOR_EACH.source_fields, - bindings=FOR_EACH.bindings, + bindings=frozenset({ + "item", "index", "is_first", "is_last", "position", + "length", "remaining", "has_next", "has_prev", "loop_id", + }), + dollar_only=frozenset({"loop_name"}), + #: The iterable resolves before there is an item to bind. `{{ item }}` + #: there is an error at run time; the other names render only because a + #: zeroed context is in scope, which is the leak in #477. + scopes={"for_each": frozenset()}, ) +#: `while` body, observed: `iteration`, `index`, `is_first`, `position`, +#: `loop_id`, `loop_name`, `loop_state`. `item` renders `None` and `length`, +#: `remaining`, `has_next`, `has_prev` render collection values for a construct +#: with no collection (#477), so none of them are bindings. WHILE = LoopContract( key="while", - source_fields=("while", "until"), bindings=frozenset({ - "iteration", "index", "is_first", "position", "loop_state", "loop_id", + "iteration", "index", "is_first", "position", + "loop_id", "loop_name", "loop_state", }), + #: Unlike a `for_each` iterable, a `while` condition is re-evaluated every + #: iteration, against the context `WhileLoopHandler.should_continue` + #: assembles at `control_flow/loops.py:428` -- which holds exactly + #: `iteration` and `loop_state`. `until` is evaluated from that same + #: context a few lines later, so both conditions share the scope. + scopes={ + "while": frozenset({"iteration", "loop_state"}), + "until": frozenset({"iteration", "loop_state"}), + }, ) +#: `create_parallel_queue` body, observed: `item`, `index`, `is_first`, +#: `is_last`, `queue`, `queue_size`, `parallel_queue_id`, `parent_task`. CREATE_PARALLEL_QUEUE = LoopContract( key="create_parallel_queue", - source_fields=("create_parallel_queue",), bindings=frozenset({ - "item", "index", "queue", "queue_size", "is_first", "is_last", - "parallel_queue_id", "parent_task", + "item", "index", "is_first", "is_last", + "queue", "queue_size", "parallel_queue_id", "parent_task", }), + #: `on` generates the queue, so nothing per-item exists while it resolves. + #: The whole-object `source_fields` entry #473 used put the nested action + #: list outside the loop as well, which is the opposite error. + scopes={"create_parallel_queue.on": frozenset()}, ) +#: `action_loop` body, observed: `iteration`, `is_first`, `loop_id`, +#: `has_previous`, `total_duration`, `termination_reason`. An action loop +#: repeats actions rather than walking a collection, so `item`, `index` and +#: `position` are not bound -- they pass through unrendered. +#: +#: The `action_loop` key *is* the body: `_build_iteration_context` +#: (`control_flow/action_loop_handler.py:448`) builds iteration state before +#: the action list executes. Its `until` shares that context at line 404 -- +#: though nothing evaluates it today, which is #476. ACTION_LOOP = LoopContract( key="action_loop", - source_fields=("action_loop", "until"), bindings=frozenset({ - "loop_id", "iteration", "is_first", "has_previous", "total_duration", - "termination_reason", + "iteration", "is_first", "loop_id", + "has_previous", "total_duration", "termination_reason", }), ) LOOP_CONTRACTS: Dict[str, LoopContract] = { contract.key: contract - for contract in (FOR_EACH, FOREACH, WHILE, CREATE_PARALLEL_QUEUE, ACTION_LOOP) + for contract in (FOR_EACH, WHILE, CREATE_PARALLEL_QUEUE, ACTION_LOOP) } #: Every name any construct binds. Used where a construct is not known -- never #: as a substitute for the per-construct set, which is what made #: `{{ is_last }}` acceptable inside a `while` loop. -ALL_BINDINGS: FrozenSet[str] = frozenset().union( +ALL_BINDINGS: FrozenSet[str] = JINJA_LOOP.union( *(contract.all_bindings() for contract in LOOP_CONTRACTS.values()) ) -def contract_for(step: Any) -> Optional[LoopContract]: - """The loop contract a step declares, or None if it is not a loop. +def contracts_for(step: Any) -> Tuple[LoopContract, ...]: + """Every loop contract a step declares. - Checked in declaration order of `LOOP_CONTRACTS` so a step carrying more - than one loop key resolves deterministically. + More than one means the step is ambiguous. Returning them all instead of + silently taking the first lets the caller say so: which construct wins is + otherwise decided by this module's declaration order, which no engine + agrees to. """ if not isinstance(step, dict): - return None - for key, contract in LOOP_CONTRACTS.items(): - if key in step: - return contract - return None - - -def is_source_field(contract: Optional[LoopContract], field: str) -> bool: - """Whether `field` is evaluated before the loop's bindings exist.""" - return contract is not None and field in contract.source_fields + return () + return tuple( + contract for key, contract in LOOP_CONTRACTS.items() if key in step + ) diff --git a/src/orchestrator/core/template_globals.py b/src/orchestrator/core/template_globals.py index 9cddd59d..8696cd8f 100644 --- a/src/orchestrator/core/template_globals.py +++ b/src/orchestrator/core/template_globals.py @@ -118,7 +118,7 @@ def arity(self) -> str: #: real tables, and anything needing "a name some loop binds" -- rather than #: "a name *this* loop binds" -- reads the union below. #: -#: Prefer `loop_contracts.contract_for(step)` wherever the construct is known. +#: Prefer `loop_contracts.contracts_for(step)` wherever the construct is known. #: The union cannot tell `{{ is_last }}` in a `create_parallel_queue` from the #: same text in a `while` loop, and that distinction is the whole point. LOOP_VARIABLES: FrozenSet[str] = frozenset( diff --git a/src/orchestrator/validation/template_validator.py b/src/orchestrator/validation/template_validator.py index 2fbb5b44..791b3817 100644 --- a/src/orchestrator/validation/template_validator.py +++ b/src/orchestrator/validation/template_validator.py @@ -12,13 +12,13 @@ import logging import re -from typing import Any, Dict, FrozenSet, List, Optional, Set, Union +from typing import Any, Dict, FrozenSet, List, Optional, Set, Tuple, Union from dataclasses import dataclass -from jinja2 import Environment, TemplateSyntaxError, meta +from jinja2 import TemplateSyntaxError, meta from jinja2.sandbox import SandboxedEnvironment from ..core.runtime_context import BARE_RUNTIME_NAMES, RUNTIME_NAMESPACE -from ..core.loop_contracts import ALL_BINDINGS, contract_for, is_source_field +from ..core.loop_contracts import ALL_BINDINGS, LoopContract, contracts_for from ..core.template_globals import ( ALL_LOOP_VARIABLES, DOLLAR_LOOP_VARIABLES, @@ -35,6 +35,10 @@ #: does not populate it, and papering over that here would hide a real bug. PARAMETER_NAMESPACES = frozenset({"parameters", "inputs"}) +#: Where a step sits in a pipeline document. A loop is declared by a step, so +#: this is where a loop construct is looked for -- see `_validate_object_templates`. +_STEP_PATH = re.compile(r"steps\[\d+\]$") + def _binding_set(value: Union[bool, FrozenSet[str], None]) -> FrozenSet[str]: """Normalise the loop-scope argument to a set of names. @@ -600,8 +604,16 @@ def _validate_object_templates( used_variables: Set, undefined_variables: Set, loop_bindings: FrozenSet[str] = frozenset(), + loop_scope: Optional[Tuple[LoopContract, str, FrozenSet[str]]] = None, ): - """Recursively validate templates in an object.""" + """Recursively validate templates in an object. + + `loop_bindings` is what the enclosing loops bind. `loop_scope` is the + innermost loop construct still being walked, paired with the field + path reached inside it, because scope is a property of the field and + not of the step: a `create_parallel_queue`'s `on` resolves before any + item exists while the action list beside it runs per item. + """ if isinstance(obj, str): # Check if this contains templates if '{{' in obj or '{%' in obj: @@ -612,33 +624,60 @@ def _validate_object_templates( warnings.extend(result.warnings) used_variables.update(result.used_variables) undefined_variables.update(result.undefined_variables) - + elif isinstance(obj, dict): - # A loop's source expression is evaluated before the loop exists, - # so it does not see the loop's own bindings. Passing one boolean - # down to every field accepted `for_each: "{{ item.children }}"`, - # which can never resolve: there is no item until the iterable - # has been evaluated. - contract = contract_for(obj) - body_bindings = ( - loop_bindings | contract.all_bindings() if contract else loop_bindings - ) + # Only a *step* declares a loop. Matching any dict that happens to + # hold a loop key made `create_parallel_queue`'s own nested + # `action_loop` look like a second, separate loop, which replaced + # the queue's scope with the action loop's and let `{{ item }}` + # through in the `on` expression that generates the queue. + declared = contracts_for(obj) if _STEP_PATH.search(path) else () + if len(declared) > 1: + # Which construct wins is decided by declaration order in + # `loop_contracts`, and no engine agreed to that order. The + # step binds one set of names or another depending on an + # implementation detail, so it has no meaning to validate. + errors.append(TemplateValidationError( + template="", + error_type="ambiguous_loop_construct", + message=( + "Step declares more than one loop construct: " + + ", ".join(sorted(c.key for c in declared)) + ), + context_path=path, + suggestions=["Split these into separate steps"], + )) + if len(declared) == 1: + # Entering a loop. What the enclosing loops bind is kept + # separately rather than subtracted back out later: an inner + # `for_each` inside an outer one shares every name with it, so + # subtracting the inner contract's names would take the outer + # loop's `item` away from the inner iterable that is normally + # written from exactly that. + loop_scope = (declared[0], "", loop_bindings) + loop_bindings = loop_bindings | declared[0].all_bindings() for key, value in obj.items(): new_path = f"{path}.{key}" if path else key + child_bindings, child_scope = loop_bindings, loop_scope + if loop_scope is not None: + contract, prefix, enclosing = loop_scope + relative = f"{prefix}.{key}" if prefix else key + child_scope = (contract, relative, enclosing) + child_bindings = enclosing | contract.bindings_for(relative) self._validate_object_templates( value, context, step_ids, new_path, errors, warnings, used_variables, undefined_variables, - loop_bindings if is_source_field(contract, key) else body_bindings, + child_bindings, child_scope, ) - + elif isinstance(obj, list): for i, item in enumerate(obj): new_path = f"{path}[{i}]" self._validate_object_templates( item, context, step_ids, new_path, errors, warnings, used_variables, undefined_variables, - loop_bindings, + loop_bindings, loop_scope, ) def _is_step_result_reference(self, var_name: str, step_ids: List[str]) -> bool: diff --git a/tests/test_loop_contracts.py b/tests/test_loop_contracts.py index 24a75a66..9b8295c1 100644 --- a/tests/test_loop_contracts.py +++ b/tests/test_loop_contracts.py @@ -16,9 +16,16 @@ to render. That false negative arrived with #470, whose description claimed the opposite; the correction is on #469. -`core.loop_contracts` holds one table per construct, read from the runtime -rather than from `docs/loop_variables.md`, which claims names the runtime does -not bind in every construct. +#473 replaced the boolean with a flat tuple of "source fields" and made the +opposite error: it declared `action_loop` and `until` sources, which put the +*body* of an action loop outside the loop it is the body of, and it modelled +a `while` condition -- re-evaluated every iteration -- as if it resolved once +like a `for_each` iterable. Scope is now per field path. + +These are validator tests: they check that the validator applies +`core.loop_contracts`. Whether those tables are *true* is a separate question, +and asking it here would be circular -- both sides would come from the same +table. `tests/test_loop_runtime_parity.py` answers it by executing pipelines. """ import pytest @@ -29,9 +36,9 @@ CREATE_PARALLEL_QUEUE, FOR_EACH, LOOP_CONTRACTS, + PRESTRIPPED_DOLLAR_NAMES, WHILE, - contract_for, - is_source_field, + contracts_for, ) from orchestrator.validation.template_validator import TemplateValidator @@ -47,7 +54,7 @@ def _errors(step, extra_context=None): # --------------------------------------------------------------------------- -# Source expressions are outside the loop they introduce +# A source expression is outside the loop it introduces # --------------------------------------------------------------------------- def test_a_loop_variable_in_the_iterable_is_rejected(): @@ -74,14 +81,53 @@ def test_the_same_name_is_accepted_in_the_body(): assert kinds == [], kinds -@pytest.mark.parametrize("contract", sorted(set(LOOP_CONTRACTS.values()), key=lambda c: c.key)) -def test_every_construct_declares_its_source_fields(contract): - assert contract.source_fields, ( - f"{contract.key} declares no source field, so its iterable or " - f"condition would be validated in loop scope" - ) - for field in contract.source_fields: - assert is_source_field(contract, field) +# --------------------------------------------------------------------------- +# ...but a condition is not a source: it is re-evaluated per iteration +# --------------------------------------------------------------------------- + +def test_a_while_condition_reads_the_iteration_count(): + """`WhileLoopHandler.should_continue` assembles `iteration` before + evaluating the condition, every iteration. #473 called `while` a source + field and rejected the ordinary way of writing a bounded loop.""" + kinds = [k for k, _ in _errors( + {"id": "s", "while": "{{ iteration < 3 }}", "steps": []} + )] + assert kinds == [], kinds + + +def test_a_while_condition_does_not_see_the_whole_body_scope(): + """It sees exactly what `should_continue` puts in scope -- not `index` or + `position`, which the body has and the condition does not.""" + kinds = [k for k, _ in _errors( + {"id": "s", "while": "{{ position > 1 }}", "steps": []} + )] + assert "loop_variable_wrong_construct" in kinds, kinds + + +def test_an_action_loop_body_is_inside_its_own_loop(): + """The `action_loop` key *is* the body. Declaring it a source field put + iteration state out of reach of the only place it exists.""" + kinds = [k for k, _ in _errors({ + "id": "s", + "action_loop": [{"action": "noop", "parameters": {"v": "{{ iteration }}"}}], + "until": "{{ iteration >= 3 }}", + })] + assert kinds == [], kinds + + +def test_a_parallel_queue_source_is_outer_but_its_actions_are_not(): + """One object, two scopes: `on` builds the queue, the actions run per + item. A whole-object source field cannot express that.""" + found = _errors({ + "id": "s", + "create_parallel_queue": { + "on": "{{ item }}", + "action_loop": [{"action": "noop", "parameters": {"v": "{{ item }}"}}], + }, + }) + assert found == [ + ("loop_variable_outside_loop", "steps[0].create_parallel_queue.on") + ], found # --------------------------------------------------------------------------- @@ -156,6 +202,35 @@ def test_an_action_loop_binds_no_item(): assert "loop_variable_wrong_construct" in kinds, kinds +# --------------------------------------------------------------------------- +# A step declares one loop +# --------------------------------------------------------------------------- + +def test_a_step_declaring_two_loop_constructs_is_rejected(): + """Which one wins was decided by declaration order in `loop_contracts`, + an order no engine agreed to, so the step's bindings depended on an + implementation detail.""" + kinds = [k for k, _ in _errors( + {"id": "s", "for_each": "{{ rows }}", "while": "{{ go }}", "parameters": {}}, + {"rows": [1], "go": True}, + )] + assert "ambiguous_loop_construct" in kinds, kinds + + +def test_one_loop_construct_is_not_ambiguous(): + kinds = [k for k, _ in _errors( + {"id": "s", "for_each": "{{ rows }}", "parameters": {}}, {"rows": [1]} + )] + assert "ambiguous_loop_construct" not in kinds, kinds + + +def test_contracts_for_reports_every_construct_a_step_declares(): + both = contracts_for({"for_each": "x", "while": "y"}) + assert {c.key for c in both} == {"for_each", "while"} + assert contracts_for({"id": "s", "parameters": {}}) == () + assert contracts_for("not a step") == () + + # --------------------------------------------------------------------------- # Scope still behaves # --------------------------------------------------------------------------- @@ -189,6 +264,24 @@ def test_a_nested_loop_sees_both_constructs(): assert kinds == [], kinds +def test_an_inner_loop_source_still_sees_the_outer_loop(): + """The inner iterable resolves inside the outer iteration, so the outer + item is exactly what it is normally written from.""" + kinds = [k for k, _ in _errors( + { + "id": "outer", + "for_each": "{{ rows }}", + "steps": [{ + "id": "inner", + "for_each": "{{ item.children }}", + "parameters": {"t": "x"}, + }], + }, + {"rows": [1]}, + )] + assert kinds == [], kinds + + # --------------------------------------------------------------------------- # One declaration # --------------------------------------------------------------------------- @@ -211,27 +304,37 @@ def test_every_loop_step_key_has_a_contract(): ) -def test_foreach_is_the_same_construct_as_for_each(): - """The compiler treats them identically, so validation must too.""" - assert LOOP_CONTRACTS["foreach"].bindings == FOR_EACH.bindings - assert contract_for({"foreach": "x"}).bindings == FOR_EACH.bindings +def test_foreach_is_not_a_loop_construct(): + """#470 and #473 accepted `foreach` as an alias of `for_each` on the + strength of a comment. No engine expands it -- a `foreach` step fails + schema validation outright (#475) -- so validating it as a working loop + tells a reader the opposite of the truth.""" + assert "foreach" not in LOOP_CONTRACTS + assert contracts_for({"foreach": "{{ rows }}"}) == () -def test_a_step_with_no_loop_key_has_no_contract(): - assert contract_for({"id": "s", "parameters": {}}) is None - assert not is_source_field(None, "for_each") +def test_the_dollar_spelling_is_an_observed_list_not_a_derived_one(): + """`{{ $position }}` is a compile error while `{{ $item }}` resolves, + because the rewrite that makes `$` work does not reach every path (#474). + Deriving the `$` set as "`$` + every binding" would accept the first.""" + assert "$item" in FOR_EACH.all_bindings() + assert "$position" not in FOR_EACH.all_bindings() + assert "position" in FOR_EACH.all_bindings() + assert all(name.startswith("$") for name in PRESTRIPPED_DOLLAR_NAMES) -@pytest.mark.parametrize("contract", sorted(set(LOOP_CONTRACTS.values()), key=lambda c: c.key)) -def test_both_spellings_are_bound(contract): - """The runtime registers `item` and `$item` alike.""" - for name in contract.bindings: - assert f"${name}" in contract.all_bindings() +def test_a_narrowed_field_does_not_narrow_its_siblings(): + """`bindings_for` matches the longest prefix, so scoping + `create_parallel_queue.on` leaves the object around it alone.""" + assert CREATE_PARALLEL_QUEUE.bindings_for("create_parallel_queue.on") == frozenset() + assert "item" in CREATE_PARALLEL_QUEUE.bindings_for("create_parallel_queue.action_loop") + assert "item" in CREATE_PARALLEL_QUEUE.bindings_for("parameters") -def test_the_parallel_queue_binds_more_than_the_others(): - """A guard on the tables being genuinely different -- if every construct - ended up with the same set, the per-construct split would be decorative.""" +def test_the_tables_are_genuinely_different(): + """A guard on the per-construct split being real -- if every construct + ended up with the same set, splitting them would be decorative.""" assert CREATE_PARALLEL_QUEUE.bindings - WHILE.bindings assert WHILE.bindings - CREATE_PARALLEL_QUEUE.bindings assert FOR_EACH.bindings != WHILE.bindings + assert "item" not in ACTION_LOOP.bindings diff --git a/tests/test_loop_runtime_parity.py b/tests/test_loop_runtime_parity.py new file mode 100644 index 00000000..d3fd5f3a --- /dev/null +++ b/tests/test_loop_runtime_parity.py @@ -0,0 +1,231 @@ +"""The loop contracts are checked against a running pipeline, not against +themselves. + +`tests/test_loop_contracts.py` asks whether the validator agrees with +`core.loop_contracts`. Both sides of that question come from the same table, +so it proves the table is applied consistently and nothing at all about +whether the table is *true*. That gap is how #473 shipped a contract declaring +`foreach` an alias of `for_each` -- consistent everywhere, and false: no engine +expands it (#475). It is also how `action_loop`'s body came to be validated +outside the loop it is the body of. + +So this module executes pipelines. For every name a contract claims, a real +run must render it; for names withheld, a real run must leave them unrendered. +The table cannot drift from the runtime without a failure here. + +No model is involved: every pipeline writes files with the filesystem tool, so +these are deterministic and hermetic. +""" + +import asyncio +import re +from pathlib import Path + +import pytest + +from orchestrator.core.loop_contracts import ( + ACTION_LOOP, + CREATE_PARALLEL_QUEUE, + FOR_EACH, + WHILE, +) +from orchestrator.validation.template_validator import TemplateValidator +from tests.test_infrastructure import create_test_orchestrator + +pytestmark = [pytest.mark.contract, pytest.mark.e2e] + + +#: Each construct in the shape that actually runs, with `{probe}` where the +#: name under test is substituted and `{out}` for the run directory. The +#: shapes are not interchangeable and were established by execution: +#: `create_parallel_queue` needs the `action:` key beside its block, and the +#: actions inside it cannot use `tool:`. +BODY_PIPELINES = { + FOR_EACH.key: """ +id: probe_for_each +name: probe +steps: + - id: loop + for_each: "['A', 'B']" + steps: + - id: emit + tool: filesystem + action: write + parameters: + path: "{out}/probe.txt" + content: "<{{{{ {probe} }}}}>" +""", + WHILE.key: """ +id: probe_while +name: probe +steps: + - id: loop + while: "{{{{ iteration < 1 }}}}" + max_iterations: 2 + steps: + - id: emit + tool: filesystem + action: write + parameters: + path: "{out}/probe.txt" + content: "<{{{{ {probe} }}}}>" +""", + ACTION_LOOP.key: """ +id: probe_action_loop +name: probe +steps: + - id: loop + action_loop: + - action: filesystem + parameters: + action: write + path: "{out}/probe.txt" + content: "<{{{{ {probe} }}}}>" + until: "{{{{ iteration >= 1 }}}}" + max_iterations: 1 +""", + CREATE_PARALLEL_QUEUE.key: """ +id: probe_queue +name: probe +steps: + - id: loop + action: create_parallel_queue + create_parallel_queue: + "on": "['A']" + action_loop: + - action: filesystem + parameters: + action: write + path: "{out}/probe.txt" + content: "<{{{{ {probe} }}}}>" +""", +} + +CONTRACTS = {c.key: c for c in (FOR_EACH, WHILE, ACTION_LOOP, CREATE_PARALLEL_QUEUE)} + +#: Names each construct must NOT bind, chosen from what a *different* +#: construct binds so the check is about this construct rather than about the +#: name being unknown everywhere. +WITHHELD = { + FOR_EACH.key: ["queue_size", "termination_reason"], + WHILE.key: ["queue", "has_previous"], + ACTION_LOOP.key: ["item", "queue_size"], + CREATE_PARALLEL_QUEUE.key: ["iteration", "total_duration"], +} + + +def _run(pipeline: str, out: Path) -> str: + """Run a probe pipeline and return what reached the file. + + An empty string means the step never wrote, which is how an unresolved + reference surfaces here -- `create_parallel_queue` reports success while + skipping the write (#478), so the artifact is the evidence, not the + result payload. + """ + try: + asyncio.run(create_test_orchestrator().execute_yaml(pipeline, {})) + except Exception as exc: # a construct that cannot run is a real failure + return f"!{type(exc).__name__}: {exc}" + probe = out / "probe.txt" + return probe.read_text() if probe.exists() else "" + + +def _rendered(written: str, name: str) -> bool: + """Whether `{{ name }}` became a value rather than passing through.""" + if not written or written.startswith("!"): + return False + body = re.fullmatch(r"<(.*)>", written, re.S) + if body is None: + return False + value = body.group(1) + return value != "" and "{{" not in value + + +@pytest.mark.parametrize( + "key,name", + [(key, name) for key, c in CONTRACTS.items() for name in sorted(c.bindings)], +) +def test_every_declared_binding_renders_in_a_real_run(key, name, tmp_path): + """A contract that claims a name the runtime does not bind is a false + acceptance: validation passes and the pipeline fails to render.""" + written = _run(BODY_PIPELINES[key].format(out=tmp_path, probe=name), tmp_path) + assert _rendered(written, name), ( + f"{key} declares '{name}' but a real run produced {written!r}" + ) + + +@pytest.mark.parametrize( + "key,name", + [(key, name) for key, names in WITHHELD.items() for name in names], +) +def test_a_withheld_name_does_not_render(key, name, tmp_path): + """The other direction: a name the contract omits must genuinely be + absent, or omitting it is a false rejection of a working pipeline.""" + written = _run(BODY_PIPELINES[key].format(out=tmp_path, probe=name), tmp_path) + assert not _rendered(written, name), ( + f"{key} withholds '{name}' but a real run rendered it as {written!r}, " + f"so validation rejects a pipeline that works" + ) + + +@pytest.mark.parametrize( + "key,name", + [(key, name) for key, c in CONTRACTS.items() for name in sorted(c.bindings)], +) +def test_validation_accepts_every_name_the_runtime_binds(key, name, tmp_path): + """The two surfaces meet here: what runs must also validate.""" + result = TemplateValidator().validate_pipeline_templates( + {"id": "p", "steps": [{ + "id": "s", key: _minimal_construct_value(key), + "parameters": {"t": "{{ %s }}" % name}, + }]}, + {}, + ) + loop_errors = [ + e for e in result.errors + if e.error_type.startswith("loop_variable") + ] + assert not loop_errors, ( + f"{key} binds '{name}' at run time but validation rejected it: " + f"{[(e.error_type, e.message) for e in loop_errors]}" + ) + + +def _minimal_construct_value(key): + """The smallest value that makes a step declare `key`.""" + if key == ACTION_LOOP.key: + return [{"action": "noop"}] + if key == CREATE_PARALLEL_QUEUE.key: + return {"on": "['A']", "action_loop": [{"action": "noop"}]} + return "['A']" + + +def test_the_while_condition_can_read_its_own_iteration_count(): + """The condition is re-evaluated per iteration, so `iteration` is bound + there -- unlike a `for_each` iterable, which resolves once, before there + is anything to bind. #473 modelled both as the same kind of field and + rejected this pipeline, which runs.""" + result = TemplateValidator().validate_pipeline_templates( + {"id": "p", "steps": [{ + "id": "s", + "while": "{{ iteration < 3 }}", + "steps": [{"id": "b", "parameters": {"t": "x"}}], + }]}, + {}, + ) + assert result.is_valid, [(e.error_type, e.message) for e in result.errors] + + +def test_the_while_condition_stops_the_loop_at_the_iteration_it_names(tmp_path): + """And it is bound in the sense that matters: the count controls the run.""" + pipeline = BODY_PIPELINES[WHILE.key].replace( + 'while: "{{{{ iteration < 1 }}}}"', 'while: "{{{{ iteration < 3 }}}}"' + ).replace("max_iterations: 2", "max_iterations: 9").replace( + '"{out}/probe.txt"', '"{out}/{{{{ iteration }}}}.txt"' + ) + _run(pipeline.format(out=tmp_path, probe="iteration"), tmp_path) + written = sorted(p.name for p in tmp_path.glob("*.txt")) + assert written == ["0.txt", "1.txt", "2.txt"], ( + f"the condition names iteration < 3, so the body runs three times; " + f"got {written}" + )