The run's bare names, and one clock behind them (#104)

src/orchestrator/core/runtime_context.py

@@ -55,6 +55,16 @@
 
 EXECUTION_FIELDS: FrozenSet[str] = frozenset(EXECUTION_FIELD_NAMES)
 
+#: Names the runtime registers directly, without the `execution.` prefix.
+#: They render correctly today and were reported as undefined variables, which
+#: is the same false positive `execution.timestamp` had before it was declared.
+#: `timestamp` is the run's start time, so it must be *the same instant* as
+#: `execution.timestamp` -- it used to be a separate reading of the clock, in a
+#: different format, which is the divergence this module exists to end.
+BARE_RUNTIME_NAMES: FrozenSet[str] = frozenset(
+    {"pipeline_id", "execution_id", "timestamp"}
+)
+
 
 @dataclass(frozen=True)
 class RuntimeContext:

src/orchestrator/orchestrator.py

@@ -304,8 +304,14 @@ async def execute_pipeline(
 
         # What the run knows about itself. Created here, once, and read
         # everywhere else -- see core/runtime_context.py.
+        execution_namespace = execution_namespace_for(context)
+        self.template_manager.register_context("execution", execution_namespace)
+        # `{{ timestamp }}` is the same instant under a bare name.
+        # `TemplateManager._setup_base_context` seeds it from the clock when the
+        # manager is *constructed*, which is neither the run's start nor the
+        # same value as `execution.timestamp`; the run's own answer wins.
         self.template_manager.register_context(
-            "execution", execution_namespace_for(context)
+            "timestamp", execution_namespace["started_at"]
         )
 
         # Register all pipeline context (including inputs)

src/orchestrator/runtime/execution_state.py

@@ -15,6 +15,7 @@
 from enum import Enum
 import json
 import copy
+from ..core.runtime_context import execution_namespace_for
 
 logger = logging.getLogger(__name__)
 
@@ -299,7 +300,11 @@ def get_available_context(self) -> Dict[str, Any]:
         # Add system variables
         context['pipeline_id'] = self.pipeline_id
         context['execution_time'] = (datetime.now() - self.start_time).total_seconds()
-        context['timestamp'] = datetime.now().isoformat()
+        # The run's start time, not a fresh reading. `{{ timestamp }}` and
+        # `{{ execution.timestamp }}` name the same instant and disagreed:
+        # this returned local time without a zone while the run context
+        # returned UTC.
+        context['timestamp'] = execution_namespace_for(context)['started_at']
 
         return context
 

src/orchestrator/validation/data_flow_validator.py

@@ -16,7 +16,11 @@
 from dataclasses import dataclass, field
 from jinja2 import TemplateSyntaxError, Undefined, meta
 
-from ..core.runtime_context import EXECUTION_FIELDS, RUNTIME_NAMESPACE
+from ..core.runtime_context import (
+    BARE_RUNTIME_NAMES,
+    EXECUTION_FIELDS,
+    RUNTIME_NAMESPACE,
+)
 from ..core.template_sandbox import create_sandboxed_environment, pipeline_global_names
 
 logger = logging.getLogger(__name__)
@@ -546,6 +550,11 @@ def _validate_variable_reference(self,
         # call syntax attached, so `now()` was looked up as a *task id* and
         # reported as `Undefined task reference: 'now()'` on pipelines that run
         # correctly.
+        # Registered by the runtime under a bare name rather than under
+        # `execution.`; they render correctly and were reported undefined.
+        if base_var in BARE_RUNTIME_NAMES:
+            return {"valid": True, "type": "runtime_namespace"}
+
         if base_var.split("(", 1)[0] in pipeline_global_names():
             return {"valid": True, "type": "pipeline_global"}
 

src/orchestrator/validation/template_validator.py

@@ -17,7 +17,7 @@
 from jinja2 import Environment, TemplateSyntaxError, meta
 from jinja2.sandbox import SandboxedEnvironment
 
-from ..core.runtime_context import RUNTIME_NAMESPACE
+from ..core.runtime_context import BARE_RUNTIME_NAMES, RUNTIME_NAMESPACE
 from ..core.template_globals import find_global_misuse
 from ..core.template_sandbox import pipeline_global_names
 
@@ -411,7 +411,7 @@ def _validate_variables(
                 # 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:
+                if var_name == RUNTIME_NAMESPACE or var_name in BARE_RUNTIME_NAMES:
                     continue
 
                 # Check if variable is available in context

tests/test_runtime_context.py

@@ -29,6 +29,7 @@
 import pytest
 
 from orchestrator.core.runtime_context import (
+    BARE_RUNTIME_NAMES,
     EXECUTION_FIELDS,
     RUNTIME_NAMESPACE,
     RuntimeContext,
@@ -280,3 +281,72 @@ def test_the_run_the_template_sees_is_the_run_the_result_reports(tmp_path):
     assert rendered == reported, (
         f"the template saw run {rendered!r}, the result reports {reported!r}"
     )
+
+
+# ---------------------------------------------------------------------------
+# The same run, under its bare names
+# ---------------------------------------------------------------------------
+
+@pytest.mark.e2e
+@pytest.mark.parametrize("name", sorted(BARE_RUNTIME_NAMES))
+def test_a_bare_runtime_name_both_validates_and_renders(name, tmp_path):
+    """`{{ pipeline_id }}` is registered by the runtime and rendered correctly.
+
+    All three were reported as undefined variables -- the same false positive
+    `execution.timestamp` had before it was declared, just without the prefix.
+    """
+    pipeline = tmp_path / "p.yaml"
+    pipeline.write_text(_pipeline(f"{{{{ {name} }}}}"))
+
+    validated = _cli("validate", pipeline, tmp_path)
+    ran = _cli("run", pipeline, tmp_path)
+
+    assert ran.returncode == 0, f"{name} did not run: {ran.stdout[-400:]}"
+    assert (tmp_path / "out_0.txt").read_text().strip(), f"{name} rendered nothing"
+    assert validated.returncode == 0, (
+        f"{name} renders correctly but validation rejects it: "
+        f"{validated.stdout[-400:]}"
+    )
+
+
+@pytest.mark.e2e
+def test_the_bare_timestamp_is_the_run_s_own(tmp_path):
+    """`{{ timestamp }}` and `{{ execution.timestamp }}` are one instant.
+
+    They were two readings of two different clocks: `TemplateManager` seeds a
+    base context from `datetime.now()` when the *manager* is constructed --
+    neither the run's start nor the same value, and in local time without a
+    zone while the run context is UTC:
+
+        bare: 2026-08-03T08:07:56.022350
+        exec: 2026-08-03T12:07:56.022379+00:00
+
+    Four hours and 29 microseconds apart, in one run.
+    """
+    pipeline = tmp_path / "p.yaml"
+    pipeline.write_text(_pipeline("{{ timestamp }}|{{ execution.timestamp }}"))
+
+    result = _cli("run", pipeline, tmp_path)
+    assert result.returncode == 0, f"{result.stdout[-400:]}"
+
+    bare, prefixed = (tmp_path / "out_0.txt").read_text().split("|")
+    assert bare == prefixed, (
+        f"one run reported two start times: bare={bare!r} execution={prefixed!r}"
+    )
+
+
+@pytest.mark.e2e
+@pytest.mark.parametrize("name", ["current_timestamp", "current_date"])
+def test_a_name_the_runtime_does_not_provide_is_still_refused(name, tmp_path):
+    """Declaring three names must not wave through every bare name.
+
+    These two appear in the catalogue as often as the real ones and are
+    populated by nothing.
+    """
+    pipeline = tmp_path / "p.yaml"
+    pipeline.write_text(_pipeline(f"{{{{ {name} }}}}"))
+
+    assert _cli("run", pipeline, tmp_path).returncode != 0, (
+        f"{name} runs after all; it should be declared rather than refused"
+    )
+    assert _cli("validate", pipeline, tmp_path).returncode != 0