Replacement scorecard: head-to-head yardstick and the incumbent's score

packages/microcosm-build/src/microcosm/build/us_runtime/h5_io.py

@@ -46,6 +46,7 @@
     "US_MULTISPINE_POOL_MANIFEST_ARTIFACT_KIND",
     "US_MULTISPINE_POOL_MANIFEST_SCHEMA_VERSION",
     "US_STACKED_POOL_OPERATOR_ORDER",
+    "load_authenticated_us_multispine_pool_for_scoring",
     "load_legacy_calibrated_us_h5",
     "load_simulation_ready_us_multispine_pool",
     "load_simulation_ready_us_multispine_pool_manifest",
@@ -371,8 +372,15 @@ def _load_authenticated_us_multispine_pool_manifest(
     path: str | Path,
     *,
     expected_manifest_sha256: str | None = None,
+    allow_terminal_gate_failure: bool = False,
 ) -> tuple[dict[str, object], AuthenticatedPoolH5]:
-    """Return the validated manifest and its authenticated pool-H5 identity."""
+    """Return a validated manifest and its authenticated pool-H5 identity.
+
+    By default this is the production readiness boundary. The scoring-only
+    caller may also authenticate a current stacked publication whose terminal
+    gates failed. That exception changes no digest, schema, run-ID, H5,
+    diagnostics, or gate-alias validation and never labels the artifact ready.
+    """
 
     manifest_path = Path(path)
     manifest, manifest_sha256, _ = _read_json_object_with_identity(
@@ -385,10 +393,15 @@ def _load_authenticated_us_multispine_pool_manifest(
             f"US multispine pool manifest {manifest_path} has an unsupported "
             "artifact binding."
         )
-    if (
-        manifest.get("simulation_ready") is not True
-        or manifest.get("status") != "simulation_ready"
-    ):
+    is_ready = (
+        manifest.get("simulation_ready") is True
+        and manifest.get("status") == "simulation_ready"
+    )
+    is_terminal_gate_failure = (
+        manifest.get("simulation_ready") is False
+        and manifest.get("status") == "gate_failed"
+    )
+    if not is_ready and not (allow_terminal_gate_failure and is_terminal_gate_failure):
         raise ValueError(
             f"US multispine pool manifest {manifest_path} is not simulation-ready."
         )
@@ -401,6 +414,12 @@ def _load_authenticated_us_multispine_pool_manifest(
         if envelope == "stacked"
         else _LEGACY_MULTISPINE_POOL_MANIFEST_SCHEMA_VERSION
     )
+    if is_terminal_gate_failure and envelope != "stacked":
+        raise ValueError(
+            f"US multispine pool manifest {manifest_path} may expose a failed "
+            "terminal-gate publication for scoring only under the current "
+            "stacked envelope."
+        )
     _validate_stacked_late_dag_manifest_binding(
         manifest,
         manifest_path=manifest_path,
@@ -534,12 +553,12 @@ def _load_authenticated_us_multispine_pool_manifest(
         diagnostics.get("artifact_kind")
         != US_MULTISPINE_AGREEMENT_DIAGNOSTICS_ARTIFACT_KIND
         or diagnostics.get("schema_version") != expected_schema_version
-        or diagnostics.get("simulation_ready") is not True
+        or diagnostics.get("simulation_ready") != manifest.get("simulation_ready")
         or diagnostics.get("publication_run_id") != publication_run_id
     ):
         raise ValueError(
             f"US multispine pool diagnostics {diagnostics_path} do not match "
-            "the ready manifest publication."
+            "the authenticated manifest publication."
         )
     manifest_agreement_gate = _mapping(
         manifest.get("agreement_gate"),
@@ -560,7 +579,7 @@ def _load_authenticated_us_multispine_pool_manifest(
     if diagnostics_agreement_gate != manifest_agreement_gate:
         raise ValueError(
             f"US multispine pool diagnostics {diagnostics_path} agreement-gate "
-            "verdict does not match the ready manifest."
+            "verdict does not match the authenticated manifest."
         )
     return manifest, AuthenticatedPoolH5(
         path=pool_path.resolve(),
@@ -690,16 +709,55 @@ def load_simulation_ready_us_multispine_pool(
         immutable identity object for every downstream H5 consumer.
     """
 
+    return _load_us_multispine_pool(
+        path,
+        expected_manifest_sha256=expected_manifest_sha256,
+        require_simulation_ready=True,
+    )
+
+
+def load_authenticated_us_multispine_pool_for_scoring(
+    path: str | Path,
+    *,
+    expected_manifest_sha256: str | None = None,
+) -> tuple[Frame, dict[str, object], AuthenticatedPoolH5]:
+    """Load authenticated pool evidence without promoting failed gates.
+
+    A current stacked publication with ``status=gate_failed`` and
+    ``simulation_ready=false`` remains useful as head-to-head evidence. This
+    loader accepts that exact status pair, authenticates the same manifest,
+    diagnostics, H5 bytes, run ID, terminal-gate aliases, and row counts as the
+    production loader, and preserves the failed receipt. It is deliberately
+    separate from :func:`load_simulation_ready_us_multispine_pool`, whose
+    readiness contract is unchanged.
+    """
+
+    return _load_us_multispine_pool(
+        path,
+        expected_manifest_sha256=expected_manifest_sha256,
+        require_simulation_ready=False,
+    )
+
+
+def _load_us_multispine_pool(
+    path: str | Path,
+    *,
+    expected_manifest_sha256: str | None,
+    require_simulation_ready: bool,
+) -> tuple[Frame, dict[str, object], AuthenticatedPoolH5]:
+    """Shared authenticated H5 reconstruction for readiness and scoring."""
+
     manifest_path = Path(path)
     manifest, authenticated_pool_h5 = _load_authenticated_us_multispine_pool_manifest(
         manifest_path,
         expected_manifest_sha256=expected_manifest_sha256,
+        allow_terminal_gate_failure=not require_simulation_ready,
     )
     agreement_gate = _mapping(
         manifest.get("agreement_gate"),
         label=f"US multispine pool manifest {manifest_path}.agreement_gate",
     )
-    if agreement_gate.get("passed") is not True:
+    if require_simulation_ready and agreement_gate.get("passed") is not True:
         raise ValueError(
             f"US multispine pool manifest {manifest_path} has no passing "
             "agreement-gate verdict."

packages/microcosm-build/src/microcosm/build/us_runtime/stacked_spine.py

@@ -9427,6 +9427,98 @@ def _gap_fill_absence_rule_mask(
     }
 
 
+def _gap_fill_absence_rule_mask_artifact_evidence(
+    frame: Frame,
+    *,
+    direction: GapFillDirection,
+    rule: GapFillAbsenceRule,
+) -> tuple[pd.Series, dict[str, object]]:
+    """Resolve the one structural scope from finished artifact columns only."""
+
+    if (
+        rule.rule_id != _ACS_GQ_RENT_ABSENCE_RULE_ID
+        or rule.selection != _ACS_GQ_RENT_ABSENCE_SELECTION
+        or rule.entity != "person"
+        or rule.column != "pre_subsidy_rent"
+        or direction.recipient_channel != ACS_STACKED_SUPPORT_CHANNEL
+    ):
+        raise ValueError(
+            f"Gap-fill direction {direction.name!r} declares unsupported "
+            f"recipient absence rule {rule.rule_id!r}."
+        )
+    household = frame.table("household")
+    person = frame.table("person")
+    required_household = {
+        "household_id",
+        "TYPEHUGQ",
+        "tenure_type",
+        support_channel_column("household"),
+        support_clone_index_column("household"),
+    }
+    required_person = {
+        "person_household_id",
+        support_channel_column("person"),
+        support_clone_index_column("person"),
+    }
+    missing_household = sorted(required_household - set(household))
+    missing_person = sorted(required_person - set(person))
+    if missing_household or missing_person:
+        raise ValueError(
+            f"Gap-fill absence rule {rule.rule_id!r} cannot resolve its "
+            "artifact-evidence universe; "
+            f"missing_household={missing_household}, missing_person={missing_person}."
+        )
+    household_channel = household[support_channel_column("household")].astype(str)
+    household_clone = pd.to_numeric(
+        household[support_clone_index_column("household")], errors="raise"
+    ).astype("int64")
+    native_acs = household_channel.eq(ACS_STACKED_SUPPORT_CHANNEL) & (
+        household_clone == 0
+    )
+    kind = pd.to_numeric(household["TYPEHUGQ"], errors="coerce")
+    invalid_kind = native_acs & ~kind.isin((1, 2, 3))
+    if invalid_kind.any():
+        raise ValueError(
+            f"Gap-fill absence rule {rule.rule_id!r} found "
+            f"{int(invalid_kind.sum())} native ACS household row(s) without "
+            "TYPEHUGQ 1/2/3."
+        )
+    gq_households = native_acs & kind.isin((2, 3))
+    nonnull_gq_tenure = gq_households & household["tenure_type"].notna()
+    if nonnull_gq_tenure.any():
+        raise ValueError(
+            f"Gap-fill absence rule {rule.rule_id!r} found "
+            f"{int(nonnull_gq_tenure.sum())} ACS group-quarters household "
+            "row(s) with synthesized tenure."
+        )
+    gq_household_ids = household.loc[gq_households, "household_id"]
+    person_channel = person[support_channel_column("person")].astype(str)
+    person_clone = pd.to_numeric(
+        person[support_clone_index_column("person")], errors="raise"
+    ).astype("int64")
+    mask = (
+        person_channel.eq(ACS_STACKED_SUPPORT_CHANNEL)
+        & person_clone.eq(0)
+        & person["person_household_id"].isin(gq_household_ids)
+    )
+    linked_counts = person.loc[mask, "person_household_id"].value_counts()
+    if len(linked_counts) != int(gq_households.sum()) or not linked_counts.eq(1).all():
+        raise ValueError(
+            f"Gap-fill absence rule {rule.rule_id!r} artifact evidence does "
+            "not contain exactly one native ACS person per group-quarters "
+            "household."
+        )
+    return mask, {
+        "rule_id": rule.rule_id,
+        "selection": rule.selection,
+        "reason": rule.reason,
+        "status": "observed_structural_absence_artifact_evidence",
+        "assembly_manifest_authenticated": False,
+        "rows": int(mask.sum()),
+        "by_origin_role": {f"{ACS_STACKED_SUPPORT_CHANNEL}/clone_0": int(mask.sum())},
+    }
+
+
 def _origin_projection(frame: Frame, *, channel: str) -> Frame:
     """Project one origin's native lineages as a standalone donor frame."""
 
@@ -13048,6 +13140,28 @@ def by_origin_battery(
         authority=_production_stacked_authority(),
         production=True,
         tail_manifest=tail_manifest,
+        authenticate_stacked_artifact=True,
+    )
+
+
+def by_origin_battery_artifact_evidence(frame: Frame) -> GateResult:
+    """Evaluate canonical battery formulas on finished artifact columns.
+
+    A released entity-table H5 does not preserve the assembly manifest or a
+    clone-2 tail manifest. Those receipts authenticate how a production frame
+    was built; they do not change the clone-0, positive-weight ASEC-vs-ACS
+    formulas. This evidence-only seam therefore uses the exact production
+    authority, surface, support rules, metrics, and tolerances while skipping
+    only assembly/tail authentication. Callers must not represent its result
+    as an authenticated production gate receipt.
+    """
+
+    return _by_origin_battery_evaluate(
+        frame,
+        authority=_production_stacked_authority(),
+        production=True,
+        tail_manifest=None,
+        authenticate_stacked_artifact=False,
     )
 
 
@@ -13065,6 +13179,7 @@ def _by_origin_battery_with_test_authority(
         authority=authority,
         production=False,
         tail_manifest=tail_manifest,
+        authenticate_stacked_artifact=True,
     )
 
 
@@ -13091,6 +13206,7 @@ def _by_origin_battery_evaluate(
     authority: _StackedAuthority,
     production: bool,
     tail_manifest: Mapping[str, object] | None = None,
+    authenticate_stacked_artifact: bool = True,
     _canonical_gap_fill_plan: tuple[
         GapFillDirection, ...
     ] = CANONICAL_STACKED_GAP_FILL_PLAN,
@@ -13114,10 +13230,14 @@ def _by_origin_battery_evaluate(
     both origins carry ample support.
     """
 
-    tail_support_receipt = _terminal_tail_support_gate_receipt(
-        frame,
-        tail_manifest,
-        boundary="by-origin battery",
+    tail_support_receipt = (
+        _terminal_tail_support_gate_receipt(
+            frame,
+            tail_manifest,
+            boundary="by-origin battery",
+        )
+        if authenticate_stacked_artifact
+        else None
     )
     authority_receipt = _authority_receipt(authority)
     specs = _battery_specs_from_metric_registry(authority.metric_registry)
@@ -13179,7 +13299,8 @@ def _by_origin_battery_evaluate(
                 ),
             },
         )
-    validate_stacked_spine_frame(frame, boundary="by-origin battery")
+    if authenticate_stacked_artifact:
+        validate_stacked_spine_frame(frame, boundary="by-origin battery")
 
     failures: list[str] = []
     comparisons: dict[str, object] = {}
@@ -13218,10 +13339,18 @@ def _by_origin_battery_evaluate(
             structural_rule = declared_absence_rules.get((spec.entity, column))
             if structural_rule is not None:
                 rule_direction, rule = structural_rule
-                structural_mask, structural_receipt = _gap_fill_absence_rule_mask(
-                    frame,
-                    direction=rule_direction,
-                    rule=rule,
+                structural_mask, structural_receipt = (
+                    _gap_fill_absence_rule_mask(
+                        frame,
+                        direction=rule_direction,
+                        rule=rule,
+                    )
+                    if authenticate_stacked_artifact
+                    else _gap_fill_absence_rule_mask_artifact_evidence(
+                        frame,
+                        direction=rule_direction,
+                        rule=rule,
+                    )
                 )
                 scoped_structural = structural_mask.to_numpy(dtype=bool) & scope
                 scoped_null_mask = series.isna().to_numpy(dtype=bool) & scope