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/notebook.py

@@ -20,7 +20,13 @@
 import html
 import logging
 
-from zagg.config import PipelineConfig, get_pipeline_type, get_store_layout, get_windowing
+from zagg.config import (
+    PipelineConfig,
+    get_pipeline_type,
+    get_store_layout,
+    get_windowing,
+    validate_config,
+)
 from zagg.dispatch import BENIGN_ERRORS, LAMBDA_ARCH, max_cost_usd
 
 logger = logging.getLogger(__name__)
@@ -285,6 +291,15 @@ def run(config: PipelineConfig, **kwargs) -> RunView:
     """
     from zagg import runner
 
+    # Refuse before the preview, not just before the invoke (issue #485).
+    # ``agg`` stays the choke point for the dispatch itself; this call is the
+    # wrapper's ordering guarantee -- ``max_cost_preview`` below reads the
+    # shardmap (a local read or an S3 GET), so without it a bad config would
+    # surface as a catalog error first, masking the real refusal. Same shape
+    # as the facade's own pre-resolution call in ``client.Run.from_config``;
+    # ``validate_config`` is pure, so the second call inside ``agg`` is free.
+    validate_config(config)
+
     total = None
     if kwargs.get("events") is not None:
         # Materialize once and rebind: a one-shot iterable (generator) would be

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,20 @@ def agg(
         Summary with keys: ``total_cells``, ``cells_with_data``,
         ``cells_error``, ``total_obs``, ``wall_time_s``, ``store_path``.
     """
+    # The validation choke point for this entry (issue #485): direct agg
+    # callers on any backend or pipeline kind, plus zagg.notebook.run, whose
+    # dispatch bottoms out here — so a PipelineConfig mutated after
+    # load_config's own call (the issue #472 graft) is refused before agg
+    # reads a catalog, touches a store, or invokes. Two sibling seams call
+    # validate_config themselves rather than routing through this one, and
+    # both must stay: client.Run.from_config dispatches via Run/StatusPoller,
+    # never through agg (client.py, same error text), and notebook.run repeats
+    # the call above its max_cost_preview branch, which reads the shardmap
+    # before reaching agg. 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 --------------------------------------------------------