The stacked spine: one origin-labeled spine, banked gap-fill, null-vs-zero doctrine, by-origin battery (#578 revision pilot)

changelog.d/stacked-spine-pilot-578.added.md

@@ -0,0 +1 @@
+Stacked-spine pilot for the #578 increment-2 revision: assemble ASEC plus a seeded ACS household sample into one origin-labeled spine, gap-fill survey-specific fields cross-origin with native predictors under the null-vs-zero doctrine, run a single PUF pass with a seeded clone attachment, and gate the result with a pre-simulation completeness proof and a by-origin battery bound to one immutable, live-digested 90-target authority bundle.

packages/populace-build/src/populace/build/gates.py

@@ -39,7 +39,9 @@
 from __future__ import annotations
 
 import math
+import pickle
 from collections.abc import Iterable, Mapping
+from copy import deepcopy
 from dataclasses import dataclass, field
 
 import numpy as np
@@ -100,6 +102,10 @@
     {kind.value for kind in WeightKind} | {UNWEIGHTED_KIND, EXPLICIT_KIND}
 )
 
+_STACKED_AUTHORITY_GATE_NAMES = frozenset(
+    {"us_stacked_completeness", "us_by_origin_battery"}
+)
+
 
 @dataclass(frozen=True)
 class GateResult:
@@ -120,6 +126,12 @@ class GateResult:
     passed: bool
     failures: tuple[str, ...] = ()
     details: Mapping[str, object] = field(default_factory=dict)
+    _details_protocol_5_snapshot: bytes | None = field(
+        init=False,
+        repr=False,
+        compare=False,
+        default=None,
+    )
 
     def __post_init__(self) -> None:
         if self.passed and self.failures:
@@ -128,6 +140,12 @@ def __post_init__(self) -> None:
             raise ValueError(
                 f"Gate {self.name!r} cannot fail without naming a failure."
             )
+        if self.name in _STACKED_AUTHORITY_GATE_NAMES:
+            object.__setattr__(
+                self,
+                "_details_protocol_5_snapshot",
+                pickle.dumps(dict(self.details), protocol=5),
+            )
 
 
 @dataclass(frozen=True)
@@ -156,13 +174,89 @@ def failures(self) -> tuple[str, ...]:
 
     def to_manifest(self) -> dict[str, object]:
         """A JSON-ready summary for the release manifest."""
+        for result in self.results:
+            authority = result.details.get("authority")
+            components = (
+                authority.get("components") if isinstance(authority, Mapping) else None
+            )
+            targets = result.details.get("targets")
+            recognizable_stacked_receipt = (
+                isinstance(authority, Mapping)
+                and (
+                    str(authority.get("authority_id", "")).startswith(
+                        "us_stacked_spine_authority"
+                    )
+                    or bool(
+                        {
+                            "authority_form",
+                            "declared_authority_form",
+                            "canonical",
+                            "canonical_identity",
+                            "canonical_content",
+                            "integrity_valid",
+                            "digest_matches_declared",
+                            "production_manifest_permitted",
+                        }
+                        & set(authority)
+                    )
+                    or (
+                        isinstance(components, Mapping)
+                        and set(components)
+                        == {
+                            "gap_fill_plan",
+                            "declared_surface",
+                            "metric_registry",
+                            "support_profile",
+                        }
+                    )
+                )
+            ) or (
+                isinstance(targets, Mapping)
+                and any(
+                    isinstance(receipt, Mapping)
+                    and (
+                        "authority_form" in receipt
+                        or {
+                            "authority_sha256",
+                            "plan_sha256",
+                            "surface_sha256",
+                        }.issubset(receipt)
+                    )
+                    for receipt in targets.values()
+                )
+            )
+            if result.name not in _STACKED_AUTHORITY_GATE_NAMES:
+                if recognizable_stacked_receipt:
+                    raise ValueError(
+                        f"Gate {result.name!r} carries a stacked authority receipt "
+                        "under an unrecognized gate name; production manifest "
+                        "emission is forbidden."
+                    )
+                continue
+            snapshot = result._details_protocol_5_snapshot
+            if (
+                snapshot is None
+                or pickle.dumps(dict(result.details), protocol=5) != snapshot
+            ):
+                raise ValueError(
+                    f"Gate {result.name!r} details changed after evaluation; "
+                    "production manifest emission is forbidden."
+                )
+            # Import lazily to keep the generic gates shard independent during
+            # module initialization. The stacked doctrine owns the trusted
+            # canonical surface and digests; emission must not trust receipts.
+            from populace.build.us_runtime.stacked_spine import (
+                _validate_stacked_gate_manifest_details,
+            )
+
+            _validate_stacked_gate_manifest_details(result.name, result.details)
         return {
             "passed": self.passed,
             "gates": {
                 result.name: {
                     "passed": result.passed,
                     "failures": list(result.failures),
-                    "details": dict(result.details),
+                    "details": deepcopy(dict(result.details)),
                 }
                 for result in self.results
             },