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

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,13 @@ 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.
     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

@@ -49,15 +49,13 @@ def _toc_word_column(cell_data: dict, config) -> np.ndarray:
     a ``per-cell`` field as its ``source`` column — so one store can never carry
     two clocks.
     """
-    from zagg.time_axis import observation_words, toc_source
+    from zagg.time_axis import TOC_NO_CLOCK_ERROR, observation_words, 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 cell_data:
         raise ValueError(
             f"output.time_source.field {source['field']!r} is not in the cell data "

src/zagg/time_axis.py

@@ -91,6 +91,17 @@
 #: 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.
+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} (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

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
@@ -32,6 +33,7 @@
     validate_config,
 )
 from zagg.processing import calculate_cell_statistics
+from zagg.time_axis import TOC_NO_CLOCK_ERROR
 
 # ---------------------------------------------------------------------------
 # Fixtures
@@ -2404,10 +2406,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 +2725,104 @@ def test_chunk_precompute_may_not_shadow_the_derived_name(self):
             validate_config(cfg)
 
 
+class TestTemporalClockAtSubmission:
+    """A ``temporal:`` companion without a resolvable clock is refused at
+    submission, not first by the worker on the fleet (issue #472).
+
+    The regression 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 — every shard then burned an invoke on the
+    worker's refusal, for an error fully determinable from the config dict.
+    """
+
+    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"])
+        assert cfg.output["time_source"]["field"] not in cfg.data_source["variables"]
+        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))