feat(delivery): the no-collapse invariant — draws.py (ADR-013 §6)

tests/test_draws.py

@@ -0,0 +1,79 @@
+"""Unit tests for the no-collapse invariant (ADR-013 §6 / views-models#149 / #45).
+
+Pure primitives in, raise-or-pass out — no framework, no pandas.
+"""
+
+import numpy as np
+import pytest
+
+from views_postprocessing.delivery.draws import (
+    DrawsCollapseError,
+    assert_draws_uncollapsed,
+)
+
+
+def _draws(n=6, s=8, seed=0):
+    """A genuine sampled payload: (n, s) with real per-row variance."""
+    return np.random.default_rng(seed).gamma(2.0, size=(n, s)).astype(np.float32)
+
+
+def test_genuine_draws_pass():
+    values = _draws()
+    assert assert_draws_uncollapsed(values, 8, s_min=2) is None
+
+
+def test_pgm_zero_inflation_is_legal():
+    # The key non-rule: most rows constant (zero-conflict cells), ONE row varied → pass.
+    values = np.zeros((100, 8), dtype=np.float32)
+    values[42] = np.arange(8)
+    assert assert_draws_uncollapsed(values, 8, s_min=2) is None
+
+
+def test_collapsed_1d_payload_raises():
+    with pytest.raises(DrawsCollapseError, match="2-D"):
+        assert_draws_uncollapsed(np.zeros(10), 8, s_min=2)
+
+
+def test_3d_payload_raises():
+    with pytest.raises(DrawsCollapseError, match="2-D"):
+        assert_draws_uncollapsed(np.zeros((4, 3, 8)), 8, s_min=2)
+
+
+def test_header_payload_mismatch_raises_naming_both():
+    with pytest.raises(DrawsCollapseError) as exc:
+        assert_draws_uncollapsed(_draws(s=8), 1024, s_min=2)
+    msg = str(exc.value)
+    assert "1024" in msg  # declared
+    assert "S=8" in msg  # found
+
+
+def test_below_s_min_floor_raises():
+    # A conformant-but-thin payload below the delivery floor must not ship.
+    with pytest.raises(DrawsCollapseError, match="s_min"):
+        assert_draws_uncollapsed(_draws(s=2), 2, s_min=4)
+
+
+def test_skeleton_floor_is_two():
+    # The walking-skeleton configuration: S=2, s_min=2 passes.
+    values = np.column_stack([np.zeros(5), np.ones(5)]).astype(np.float32)
+    assert assert_draws_uncollapsed(values, 2, s_min=2) is None
+
+
+def test_fully_degenerate_payload_raises():
+    # Every row constant across draws = a collapsed forecast wearing a sampled header.
+    values = np.tile(np.arange(6, dtype=np.float32).reshape(-1, 1), (1, 8))
+    with pytest.raises(DrawsCollapseError, match="draw-degenerate"):
+        assert_draws_uncollapsed(values, 8, s_min=2)
+
+
+def test_nan_rows_count_as_varied_documented_behavior():
+    # Documented: NaN compares unequal to everything, so a NaN row is "varied" here —
+    # NaN payloads are the null gates' job, not this invariant's.
+    values = np.zeros((3, 4), dtype=np.float32)
+    values[1, 2] = np.nan
+    assert assert_draws_uncollapsed(values, 4, s_min=2) is None
+
+
+def test_label_appears_in_message():
+    with pytest.raises(DrawsCollapseError, match="second-store"):
+        assert_draws_uncollapsed(np.zeros(3), 8, s_min=2, label="second-store")

views_postprocessing/delivery/draws.py

@@ -0,0 +1,86 @@
+"""No-collapse invariant: a sampled forecast must still be draws when it ships
+(ADR-013 §6 — the boundary named by views-models#149; register #45).
+
+Representation-free — primitives only (a 2-D numpy array + declared ints). No pandas,
+no views_frames, no datastore types. The manager feeds it the payload and the *declared*
+contract values (the header's ``sample_count``, the configured ``s_min``); the rule
+lives here. Declare-don't-guess: nothing is inferred from the payload — the caller
+states what the payload claims to be, and this invariant fails loud if it is not.
+
+Why this matters: the whole point of the sampled-forecast wire (ADR-013) is that FAO
+receives the full posterior mixture — ``(N, S≈1024)`` draws per cell — collapsed to a
+point exactly once, downstream, in the summarizer. Any hop that bakes out a mean/median
+or ships a mislabeled point payload silently flattens the mixture. This is the single
+**policy** gate (the per-hop emission/ingest asserts are mechanism guards); it runs
+before every FAO-facing forecast upload.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+
+
+class DrawsCollapseError(ValueError):
+    """The forecast payload is not the uncollapsed draws its header declares."""
+
+
+def assert_draws_uncollapsed(
+    values,
+    sample_count: int,
+    *,
+    s_min: int,
+    label: str = "forecast",
+) -> None:
+    """Raise unless ``values`` is genuinely the declared ``(N, sample_count)`` draws.
+
+    Checks, in order (ADR-013 §6):
+
+    1. ``values`` is 2-D ``(N, S)`` — a collapsed 1-D column or a stacked 3-D block is
+       not a sampled forecast payload.
+    2. payload ``S == sample_count`` — the header's declaration must match the bytes.
+    3. ``sample_count >= s_min`` — the consumer-facing floor (2 for the walking
+       skeleton; region-pinned for production). Declared by the caller from config,
+       never guessed here.
+    4. **Global** non-degeneracy — at least one row carries more than one distinct
+       draw value. Per-row zero variance is explicitly **legal** (zero-conflict cells
+       dominate PGM); only a payload where *every* row is draw-degenerate fails — that
+       is a collapsed forecast wearing a sampled header, not an edge case.
+
+    NaN note: a NaN draw compares unequal to everything (including another NaN), so a
+    NaN-carrying row counts as non-degenerate here. That is deliberate — NaN payloads
+    are not this invariant's job; they fail loud at the delivery null gates.
+
+    Args:
+        values: the forecast payload — a 2-D numpy array of draws, one row per
+            (time, unit), one column per draw.
+        sample_count: the ``sample_count`` the artifact's contract header declares.
+        s_min: the minimum draw count the delivery accepts (config-declared).
+        label: short tag for the message (e.g. ``"forecast"``).
+
+    Raises:
+        DrawsCollapseError: naming the failed check, expected vs found.
+    """
+    arr = np.asarray(values)
+    if arr.ndim != 2:
+        raise DrawsCollapseError(
+            f"{label}: payload is {arr.ndim}-D, expected 2-D (N, S) draws — a collapsed "
+            f"or mis-shaped payload must not ship as a sampled forecast."
+        )
+    payload_s = arr.shape[1]
+    if payload_s != sample_count:
+        raise DrawsCollapseError(
+            f"{label}: header declares sample_count={sample_count} but the payload "
+            f"carries S={payload_s} — the artifact lies about its own draws."
+        )
+    if sample_count < s_min:
+        raise DrawsCollapseError(
+            f"{label}: sample_count={sample_count} is below the delivery floor "
+            f"s_min={s_min} — a (near-)point payload must not ship as a sampled forecast."
+        )
+    varied_rows = np.any(arr != arr[:, :1], axis=1)
+    if not bool(varied_rows.any()):
+        raise DrawsCollapseError(
+            f"{label}: every one of the {arr.shape[0]} rows is draw-degenerate (all "
+            f"{sample_count} draws identical per row) — this is a collapsed forecast "
+            f"wearing a sampled header. (Per-row zero variance is legal; all-rows is not.)"
+        )