One run, one answer about itself (#104)

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.

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)
 
 

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()

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

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

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(

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: