D24 composability: admit build_waveform_digest via a shared digest-family registry (issue #508)

src/zagg/configs/gedi01b_waveform_healpix_hive.yaml

@@ -267,7 +267,9 @@ output:
     sharded: true
     child_order: 18              # leaf cell resolution (~20 m, ~GEDI footprint)
   store_layout: hive
-  pyramid: false                 # composability none initially (issue #422)
+  pyramid: false                 # opt-in (runbook: CA builds pyramids-off on
+                                 # 0.49); rx_flux classifies approximate since
+                                 # issue #508, so pyramid: {} builds the ladder
 
 worker:
   memory: 4096                   # optimized probe (run 5d081b68, 4/4 green):

src/zagg/processing/streaming.py

@@ -61,6 +61,26 @@
 _TDIGEST_WHERE_FUNCTION = "zagg.stats.tdigest.build_tdigest_where"
 _TDIGEST_SPILL_FUNCTIONS = (*_TDIGEST_FUNCTIONS, _TDIGEST_WHERE_FUNCTION)
 
+#: The digest FAMILY (issue #508): reducers whose STORED payloads are spec-§2
+#: weights-sorted ``(k, 2)`` centroid arrays and therefore fold by the
+#: order-independent k-way t-digest law (``merge_tdigests_kway``) wherever a
+#: fold operates on stored payloads rather than raw rows — the D24
+#: composability arm (:func:`zagg.semantics.field_composability`) and the
+#: overview pyramid it licenses (:func:`zagg.sweep_overview.fold_digests`).
+#: The k-way law is weight-agnostic — flux weights fold like counts (spec
+#: §2.0, issue #431) — which is what admits ``build_waveform_digest``: its
+#: build is waveform-specific, but its stored payload is a standard §2
+#: centroid array whose companions ride the same channel overloads.
+#: Deliberately NOT consumed by :func:`validate_streaming` /
+#: :func:`validate_spill_fold`: those gates re-run BUILDERS over raw rows (per
+#: flush / per block), and the waveform builder's noise-model columns are not
+#: threaded there — the tuples above keep their exact members (the issue #508
+#: phase-1 characterization pins that posture). ``build_tdigest_where`` is
+#: likewise NOT here: its D24 class stays ``none`` pending the gate-drift
+#: ruling raised on issue #508.
+_WAVEFORM_DIGEST_FUNCTION = "zagg.stats.waveform.build_waveform_digest"
+_DIGEST_FAMILY_FUNCTIONS = (*_TDIGEST_FUNCTIONS, _WAVEFORM_DIGEST_FUNCTION)
+
 #: The packed composition reducer (issue #321): the SPILL fold collapses its
 #: per-block ``(word, n_signal)`` pairs in one pass via
 #: ``merge_composition_kway`` — issue #370 option (a), accepting the documented

src/zagg/semantics.py

@@ -373,9 +373,12 @@ def field_composability(meta: dict) -> str:
     ``validate_streaming`` accepts, widened by the exact sum/min/max laws):
 
     - ``exact`` — scalar ``function`` in :data:`EXACT_MERGE_LAWS`;
-    - ``approximate`` — a ragged t-digest field with the standard ``(2,)``
-      centroid inner shape, **located or not** (merge is order-dependent;
-      ``np.isclose`` equality class, cf. D24);
+    - ``approximate`` — a ragged digest-family field with the standard
+      ``(2,)`` centroid inner shape, **located or not** (merge is
+      order-dependent; ``np.isclose`` equality class, cf. D24). The family is
+      :data:`zagg.processing.streaming._DIGEST_FAMILY_FUNCTIONS` — the
+      t-digest builders plus the waveform flux digest (issue #508), every
+      reducer whose stored payload folds by the k-way law;
     - ``none`` — everything else: expressions, vector fields, chunk-resolution
       companions, a ``temporal: per-cell`` dense companion (see below), and any
       scalar reducer without an exact law (mean, std, median, quantiles, ...).
@@ -403,8 +406,8 @@ def field_composability(meta: dict) -> str:
     word grammar's join over a cell group, not its own reducer, so classifying
     it by ``function`` would fold e.g. ``nanmax`` over toc words and emit a word
     whose conservative-envelope claim is false. Wiring a dense toc law is not
-    what the ruling asked for, and no shipped config needs it (the GEDI template
-    declares ``pyramid: false``), so it is left out rather than guessed at.
+    what the ruling asked for, and no shipped config pairs a per-cell
+    companion with a pyramid, so it is left out rather than guessed at.
     """
     from zagg.config import get_output_signature
     from zagg.time_axis import TOC_SHAPE_PER_CELL
@@ -416,9 +419,9 @@ def field_composability(meta: dict) -> str:
         return "none"
     function = _fold_function_name(meta.get("function"))
     if sig["kind"] == "ragged":
-        from zagg.processing.streaming import _TDIGEST_FUNCTIONS
+        from zagg.processing.streaming import _DIGEST_FAMILY_FUNCTIONS
 
-        if meta.get("function") in _TDIGEST_FUNCTIONS and tuple(sig["inner_shape"]) == (2,):
+        if meta.get("function") in _DIGEST_FAMILY_FUNCTIONS and tuple(sig["inner_shape"]) == (2,):
             return "approximate"
         return "none"
     if sig["kind"] == "scalar" and function in EXACT_MERGE_LAWS:

src/zagg/stats/waveform.py

@@ -203,13 +203,15 @@ def build_waveform_digest(
     the samples that survived into it
     (:func:`zagg.stats.tdigest._centroid_envelopes`). Given, the return is a
     ``(digest, words)`` pair. The per-centroid **shape** is the one the espg
-    ruling of 2026-08-17 makes universal, identical to the located channel's;
-    the ruling's *at every level* half does not describe this reducer's stores,
-    because ``build_waveform_digest`` is absent from
-    ``zagg.processing.streaming._TDIGEST_FUNCTIONS`` — so a waveform field is
-    D24 class ``none`` (issue #422, and the GEDI template's ``pyramid: false``)
-    and exists at native resolution only. There is no waveform overview to carry
-    a companion, and a reader must not expect one.
+    ruling of 2026-08-17 makes universal, identical to the located channel's,
+    and its *at every level* half describes this reducer's stores too:
+    ``build_waveform_digest`` is a member of the shared digest family
+    (``zagg.processing.streaming._DIGEST_FAMILY_FUNCTIONS``, issue #508), so a
+    waveform field is D24 class ``approximate`` and folds through the overview
+    pyramid when one is declared — each overview level carries the
+    per-centroid companion beside the folded payload. (The build-time gates
+    are unchanged: the merge/spill folds still refuse the builder, so a
+    waveform shard aggregates pooled or single-block-spill only.)
 
     What the words say is ruling 2's honesty property: a waveform record's
     samples share one instant, so a single-shot cell's centroids all carry that

tests/test_spill_crossblock.py

@@ -347,6 +347,160 @@ def _pooled_build(meta, n=8):
     return resolve_function(meta["function"])(cell_data["h_ph"], **params)
 
 
+_WAVEFORM_FUNCTION = "zagg.stats.waveform.build_waveform_digest"
+
+
+def _waveform_variables(temporal=True, delta=64):
+    """A ``rx_flux``-shaped field: the gedi01b template's reducer + channel.
+
+    Sources ``h_ph`` and reads the noise-model columns via the params-as-column
+    path, exactly as the shipped template wires ``rxwaveform``/``noise_mean``/
+    ``noise_stddev`` (issue #508 phase 1 harness).
+    """
+    field = {
+        "kind": "ragged",
+        "function": _WAVEFORM_FUNCTION,
+        "source": "h_ph",
+        "inner_shape": [2],
+        "dtype": "float32",
+        "fill_value": 0,
+        "params": {
+            "delta": delta,
+            "counts": "wf_counts",
+            "noise_mean": "wf_noise_mean",
+            "noise_stddev": "wf_noise_stddev",
+            "gain": 1.0,
+            "false_positive_rate": 1.0e-3,
+            "samples_per_record": 60,
+        },
+    }
+    if temporal:
+        field["temporal"] = "per-centroid"
+    return {
+        "count": {"function": "len", "source": "h_ph", "dtype": "int32", "fill_value": 0},
+        "rx_flux": field,
+    }
+
+
+def _with_waveform_columns(dfs, seed=0):
+    """Add the noise-model columns ``_waveform_variables`` declares.
+
+    Counts sit well above the zero-mean noise floor, so every row survives the
+    clip and ``sum(weights)`` over a cell is the sum of its rows' counts.
+    """
+    rng = np.random.default_rng(seed)
+    for df in dfs:
+        n = len(df)
+        df["wf_counts"] = rng.uniform(5.0, 50.0, n)
+        df["wf_noise_mean"] = np.zeros(n, dtype=np.float64)
+        df["wf_noise_stddev"] = np.full(n, 0.1, dtype=np.float64)
+    return dfs
+
+
+class TestWaveformSpillBaseline:
+    """How the spill path treats ``build_waveform_digest`` TODAY (issue #508).
+
+    Phase 1 characterization, recorded before any registry changes: the SERC
+    GEDI fleet runs (0.47–0.49) completed under ``{mode: spill}``, and the
+    mechanism that carried them is the NON-mergeable single-block regime.
+    ``validate_spill_fold`` refuses the builder (it is outside
+    ``_TDIGEST_SPILL_FUNCTIONS``) — and on the SHIPPED template the refusal is
+    OVER-determined, naming the per-shot ``single_shot_value``/``shot_*``
+    scalar companions alongside it — so ``SpillAggregator`` records the
+    verdict, replays the pooled machinery byte-identically while the shard
+    fits in one block, and raises ``SpillOverflowError`` naming the field on
+    the first block close. The synthetic below declares only ``count`` +
+    ``rx_flux`` so the builder is isolated as the cause; the last test pins
+    the deployed config, where it is not the only one. Issue #508 changes
+    NONE of this: the shared digest-family registry feeds the D24 pyramid
+    classification (stored-payload folds), not the build-time spill fold,
+    which would need the noise-model columns re-threaded per block.
+    """
+
+    def test_probe_refuses_the_waveform_builder(self):
+        # The per-centroid channel passes the temporal arm (issue #477); the
+        # refusal is the ragged function arm — no cross-block fold law.
+        cfg = _config(_waveform_variables(), streaming=_SPILL)
+        with pytest.raises(ValueError, match="'rx_flux'.*build_waveform_digest.*fold law"):
+            validate_spill_fold(cfg)
+
+    def test_spill_accepts_the_config_as_non_mergeable(self):
+        # Accepted — single-block exact — with the probe's verdict recorded so
+        # a block close can name the field (issue #474 message discipline).
+        cfg = _config(_waveform_variables(), streaming=_SPILL)
+        agg = SpillAggregator(cfg, _grid(cfg), "pandas", 1)
+        assert not agg._mergeable
+        assert agg._digest_fields == {}
+        assert "rx_flux" in agg._fold_problems and "fold law" in agg._fold_problems
+        agg.close()
+
+    def test_single_block_regime_is_byte_identical_to_pooled(self, monkeypatch):
+        # The SERC mechanism: one block -> the pooled replay, payload AND the
+        # per-centroid temporal channel byte-identical to the pooled path.
+        key = _shard_key()
+        pooled_cfg = _config(_waveform_variables())
+        spill_cfg = _config(_waveform_variables(), streaming=_SPILL)
+        grid = _grid(pooled_cfg)
+        dfs = _with_waveform_columns(
+            _granule_dfs(grid, key, _CELL_LISTS[:3], obs_per_cell=40, seed=6, times=True)
+        )
+        df_p, ragged_p, _ = _run(monkeypatch, pooled_cfg, grid, key, list(dfs))
+        df_s, ragged_s, _ = _run(monkeypatch, spill_cfg, _grid(spill_cfg), key, list(dfs))
+        pd.testing.assert_series_equal(df_p["count"], df_s["count"])
+        assert set(ragged_p) == set(ragged_s) == {"rx_flux"}
+        pay_p, idx_p, locs_p, times_p = _channels_of(ragged_p["rx_flux"])
+        pay_s, idx_s, locs_s, times_s = _channels_of(ragged_s["rx_flux"])
+        assert idx_p == idx_s and len(pay_p) > 0
+        assert locs_p is None and locs_s is None
+        for a, b in zip(pay_p, pay_s, strict=True):
+            np.testing.assert_array_equal(a, b)
+        for a, b in zip(times_p, times_s, strict=True):
+            np.testing.assert_array_equal(a, b)
+
+    def test_block_close_raises_overflow_naming_the_field(self, monkeypatch):
+        # The recorded boundary of the mechanism above: past one block there is
+        # no fold law, and the overflow splices the probe's verdict.
+        from zagg.processing.spill import SpillOverflowError
+
+        _force_tiny_blocks(monkeypatch)
+        key = _shard_key()
+        cfg = _config(_waveform_variables(), streaming=_SPILL)
+        grid = _grid(cfg)
+        dfs = _with_waveform_columns(
+            _granule_dfs(grid, key, _CELL_LISTS[:2], obs_per_cell=10, seed=1, times=True)
+        )
+        with pytest.raises(SpillOverflowError, match="'rx_flux'.*build_waveform_digest.*fold law"):
+            _run(monkeypatch, cfg, grid, key, dfs)
+
+    def test_the_shipped_template_is_refused_by_more_than_the_builder(self):
+        # The synthetic above isolates the builder; the DEPLOYED config is
+        # over-determined. Seven of its eight fields are per-shot scalars with
+        # no cross-block fold either, so even if issue #508 admitted
+        # ``build_waveform_digest`` to a shared digest-family registry the GEDI
+        # store would stay non-mergeable -- ``spill_blocks_closed: 0`` remains a
+        # hard precondition of the shipped template. Its own
+        # ``worker.extra_disk`` comment says the same: "validate_spill_fold
+        # rejects it (build_waveform_digest and the single_shot_value
+        # companions have no cross-block fold law)".
+        from zagg.config import default_config
+
+        with pytest.raises(ValueError) as exc:
+            validate_spill_fold(default_config("gedi01b_waveform_healpix_hive"))
+        message = str(exc.value)
+        assert "'rx_flux'" in message and "build_waveform_digest" in message
+        for name in (
+            "shot_count",
+            "shot_number",
+            "noise_mean",
+            "noise_stddev",
+            "rx_energy",
+            "elevation_bin0",
+            "elevation_lastbin",
+        ):
+            assert f"'{name}'" in message
+        assert "zagg.stats.waveform.single_shot_value" in message
+
+
 class TestSpillFoldProbe:
     """The spill mergeability probe (validate_spill_fold) vs merge mode."""
 

tests/test_streaming.py

@@ -177,6 +177,29 @@ def test_non_tdigest_ragged_rejected(self):
         with pytest.raises(ValueError, match="merge law"):
             validate_streaming(cfg)
 
+    def test_waveform_digest_ragged_rejected(self):
+        # Issue #508 phase 1 baseline: build_waveform_digest is outside
+        # _TDIGEST_FUNCTIONS, so merge mode refuses it through the same
+        # no-merge-law arm as any other non-tdigest ragged reducer.
+        cfg = _config()
+        cfg.aggregation["variables"]["h_tdigest"]["function"] = (
+            "zagg.stats.waveform.build_waveform_digest"
+        )
+        with pytest.raises(ValueError, match="h_tdigest.*build_waveform_digest.*merge law"):
+            validate_streaming(cfg)
+
+    def test_waveform_temporal_field_refused_via_the_temporal_arm(self):
+        # The shipped template shape (rx_flux carries temporal: per-centroid):
+        # the companion refusal fires before the function arm ever sees the
+        # builder, and routes to mode: spill (issue #508 phase 1 baseline).
+        cfg = _config()
+        cfg.aggregation["variables"]["h_tdigest"]["function"] = (
+            "zagg.stats.waveform.build_waveform_digest"
+        )
+        cfg.aggregation["variables"]["h_tdigest"]["temporal"] = "per-centroid"
+        with pytest.raises(ValueError, match="temporal companions.*cannot stream.*mode: spill"):
+            validate_streaming(cfg)
+
     def test_pairwise_tdigest_reducer_is_streamable(self):
         # build_tdigest_pairwise carries the pairwise merge law (issue #279),
         # so it must validate as a mergeable ragged reducer just like the