validate_config: refuse a temporal companion without a resolvable clock at submission (issue #472)

src/zagg/client.py

@@ -440,8 +440,9 @@ def from_config(
         ----------
         config : PipelineConfig, dict, or str
             A loaded :class:`~zagg.config.PipelineConfig`, a plain config
-            dict, or a path to a YAML config file. Dicts and paths are
-            validated on load.
+            dict, or a path to a YAML config file. **Every** shape is
+            cross-validated here (issue #472), including an already-built
+            ``PipelineConfig`` that was mutated after ``default_config``.
         shardmap : dict or str, optional
             A loaded ShardMap manifest dict or a path to its JSON. Falls back
             to the config's ``catalog:`` key. The map's grid signature must
@@ -479,7 +480,6 @@ def from_config(
             config = load_config(config)
         elif isinstance(config, dict):
             config = load_config_from_dict(config)
-            validate_config(config)
 
         # v1 scope gate: the spatial point path only. The other pipelines
         # already run through agg()/zagg.notebook.run; refusing here beats a
@@ -502,6 +502,17 @@ def from_config(
                 "zagg.notebook.run until the v2 transport (issue #327)"
             )
 
+        # Cross-validate EVERY input shape, not just the dict/path ones (issue
+        # #472): a ``PipelineConfig`` from ``default_config`` validates once, at
+        # build time, so a notebook that then grafts another template's
+        # ``aggregation.variables`` onto it (the 02_write graft) dispatched a
+        # config nothing had re-checked — and the error surfaced one invoke per
+        # shard later, fleet-side. Runs after the scope gates above so an
+        # out-of-scope pipeline still gets its "use agg" pointer rather than a
+        # config error for a facade it was never going to run through, and
+        # before the shard map / store resolution so nothing touches AWS first.
+        validate_config(config)
+
         if isinstance(shardmap, dict):
             catalog_data = shardmap
             if not all(k in catalog_data for k in ("shard_keys", "granules", "grid_signature")):

src/zagg/config.py

@@ -1960,6 +1960,7 @@ def _validate_temporal_producer(name: str, meta: dict, config=None) -> None:
     ``validate_config`` path passes it.
     """
     from zagg.time_axis import (
+        TOC_NO_CLOCK_ERROR,
         TOC_PER_CELL_FUNCTIONS,
         TOC_PRODUCING_FUNCTIONS,
         TOC_SHAPE_PER_CELL,
@@ -1983,13 +1984,26 @@ def _validate_temporal_producer(name: str, meta: dict, config=None) -> None:
             f"sibling, and a digest kernel's channel is not a dense per-cell array "
             f"(spec §8.2/§8.3)"
         )
+    # The clock cross-check runs through the same resolver the worker encodes
+    # with (``toc_source``: ``output.time_source``, falling back to a
+    # continuous-scale ``output.windowing`` block), and raises the worker's
+    # exact message so the two seams read identically (issue #472) — the
+    # worker's copy stays as defense in depth. Two limits on that parity, both
+    # named here because this is where a reader will look (fold review):
+    #   * this seam is reachable only from ``validate_config``'s spatial,
+    #     non-raster branch (the aggregation-variable loop, after both early
+    #     returns), while ``_toc_word_column`` is ungated. Benign today —
+    #     ``calculate_cell_statistics`` is only reached from the spatial point
+    #     path, so no other branch can produce toc words — but a raster or
+    #     event toc path would inherit the gap silently.
+    #   * the resolver is shared, but the two clock declarations validate their
+    #     COLUMN against different sets: ``_validate_time_source`` accepts a
+    #     broadcast/segment-level column (GEDI's shot-rate clock) where
+    #     ``_validate_windowing`` refuses one (segment-rate window membership
+    #     is unsupported), so the fallback is not interchangeable with the
+    #     explicit block.
     if config is not None and toc_source(config) is None:
-        raise ValueError(
-            f"Variable '{name}': 'temporal' requires the store's per-observation clock — "
-            f"declare output.time_source {{field, epoch, scale, units}} (or an "
-            f"output.windowing block on a continuous scale, which it falls back to). "
-            f"Without it there is no column to encode toc words from (issue #410)"
-        )
+        raise ValueError(f"Variable '{name}': {TOC_NO_CLOCK_ERROR}")
 
 
 def _validate_output_kind(name: str, meta: dict, config=None) -> None:

src/zagg/processing/aggregate.py

@@ -51,15 +51,13 @@ def _checked_toc_source(columns, config) -> dict:
     a per-cell namespace or the pooled column dict can be checked without
     gathering anything.
     """
-    from zagg.time_axis import toc_source
+    from zagg.time_axis import TOC_NO_CLOCK_ERROR, toc_source
 
     source = toc_source(config)
     if source is None:
-        raise ValueError(
-            "a field declares a temporal companion but the store has no per-observation "
-            "clock — declare output.time_source {field, epoch, scale, units} (spec §8.3, "
-            "issue #410)"
-        )
+        # Defense in depth: validate_config refuses this at submission with the
+        # same single-sourced message (issue #472).
+        raise ValueError(TOC_NO_CLOCK_ERROR)
     if source["field"] not in columns:
         raise ValueError(
             f"output.time_source.field {source['field']!r} is not in the cell data "

src/zagg/runner.py

@@ -47,6 +47,7 @@
     get_store_path,
     get_sweep,
     get_windowing,
+    validate_config,
 )
 from zagg.dispatch import (
     BENIGN_ERRORS,
@@ -339,6 +340,15 @@ def agg(
         Summary with keys: ``total_cells``, ``cells_with_data``,
         ``cells_error``, ``total_obs``, ``wall_time_s``, ``store_path``.
     """
+    # The single validation choke point (issue #485): every submission path —
+    # Run.from_config's v1 facade, zagg.notebook.run's wrapper, and direct agg
+    # callers on any backend or pipeline kind — cross-validates here, so a
+    # PipelineConfig mutated after load_config's own call (the issue #472
+    # graft) is refused before any catalog read, store touch, or invoke.
+    # validate_config branches on pipeline kind itself, so the temporal and
+    # raster paths get their own (smaller) checks rather than a false refusal.
+    validate_config(config)
+
     # Pipeline kind picks the strategy (issue #12, Phase 5). The strategy seam
     # is dispatch-level: the spatial path is the existing code, moved verbatim
     # into SpatialStrategy so its behavior/output stays byte-identical; the

src/zagg/time_axis.py

@@ -91,6 +91,22 @@
 #: grid is meant to imply. Window ROUTING tolerates that; a word claiming
 #: nanosecond exactness cannot.
 TOC_SOURCE_SCALES = ("gps", "tai")
+#: The refusal for a ``temporal:`` companion whose clock does not resolve —
+#: :func:`toc_source` returns ``None``. Single-sourced (issue #472) so the
+#: submission seam (``config._validate_temporal_producer``) and the worker
+#: seam (``processing.aggregate._toc_word_column``, defense in depth) read
+#: identically: the first fleet failure burned one invoke per shard on an
+#: error fully determinable from the config dict alone. Both remedies are named
+#: in the text: on a windowed store the fallback is the only correct one, since
+#: ``_validate_time_source`` refuses an explicit block that disagrees with
+#: ``output.windowing`` — a hint in a code comment is invisible to the person
+#: reading the traceback (fold review).
+TOC_NO_CLOCK_ERROR = (
+    "a field declares a temporal companion but the store has no per-observation "
+    "clock — declare output.time_source {field, epoch, scale, units}, or an "
+    "output.windowing block on a continuous scale, which it falls back to "
+    "(spec §8.3, issue #410)"
+)
 #: The §8 word-grammar citation — a grammar REVISION token in the ecosystem's
 #: {name}/{major} style (``zagg-ragged/1``, ``morton-hive/2``), never a
 #: documentation URL or a stamp of the writer's installed mortie: store bytes
@@ -121,6 +137,7 @@
     "TOC_EPOCH",
     "TOC_FIELD_SHAPES",
     "TOC_GRAMMAR",
+    "TOC_NO_CLOCK_ERROR",
     "TOC_PER_CELL_FUNCTIONS",
     "TOC_PRODUCING_FUNCTIONS",
     "TOC_SHAPE_COORDINATE",

tests/test_client.py

@@ -6,9 +6,11 @@
 surfacing, and the worker-invoke post-run tail.
 """
 
+import copy
 import errno
 import importlib
 import json
+import re
 import sys
 import threading
 import time
@@ -229,6 +231,58 @@ def test_windowed_config_refused(self, catalog):
             Run.from_config(cfg, shardmap=catalog, store=_STORE)
 
 
+class TestSubmissionValidation:
+    """A mutated ``PipelineConfig`` is cross-validated at the submission seam.
+
+    ``from_config`` used to validate only its dict and path inputs, so the
+    observed failure (issue #472) had nothing re-check the ``02_write`` graft:
+    ``default_config`` validated the hive base template, the notebook then
+    grafted ``atl03_tdigest_located_healpix``'s ``temporal:`` variables onto it
+    without ``output.time_source``, and the config error only surfaced one
+    Lambda invoke per shard later, in the worker's refusal.
+    """
+
+    def _graft(self):
+        base = default_config("atl03_tdigest_healpix_hive", validate=False)
+        located = default_config("atl03_tdigest_located_healpix", validate=False)
+        base.aggregation["variables"] = copy.deepcopy(located.aggregation["variables"])
+        return base, located
+
+    def test_grafted_config_refused_before_any_invoke(self, catalog):
+        from zagg.time_axis import TOC_NO_CLOCK_ERROR
+
+        cfg, _ = self._graft()
+        stub = StubLambdaClient()
+        with pytest.raises(ValueError, match=re.escape(TOC_NO_CLOCK_ERROR)):
+            Run.from_config(
+                cfg,
+                shardmap=catalog,
+                store=_STORE,
+                function_name="process-shard-test",
+                lambda_client=stub,
+                source_credentials=_CREDS,
+            )
+        assert stub.events == []  # refused at submission, nothing dispatched
+
+    def test_grafted_config_with_its_clock_constructs(self, catalog):
+        # Positive control: grafting the clock block and its column too is the
+        # correct form, and it still builds a Run (no false refusal).
+        cfg, located = self._graft()
+        cfg.output["time_source"] = dict(located.output["time_source"])
+        field = cfg.output["time_source"]["field"]
+        cfg.data_source["variables"][field] = located.data_source["variables"][field]
+        cfg.output["grid"] = dict(_ATL06_SIG)
+        run = Run.from_config(
+            cfg,
+            shardmap=catalog,
+            store=_STORE,
+            function_name="process-shard-test",
+            lambda_client=StubLambdaClient(),
+            source_credentials=_CREDS,
+        )
+        assert len(run) == 3
+
+
 # -- dispatch fan-out --------------------------------------------------------
 
 

tests/test_config.py

@@ -1,6 +1,7 @@
 """Tests for the YAML pipeline configuration system."""
 
 import json
+import re
 from dataclasses import asdict
 
 import numpy as np
@@ -9,6 +10,7 @@
 
 from zagg.config import (
     PipelineConfig,
+    _segment_variable_names,
     _validate_output_kind,
     default_config,
     evaluate_expression,
@@ -32,6 +34,7 @@
     validate_config,
 )
 from zagg.processing import calculate_cell_statistics
+from zagg.time_axis import TOC_NO_CLOCK_ERROR
 
 # ---------------------------------------------------------------------------
 # Fixtures
@@ -2404,10 +2407,11 @@ def test_shape_and_reducer_must_agree(self):
 
     def test_temporal_requires_the_stores_clock(self):
         # Without output.time_source (or a continuous-scale windowing block to
-        # fall back to) there is no column to encode words from.
+        # fall back to) there is no column to encode words from. The refusal is
+        # the worker's own text (issue #472) prefixed with the variable name.
         cfg = self._centroid_cfg()
         del cfg.output["time_source"]
-        with pytest.raises(ValueError, match="requires the store's per-observation clock"):
+        with pytest.raises(ValueError, match=re.escape(TOC_NO_CLOCK_ERROR)):
             validate_config(cfg)
 
     def test_windowing_satisfies_the_clock(self):
@@ -2722,6 +2726,115 @@ def test_chunk_precompute_may_not_shadow_the_derived_name(self):
             validate_config(cfg)
 
 
+class TestTemporalClockAtSubmission:
+    """``validate_config`` refuses a ``temporal:`` companion whose clock does
+    not resolve, in the worker's own words (issue #472).
+
+    The config shape is the observed one: the ``02_write`` demo grafted
+    ``aggregation["variables"]`` from ``atl03_tdigest_located_healpix`` (whose
+    variables declare ``temporal: per-centroid``, PR #463) onto the hive base
+    template without also grafting ``output.time_source`` and the
+    ``delta_time`` source column.
+
+    These are **validator-level** pins: the checks themselves predate this PR
+    (``a67be395``/``ef68ef74``, issue #410) and pass on ``main`` — what is new
+    here is the single-sourced message text (``test_both_seams_raise_the_same_text``).
+    The regression for the reported symptom is one seam up, where the graft
+    actually dispatched unvalidated:
+    ``tests/test_client.py::TestSubmissionValidation`` (fold review).
+    """
+
+    def _graft(self):
+        import copy
+
+        base = default_config("atl03_tdigest_healpix_hive", validate=False)
+        located = default_config("atl03_tdigest_located_healpix", validate=False)
+        base.aggregation["variables"] = copy.deepcopy(located.aggregation["variables"])
+        return base, located
+
+    def test_graft_without_clock_refused_at_submission(self):
+        # The exact observed shape, pinned to the worker's message text so the
+        # two seams read identically.
+        cfg, _ = self._graft()
+        assert (cfg.output or {}).get("time_source") is None
+        with pytest.raises(ValueError, match=re.escape(TOC_NO_CLOCK_ERROR)):
+            validate_config(cfg)
+
+    def test_graft_clock_without_column_refused_at_submission(self):
+        # The graft's second missing piece: time_source present but its field
+        # not a declared data_source variable — refused here, not as a worker
+        # KeyError one seam later.
+        cfg, located = self._graft()
+        cfg.output["time_source"] = dict(located.output["time_source"])
+        # Guard the precondition against the SAME set the validator checks —
+        # declared variables plus broadcast level variables — so it tracks
+        # ``_validate_time_source`` rather than agreeing with it by luck of
+        # this template (fold review).
+        declared = set(cfg.data_source["variables"]) | _segment_variable_names(cfg.data_source)
+        assert cfg.output["time_source"]["field"] not in declared
+        with pytest.raises(ValueError, match="not a declared data_source variable"):
+            validate_config(cfg)
+
+    def test_graft_with_full_clock_validates(self):
+        # Grafting BOTH missing pieces (the clock block and its column) is the
+        # correct form of the demo's config, and it validates.
+        cfg, located = self._graft()
+        cfg.output["time_source"] = dict(located.output["time_source"])
+        field = cfg.output["time_source"]["field"]
+        cfg.data_source["variables"][field] = located.data_source["variables"][field]
+        validate_config(cfg)
+
+    def test_windowing_fallback_satisfies_the_graft(self):
+        # The continuous-scale windowing fallback (PR #463) resolves the clock
+        # through the same resolver the worker uses (toc_source), so the graft
+        # with a windowing block and its column — but no time_source — is valid.
+        cfg, _ = self._graft()
+        cfg.data_source["variables"]["delta_time"] = "{group}/heights/delta_time"
+        cfg.output["windowing"] = {
+            "schedule": "yearly",
+            "time_field": "delta_time",
+            "epoch": "2018-01-01T00:00:00",
+            "scale": "gps",
+        }
+        validate_config(cfg)
+
+    def test_both_seams_raise_the_same_text(self):
+        # Parity pin: the worker's defense-in-depth refusal is the exact string
+        # the validator embeds (single-sourced in zagg.time_axis, issue #472).
+        from zagg.processing.aggregate import _toc_word_column
+
+        cfg, _ = self._graft()
+        with pytest.raises(ValueError) as worker_exc:
+            _toc_word_column({}, cfg)
+        with pytest.raises(ValueError) as submit_exc:
+            validate_config(cfg)
+        assert str(worker_exc.value) == TOC_NO_CLOCK_ERROR
+        assert TOC_NO_CLOCK_ERROR in str(submit_exc.value)
+
+    def test_every_shipped_temporal_template_validates(self):
+        # Sweep the packaged configs for aggregation variables carrying the
+        # ``temporal:`` key; each such template must pass validate_config, and
+        # the sweep must actually find the known carriers (a guard against the
+        # discovery matching nothing).
+        from importlib import resources
+
+        import zagg.configs
+
+        names = sorted(
+            p.name[: -len(".yaml")]
+            for p in resources.files(zagg.configs).iterdir()
+            if p.name.endswith(".yaml")
+        )
+        carriers = set()
+        for name in names:
+            cfg = default_config(name, validate=False)
+            agg_vars = (cfg.aggregation or {}).get("variables") or {}
+            if any(isinstance(m, dict) and m.get("temporal") for m in agg_vars.values()):
+                carriers.add(name)
+                validate_config(cfg)
+        assert carriers >= {"atl03_tdigest_located_healpix", "gedi01b_waveform_healpix_hive"}
+
+
 class TestOverviewDelta:
     def test_valid_overview_delta_validates(self):
         validate_config(_ragged_cfg(inner_shape=[2], overview_delta=512))