diff --git a/docs/template_globals.md b/docs/template_globals.md index f47051df..47035daf 100644 --- a/docs/template_globals.md +++ b/docs/template_globals.md @@ -59,3 +59,27 @@ validate`, exit 2). A name that is not on this list is not a global. `{{ nowx() }}` is a typo and is refused at compile time rather than becoming an undefined value at run time. + +## The `execution` namespace + +What the run knows about itself. Computed once, when the run starts, +so every step of one run reports the same values -- naming an output +file after `execution.timestamp` gives one file, not one per step. + +| Field | Example | +|-|-| +| `execution.id` | `run-4f2a91c07e3b` | +| `execution.started_at` | `2026-01-15T14:30:45+00:00` | +| `execution.timestamp` | `2026-01-15T14:30:45+00:00` | +| `execution.date` | `2026-01-15` | +| `execution.time` | `14:30:45` | + +`timestamp` is `started_at` under its older name: the same instant, +not a second reading of the clock. Times are UTC, so stamps from two +machines are comparable and a run spanning a daylight-saving change +does not go backwards. + +The field list is closed. `{{ execution.strated_at }}` is a typo and +is refused at compile time; an open namespace would render it as an +empty string and report success. `pipeline`, `context` and `env` are +not namespaces -- nothing populates them. diff --git a/scripts/generate_template_globals_docs.py b/scripts/generate_template_globals_docs.py index 240d90af..e13bdf7d 100644 --- a/scripts/generate_template_globals_docs.py +++ b/scripts/generate_template_globals_docs.py @@ -16,10 +16,16 @@ import argparse import pathlib import sys +from datetime import datetime, timezone ROOT = pathlib.Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT / "src")) +from orchestrator.core.runtime_context import ( # noqa: E402 + EXECUTION_FIELD_NAMES, + RUNTIME_NAMESPACE, + RuntimeContext, +) from orchestrator.core.template_globals import GLOBAL_SPECS # noqa: E402 TARGET = ROOT / "docs" / "template_globals.md" @@ -83,6 +89,36 @@ def render() -> str: "typo and is refused at compile time rather than becoming an undefined\n" "value at run time.\n" ) + + lines.append(f"## The `{RUNTIME_NAMESPACE}` namespace\n") + lines.append( + "What the run knows about itself. Computed once, when the run starts,\n" + "so every step of one run reports the same values -- naming an output\n" + "file after `execution.timestamp` gives one file, not one per step.\n" + ) + # A fixed instant, not the current one: `--check` compares bytes, so a + # live clock here would make the page differ from itself on every run. + example = RuntimeContext( + id="run-4f2a91c07e3b", + started_at=datetime(2026, 1, 15, 14, 30, 45, tzinfo=timezone.utc), + ).as_template_namespace() + lines.append("| Field | Example |") + lines.append("|-|-|") + for name in EXECUTION_FIELD_NAMES: + lines.append(f"| `{RUNTIME_NAMESPACE}.{name}` | `{example[name]}` |") + lines.append("") + lines.append( + "`timestamp` is `started_at` under its older name: the same instant,\n" + "not a second reading of the clock. Times are UTC, so stamps from two\n" + "machines are comparable and a run spanning a daylight-saving change\n" + "does not go backwards.\n" + ) + lines.append( + "The field list is closed. `{{ execution.strated_at }}` is a typo and\n" + "is refused at compile time; an open namespace would render it as an\n" + "empty string and report success. `pipeline`, `context` and `env` are\n" + "not namespaces -- nothing populates them.\n" + ) return "\n".join(lines) diff --git a/src/orchestrator/control_systems/hybrid_control_system.py b/src/orchestrator/control_systems/hybrid_control_system.py index a681c7ea..07447ea9 100644 --- a/src/orchestrator/control_systems/hybrid_control_system.py +++ b/src/orchestrator/control_systems/hybrid_control_system.py @@ -1,6 +1,6 @@ """Hybrid control system that handles both model-based tasks and tool operations.""" -from typing import Any, Dict, Optional +from typing import Mapping, Any, Dict, Optional import logging import re from pathlib import Path @@ -56,6 +56,7 @@ ) from ..compiler.template_renderer import TemplateRenderer from ..runtime import RuntimeResolutionIntegration +from ..core.runtime_context import execution_namespace_for logger = logging.getLogger(__name__) @@ -322,27 +323,14 @@ def _prepare_template_context(self, context: Dict[str, Any]) -> TemplateResoluti return template_context - def _get_execution_metadata(self, context: Dict[str, Any]) -> Dict[str, Any]: - """Get execution metadata for templates.""" - from datetime import datetime - - # Try to get existing execution metadata - existing_execution = context.get("execution", {}) - if isinstance(existing_execution, dict) and existing_execution: - # If we have execution metadata with timestamp, use it - if "timestamp" in existing_execution: - return existing_execution - - # Generate new execution metadata - now = datetime.now() - return { - "timestamp": now.strftime("%Y-%m-%dT%H:%M:%S"), - "date": now.strftime("%Y-%m-%d"), - "time": now.strftime("%H:%M:%S"), - "iso_timestamp": now.isoformat(), - "pipeline_id": context.get("pipeline_id", "unknown"), - "execution_id": context.get("execution_id", "unknown"), - } + def _get_execution_metadata(self, context: Dict[str, Any]) -> Mapping[str, str]: + """What `{{ execution }}` resolves to for this run. + + This used to answer with its own field set -- `iso_timestamp`, + `pipeline_id`, `execution_id` -- in a format no other site used, so + which fields existed depended on which control system ran the step. + """ + return execution_namespace_for(context) def _extract_pipeline_parameters(self, pipeline_inputs: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]: """Extract original pipeline parameters for template access.""" @@ -590,13 +578,7 @@ def _build_template_context(self, context: Dict[str, Any]) -> Dict[str, Any]: template_context_obj = self._prepare_template_context(context) # Add execution metadata - execution_metadata = { - "execution": { - "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - "date": datetime.now().strftime("%Y-%m-%d"), - "time": datetime.now().strftime("%H:%M:%S"), - } - } + execution_metadata = {"execution": self._get_execution_metadata(context)} # Convert to flat dict and add execution metadata flat_context = template_context_obj.to_flat_dict() diff --git a/src/orchestrator/core/control_system.py b/src/orchestrator/core/control_system.py index 20f43c8e..ca1c503b 100644 --- a/src/orchestrator/core/control_system.py +++ b/src/orchestrator/core/control_system.py @@ -8,6 +8,7 @@ from .pipeline import Pipeline from .task import Task +from .runtime_context import execution_namespace_for class ControlAction(Enum): @@ -217,13 +218,10 @@ def _render_task_templates(self, task: Task, context: Dict[str, Any]) -> Task: for key, value in context["pipeline_context"].items(): template_manager.register_context(key, value) - # Add execution metadata - from datetime import datetime - template_manager.register_context("execution", { - "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - "date": datetime.now().strftime("%Y-%m-%d"), - "time": datetime.now().strftime("%H:%M:%S") - }) + # The run's own answer -- see core/runtime_context.py. + template_manager.register_context( + "execution", execution_namespace_for(context) + ) # Register other context values (including direct pipeline inputs like 'topic') # Skip only internal keys and already registered results diff --git a/src/orchestrator/core/runtime_context.py b/src/orchestrator/core/runtime_context.py new file mode 100644 index 00000000..186e4d2f --- /dev/null +++ b/src/orchestrator/core/runtime_context.py @@ -0,0 +1,117 @@ +"""What a run knows about itself: `{{ execution.timestamp }}` and friends. + +A pipeline can ask for facts about the run it is part of -- when it started, +which run it is. Seven places built that answer independently, in four +different formats:: + + orchestrator.py:307 %Y-%m-%d-%H:%M:%S + orchestrator.py:1424, :1994 .isoformat() + control_system.py:222 %Y-%m-%d %H:%M:%S + hybrid_control_system.py:594 %Y-%m-%d %H:%M:%S + declarative_engine.py:121 no timestamp at all -- `start_time` + +They did not merely disagree between engines. `_execute_level` rebuilt the +dict at *every level of the graph*, overwriting the one the run had already +registered, so a single run answered its own question differently each time:: + + step one -> 2026-08-02T20:01:55.182681 + step two -> 2026-08-02T20:01:55.184368 + +Two rows of a report, stamped two thousandths of a second apart, from one +run. Anything using the value to name an output file wrote several. + +So the value is computed once, when the run starts, and every later reader +gets that same value. `execution_namespace_for` is how they get it: the first +call on a run's context computes it and stores it there, every later call +returns what it finds. One run, one answer. + +The exposed fields are a closed set. `{{ execution.strated_at }}` is a typo, +not a field, and is refused at compile time rather than rendering as an empty +string into somebody's report -- which is what an open namespace would do. +""" + +from __future__ import annotations + +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Dict, FrozenSet, MutableMapping, Optional, Tuple + +#: Where the run's context dict carries its `RuntimeContext`. +CONTEXT_KEY = "_runtime_context" + +#: The name a template reaches it by: `{{ execution.timestamp }}`. +RUNTIME_NAMESPACE = "execution" + +#: Every field `{{ execution.* }}` may name. `test_runtime_context.py` asserts +#: this matches what `as_template_namespace` actually produces. +EXECUTION_FIELD_NAMES: Tuple[str, ...] = ( + "id", + "started_at", + "timestamp", + "date", + "time", +) + +EXECUTION_FIELDS: FrozenSet[str] = frozenset(EXECUTION_FIELD_NAMES) + + +@dataclass(frozen=True) +class RuntimeContext: + """One run's identity and start time. Immutable, created once.""" + + id: str + started_at: datetime + + @classmethod + def create(cls, execution_id: Optional[str] = None) -> "RuntimeContext": + """A context for a run starting now, in UTC. + + UTC rather than local time so two machines running the same pipeline + produce comparable stamps, and so a run spanning a daylight-saving + change does not go backwards. + """ + return cls( + id=execution_id or f"run-{uuid.uuid4().hex[:12]}", + started_at=datetime.now(timezone.utc), + ) + + @property + def timestamp(self) -> str: + """`started_at` under its older name. + + 59 of the catalogue's 61 `execution.*` references spell it this way. + It is the same instant, not a second reading of the clock. + """ + return self.started_at.isoformat() + + def as_template_namespace(self) -> Dict[str, str]: + """What `{{ execution }}` resolves to. + + Plain strings in a plain dict, because a run's context is checkpointed + as JSON: caching a `RuntimeContext` itself made every checkpointed run + fail with "Object of type RuntimeContext is not JSON serializable". + """ + return { + "id": self.id, + "started_at": self.started_at.isoformat(), + "timestamp": self.timestamp, + "date": self.started_at.strftime("%Y-%m-%d"), + "time": self.started_at.strftime("%H:%M:%S"), + } + + +def execution_namespace_for(context: MutableMapping[str, Any]) -> Dict[str, str]: + """The run's answer, computed on the first ask and reused after. + + Storing it back on the run's context is what makes "one run, one answer" + hold without threading an instance through every caller -- and every + caller already holds the run's context dict. + """ + cached = context.get(CONTEXT_KEY) + if isinstance(cached, dict) and EXECUTION_FIELDS <= set(cached): + return cached + + namespace = RuntimeContext.create(context.get("execution_id")).as_template_namespace() + context[CONTEXT_KEY] = namespace + return namespace diff --git a/src/orchestrator/engine/declarative_engine.py b/src/orchestrator/engine/declarative_engine.py index 4a0f9f80..48d858ef 100644 --- a/src/orchestrator/engine/declarative_engine.py +++ b/src/orchestrator/engine/declarative_engine.py @@ -10,6 +10,7 @@ from ..tools.base import default_registry from .advanced_executor import AdvancedTaskExecutor from .pipeline_spec import PipelineSpec, TaskSpec +from ..core.runtime_context import execution_namespace_for logger = logging.getLogger(__name__) @@ -118,15 +119,17 @@ def _initialize_context( "description": pipeline_spec.description, }, "config": pipeline_spec.config, - "execution": { - "start_time": datetime.now().isoformat(), - "engine_version": "1.0.0", - }, + "execution": {}, # filled in below, once the dict exists to carry it } # Add input values directly to context for easy template access context.update(inputs) + # One answer per run, shared with every other engine. This used to + # offer `start_time` and no `timestamp` at all, so `{{ execution.timestamp }}` + # rendered here and nowhere else. + context["execution"] = execution_namespace_for(context) + return context def _should_execute_step( diff --git a/src/orchestrator/orchestrator.py b/src/orchestrator/orchestrator.py index 91ac71f8..dd4e4e85 100644 --- a/src/orchestrator/orchestrator.py +++ b/src/orchestrator/orchestrator.py @@ -28,6 +28,7 @@ from .state.legacy_compatibility import LegacyStateManagerAdapter from .core.exceptions import PipelineExecutionError from .runtime import RuntimeResolutionIntegration +from .core.runtime_context import execution_namespace_for # Import checkpointing components for Issue #205 try: @@ -301,14 +302,11 @@ async def execute_pipeline( self.template_manager.register_context("pipeline_id", pipeline.id) self.template_manager.register_context("execution_id", execution_id) - # Add execution metadata - from datetime import datetime - execution_timestamp = datetime.now().strftime("%Y-%m-%d-%H:%M:%S") - self.template_manager.register_context("execution", { - "timestamp": execution_timestamp, - "date": datetime.now().strftime("%Y-%m-%d"), - "time": datetime.now().strftime("%H:%M:%S") - }) + # What the run knows about itself. Created here, once, and read + # everywhere else -- see core/runtime_context.py. + self.template_manager.register_context( + "execution", execution_namespace_for(context) + ) # Register all pipeline context (including inputs) for key, value in pipeline.context.items(): @@ -1419,13 +1417,10 @@ async def _execute_level( if step_id not in task_context: task_context[step_id] = result - # Add execution metadata for templates - from datetime import datetime - task_context["execution"] = { - "timestamp": datetime.now().isoformat(), - "date": datetime.now().strftime("%Y-%m-%d"), - "time": datetime.now().strftime("%H:%M:%S"), - } + # The run's own answer, not a fresh reading of the clock. + # Rebuilding it here is what made two steps of one run report + # timestamps milliseconds apart. + task_context["execution"] = execution_namespace_for(context) # Ensure pipeline parameters are directly accessible if isinstance(pipeline.context, dict): @@ -1989,13 +1984,8 @@ async def _expand_for_each_task( if step_id not in loop_context: loop_context[step_id] = result - # Add execution metadata - from datetime import datetime - loop_context["execution"] = { - "timestamp": datetime.now().isoformat(), - "date": datetime.now().strftime("%Y-%m-%d"), - "time": datetime.now().strftime("%H:%M:%S"), - } + # Same instant as every other step, including across iterations. + loop_context["execution"] = execution_namespace_for(context) # Process each step in the loop body for step_def in for_each_task.loop_steps: diff --git a/src/orchestrator/validation/data_flow_validator.py b/src/orchestrator/validation/data_flow_validator.py index 50aab05b..81c97f8d 100644 --- a/src/orchestrator/validation/data_flow_validator.py +++ b/src/orchestrator/validation/data_flow_validator.py @@ -16,6 +16,7 @@ from dataclasses import dataclass, field from jinja2 import TemplateSyntaxError, Undefined, meta +from ..core.runtime_context import EXECUTION_FIELDS, RUNTIME_NAMESPACE from ..core.template_sandbox import create_sandboxed_environment, pipeline_global_names logger = logging.getLogger(__name__) @@ -508,10 +509,26 @@ def _validate_variable_reference(self, if base_var in pipeline_inputs: return {"valid": True, "type": "pipeline_input"} - # Runtime namespaces injected by the executor rather than by the - # pipeline author. - if base_var in ("execution", "pipeline", "context", "env"): - return {"valid": True, "type": "runtime_namespace"} + # The one namespace the runtime injects. `pipeline`, `context` and + # `env` were accepted alongside it and are populated by nothing -- the + # template validator rejected them anyway, so the permissiveness here + # was invisible rather than harmless. They are gone; a pipeline naming + # them now gets a reason instead of an "undefined variable". + if base_var == RUNTIME_NAMESPACE: + field = parts[1] if len(parts) > 1 else None + if field is None or field in EXECUTION_FIELDS: + return {"valid": True, "type": "runtime_namespace"} + return { + "valid": False, + "error_type": "unknown_execution_field", + "message": ( + f"'execution.{field}' is not a field of the run context. " + f"Available: {', '.join(sorted(EXECUTION_FIELDS))}" + ), + "suggestions": self._suggest_similar_names( + field, sorted(EXECUTION_FIELDS) + ), + } # Check for task output references if base_var in task_schemas: diff --git a/src/orchestrator/validation/template_validator.py b/src/orchestrator/validation/template_validator.py index dd81b11c..39c8f6b3 100644 --- a/src/orchestrator/validation/template_validator.py +++ b/src/orchestrator/validation/template_validator.py @@ -17,6 +17,7 @@ from jinja2 import Environment, TemplateSyntaxError, meta from jinja2.sandbox import SandboxedEnvironment +from ..core.runtime_context import RUNTIME_NAMESPACE from ..core.template_globals import find_global_misuse from ..core.template_sandbox import pipeline_global_names @@ -397,6 +398,15 @@ def _validate_variables( if var_name in pipeline_global_names(): continue + # The run's own context. Populated by the runtime, so it is + # not an undefined variable -- `{{ execution.timestamp }}` is + # used by 32 catalogue pipelines, every one of which ran + # correctly and failed validation. Which *fields* it offers is + # checked by the data-flow validator, which sees the whole + # dotted reference; this only sees the base name. + if var_name == RUNTIME_NAMESPACE: + continue + # Check if variable is available in context if var_name not in available_context: undefined_variables.add(var_name) diff --git a/tests/test_runtime_context.py b/tests/test_runtime_context.py new file mode 100644 index 00000000..9a0e46f9 --- /dev/null +++ b/tests/test_runtime_context.py @@ -0,0 +1,282 @@ +"""One run, one answer about itself. + +`{{ execution.timestamp }}` had seven independent implementations in four +formats, and `_execute_level` rebuilt it at every level of the graph, so a +single run answered its own question differently each time: + + step one -> 2026-08-02T20:01:55.182681 + step two -> 2026-08-02T20:01:55.184368 + +Meanwhile `validate` rejected the expression outright -- 59 references across +32 catalogue pipelines, every one of which ran correctly and failed +validation. The namespace was simultaneously too permissive (the data-flow +validator accepted `execution.anything`, plus `pipeline`, `context` and `env`, +none of which anything populates) and, through the template validator, too +strict. + +These tests pin the contract from both sides: what the namespace offers, that +it offers the same thing throughout a run, and that everything else is +refused. +""" + +import json +import os +import subprocess +import sys +from datetime import timezone +from pathlib import Path + +import pytest + +from orchestrator.core.runtime_context import ( + EXECUTION_FIELDS, + RUNTIME_NAMESPACE, + RuntimeContext, + execution_namespace_for, +) + +pytestmark = [pytest.mark.contract] + +REPO = Path(__file__).resolve().parent.parent + + +def _cli(command, pipeline, cwd, *extra): + env = dict(os.environ) + env["PYTHONPATH"] = str(REPO / "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", command, str(pipeline), *extra], + cwd=str(cwd), env=env, capture_output=True, text=True, timeout=300, + ) + + +def _pipeline(*expressions): + steps = "\n".join( + f""" - id: step_{i} + tool: filesystem + action: write + parameters: + path: "./out_{i}.txt" + content: "{expression}" +""" + + (f" dependencies:\n - step_{i - 1}\n" if i else "") + for i, expression in enumerate(expressions) + ) + return f"id: rc\nname: RC\nsteps:\n{steps}" + + +# --------------------------------------------------------------------------- +# The schema +# --------------------------------------------------------------------------- + +def test_the_declared_fields_are_the_fields_produced(): + """A schema the validator trusts must match what the runtime renders. + + If they disagree, validation either rejects a field that works or accepts + one that does not -- the two failure modes this whole namespace had. + """ + produced = set(RuntimeContext.create("probe").as_template_namespace()) + assert produced == EXECUTION_FIELDS, ( + f"declared_only={sorted(EXECUTION_FIELDS - produced)}, " + f"produced_only={sorted(produced - EXECUTION_FIELDS)}" + ) + + +def test_the_run_id_is_carried_through(): + assert RuntimeContext.create("run-77").as_template_namespace()["id"] == "run-77" + + +def test_a_run_without_an_id_still_gets_one(): + namespace = RuntimeContext.create(None).as_template_namespace() + assert namespace["id"] + + +def test_started_at_is_utc(): + """Local time makes two machines' stamps incomparable, and goes backwards + across a daylight-saving change.""" + assert RuntimeContext.create("x").started_at.tzinfo is timezone.utc + assert RuntimeContext.create("x").as_template_namespace()["timestamp"].endswith( + "+00:00" + ) + + +def test_timestamp_is_started_at_under_its_older_name(): + """The same instant read once, not the clock read twice.""" + namespace = RuntimeContext.create("x").as_template_namespace() + assert namespace["timestamp"] == namespace["started_at"] + + +def test_the_namespace_survives_json(): + """A run's context is checkpointed as JSON. + + Caching the `RuntimeContext` object itself made every checkpointed run + fail with "Object of type RuntimeContext is not JSON serializable", which + is the sort of thing that only shows up when a run is long enough to + checkpoint. + """ + context = {"execution_id": "abc"} + execution_namespace_for(context) + json.dumps(context) # must not raise + + +# --------------------------------------------------------------------------- +# One run, one answer +# --------------------------------------------------------------------------- + +def test_asking_twice_gives_the_same_answer(): + context = {"execution_id": "abc"} + assert execution_namespace_for(context) == execution_namespace_for(context) + + +def test_separate_runs_get_separate_answers(): + first = execution_namespace_for({"execution_id": "a"}) + second = execution_namespace_for({"execution_id": "b"}) + assert first["id"] != second["id"] + + +@pytest.mark.e2e +def test_every_step_of_a_run_reports_the_same_timestamp(tmp_path): + """The defect, end to end. + + Two steps, one run. Before this, `_execute_level` rebuilt the namespace + per level and they differed by milliseconds -- enough for anything naming + an output file by timestamp to write two. + """ + pipeline = tmp_path / "p.yaml" + pipeline.write_text( + _pipeline("{{ execution.timestamp }}", "{{ execution.timestamp }}") + ) + + result = _cli("run", pipeline, tmp_path) + assert result.returncode == 0, f"{result.stdout[-800:]}{result.stderr[-800:]}" + + first = (tmp_path / "out_0.txt").read_text() + second = (tmp_path / "out_1.txt").read_text() + assert first, "the first step wrote nothing" + assert first == second, ( + f"one run reported two different start times: {first!r} then {second!r}" + ) + + +# --------------------------------------------------------------------------- +# What validation accepts, the runtime renders -- and the reverse +# --------------------------------------------------------------------------- + +@pytest.mark.e2e +@pytest.mark.parametrize("field", sorted(EXECUTION_FIELDS)) +def test_every_declared_field_both_validates_and_renders(field, tmp_path): + """A field in the schema must work all the way to the file. + + `execution.timestamp` used to run correctly and fail validation; the + point of a shared schema is that the two cannot disagree again. + """ + pipeline = tmp_path / "p.yaml" + pipeline.write_text(_pipeline(f"{{{{ execution.{field} }}}}")) + + validated = _cli("validate", pipeline, tmp_path) + ran = _cli("run", pipeline, tmp_path) + + assert validated.returncode == 0, ( + f"execution.{field} is in the schema and was rejected: " + f"{validated.stdout[-500:]}" + ) + assert ran.returncode == 0, f"execution.{field} did not run: {ran.stdout[-500:]}" + assert (tmp_path / "out_0.txt").read_text().strip(), ( + f"execution.{field} rendered as nothing" + ) + + +@pytest.mark.e2e +@pytest.mark.parametrize( + "expression", + [ + # A typo in a real field. Rendering this as an empty string would put + # a blank where a timestamp belongs and report success. + "{{ execution.strated_at }}", + "{{ execution.bogus }}", + # Namespaces the data-flow validator used to accept, populated by + # nothing at all. + "{{ pipeline.name }}", + "{{ context.foo }}", + "{{ env.HOME }}", + ], +) +def test_what_cannot_render_is_refused(expression, tmp_path): + pipeline = tmp_path / "p.yaml" + pipeline.write_text(_pipeline(expression)) + + validated = _cli("validate", pipeline, tmp_path) + ran = _cli("run", pipeline, tmp_path) + + assert ran.returncode != 0, f"{expression} ran; the runtime populates it after all" + assert validated.returncode != 0, ( + f"{expression} cannot render but validates, so the failure waits until " + f"run time" + ) + + +@pytest.mark.parametrize( + "reference", ["pipeline.name", "context.foo", "env.HOME", "execution.bogus"] +) +def test_the_data_flow_validator_refuses_them_on_its_own(reference): + """Asked directly, not through a whole pipeline. + + Through the CLI these are rejected either way, because the *template* + validator does not know the names -- so re-adding them here breaks no + end-to-end test, and the permissiveness sat unnoticed for exactly that + reason. This asks the line that changed. + """ + from orchestrator.validation.data_flow_validator import DataFlowValidator + + result = DataFlowValidator()._validate_variable_reference( + reference, + task_id="some_task", + parameter_name="content", + task_schemas={}, + pipeline_inputs={}, + ) + assert result["valid"] is False, ( + f"{reference} is accepted as a runtime namespace but nothing populates it" + ) + + +def test_an_unknown_field_names_the_ones_that_exist(): + """An error that lists the alternatives is the difference between a fix + and a guess.""" + from orchestrator.validation.data_flow_validator import DataFlowValidator + + result = DataFlowValidator()._validate_variable_reference( + "execution.strated_at", + task_id="some_task", + parameter_name="content", + task_schemas={}, + pipeline_inputs={}, + ) + assert result["valid"] is False + assert "started_at" in result["message"] + + +def test_the_namespace_name_is_stated_once(): + assert RUNTIME_NAMESPACE == "execution" + + +@pytest.mark.e2e +def test_the_run_the_template_sees_is_the_run_the_result_reports(tmp_path): + """`{{ execution.id }}` and the returned result must name one run. + + Two identifiers for the same execution would make a trace impossible to + follow across the boundary -- the artifact says one thing, the result + another. + """ + pipeline = tmp_path / "p.yaml" + pipeline.write_text(_pipeline("{{ execution.id }}")) + + result = _cli("run", pipeline, tmp_path) + assert result.returncode == 0, f"{result.stdout[-800:]}{result.stderr[-800:]}" + + rendered = (tmp_path / "out_0.txt").read_text().strip() + reported = json.loads(result.stdout[result.stdout.index("{"):])["execution_id"] + assert rendered == reported, ( + f"the template saw run {rendered!r}, the result reports {reported!r}" + )