Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/dense-takeup-enforcement.changed.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Dense diagnostic arm fences its SSI adult bands (populace#566/#567): delivered-weight recomputes have not landed the adult pair in band on either observed frame, and a new chain-depth guard refuses retry-of-retry bases as the deleted populace#463-class loop. Fenced misses ship in the scorecard with their adjudication; both build_manifest.json and release_manifest.json record the run's effective enforced set and fenced rows. The sparse certified default keeps hard enforcement.
Original file line number Diff line number Diff line change
Expand Up @@ -935,6 +935,24 @@ def ssi_take_up_prior_basis_from_artifact(
"final measurement; got measurement phase "
f"{measurement_phase!r}."
)
if schema_version == _DIAGNOSTICS_SCHEMA_VERSION:
artifact_prior = payload.get("prior_weight_basis")
artifact_prior_kind = (
artifact_prior.get("kind") if isinstance(artifact_prior, Mapping) else None
)
if artifact_prior_kind == US_SSI_TAKE_UP_PRIOR_BASIS_RELEASE_ARTIFACT:
# Chain-depth guard: populace#508 permits exactly ONE
# delivered-weight recompute. An artifact that was itself
# measured on a retry basis would seed retry-of-retry — the
# deleted populace#463-class loop reassembled by hand.
raise ValueError(
"US SSI take-up prior basis artifact was itself measured "
"on a delivered-weight retry (prior basis kind "
"'release_artifact'); chaining a second recompute is the "
"deleted populace#463-class loop — populace#508 permits "
"exactly one. Investigate the frame instead of retrying "
"again."
)
if schema_version == 3 and measurement_phase not in (
None,
US_SSI_TAKE_UP_PHASE_RELEASE_FINAL,
Expand Down Expand Up @@ -1609,6 +1627,7 @@ def us_ssi_take_up_delivery_gate(
diagnostics: Mapping[str, object],
*,
targets: Mapping[str, float],
enforcement_fences: Mapping[str, str] | None = None,
) -> GateResult:
"""Hard-fail enforced band misses measured on the release weights.

Expand All @@ -1622,11 +1641,44 @@ def us_ssi_take_up_delivery_gate(
the delivered weights. There is no in-build reconcile loop and no
per-target knob (populace#492). The under-18 band stays fenced pending
populace#453/#509 and is reported in the details, never enforced.

``enforcement_fences`` fences normally-enforced bands for a specific
ARM with a documented adjudication (the under-18 pattern extended):
on the dense full-pool arm, populace#508 delivered-weight recomputes
have not landed the adult pair in band on either observed frame —
P2's clean one-retry record, and P3's delivered-basis chain that
``ssi_take_up_prior_basis_from_artifact`` now refuses outright
(populace#566/#567) — so its adult bands ship in the scorecard as
known boundaries rather than enforced contracts. The fence text
rides each fenced row; the sparse certified default passes no
fences and keeps hard enforcement.
"""

expected_targets = _normalize_targets(targets)
tolerance = US_SSI_TAKE_UP_BAND_DELIVERY_RELATIVE_TOLERANCE
failures: list[str] = []
fences = dict(enforcement_fences or {})
non_string_keys = [repr(key) for key in fences if not isinstance(key, str)]
if non_string_keys:
raise ValueError(
"SSI take-up delivery fence keys must be band-name strings; got "
f"{sorted(non_string_keys)}."
)
unknown_fences = sorted(set(fences) - set(US_SSI_TAKE_UP_ENFORCED_BAND_KEYS))
if unknown_fences:
raise ValueError(
"SSI take-up delivery fences may only name normally-enforced "
f"bands {sorted(US_SSI_TAKE_UP_ENFORCED_BAND_KEYS)}; got "
f"{unknown_fences}. A fence on a never-enforced band is a "
"configuration error, not a no-op."
)
for key, fence_text in sorted(fences.items()):
if not isinstance(fence_text, str) or not fence_text.strip():
raise ValueError(
f"SSI take-up delivery fence for band {key!r} carries no "
"adjudication text (fence values must be nonblank strings); "
"a fence without its documented reason is forbidden."
)
enforced_rows: list[dict[str, object]] = []
fenced_rows: list[dict[str, object]] = []
rows: Mapping[str, Mapping[str, object]] = {}
Expand Down Expand Up @@ -1670,7 +1722,8 @@ def us_ssi_take_up_delivery_gate(
"selected_recipient_weight": selected,
"signed_relative_error": signed_relative,
}
if key in US_SSI_TAKE_UP_ENFORCED_BAND_KEYS:
fence_text = fences.get(key)
if key in US_SSI_TAKE_UP_ENFORCED_BAND_KEYS and fence_text is None:
enforced_rows.append(summary)
if abs(selected - target) > tolerance * target + 1e-6:
failures.append(
Expand All @@ -1689,11 +1742,16 @@ def us_ssi_take_up_delivery_gate(
{
**summary,
"fence": (
"Fenced pending the SIPP child qualifying-disability "
"stage (populace#453 / PR #509): certified support "
"cannot truthfully carry the child band yet, so its "
"miss ships in the scorecard — never as "
"saturation-as-success."
fence_text
if fence_text is not None
else (
"Fenced pending the SIPP child "
"qualifying-disability stage (populace#453 / PR "
"#509): certified support cannot truthfully "
"carry the child band yet, so its miss ships in "
"the scorecard — never as "
"saturation-as-success."
)
),
}
)
Expand All @@ -1703,7 +1761,13 @@ def us_ssi_take_up_delivery_gate(
failures=tuple(failures),
details={
"relative_tolerance": tolerance,
"enforced_band_keys": list(US_SSI_TAKE_UP_ENFORCED_BAND_KEYS),
# Effective enforcement for THIS run: the constant minus any
# adjudication-fenced bands — reporting the constant here would
# misdocument a fenced band as an enforced contract.
"enforced_band_keys": [
key for key in US_SSI_TAKE_UP_ENFORCED_BAND_KEYS if key not in fences
],
"adjudication_fenced_band_keys": sorted(fences),
"enforced_bands": enforced_rows,
"fenced_bands": fenced_rows,
},
Expand Down
123 changes: 119 additions & 4 deletions packages/populace-build/tests/test_us_fiscal_refresh_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,110 @@ def test_certified_release_dir_reuse_is_refused(tmp_path) -> None:
builder._refuse_certified_release_dir_reuse(release_dir)


def test_dense_ssi_fences_cover_the_enforced_bands_and_cite_the_adjudication() -> None:
"""populace#566/#567: the dense-arm fence table must cover exactly the
normally-enforced bands, and every fence must carry the recompute
adjudication, its re-adjudication trigger, and the sparse contrast —
a fence without its documented reason is forbidden."""
from populace.build.us_runtime.ssi_take_up import (
US_SSI_TAKE_UP_ENFORCED_BAND_KEYS,
)

builder = _load_builder_module()
fences = builder.US_DENSE_SSI_TAKE_UP_ENFORCEMENT_FENCES
assert set(fences) == set(US_SSI_TAKE_UP_ENFORCED_BAND_KEYS)
for band, text in fences.items():
assert "populace#566/#567" in text, band
assert "populace#508" in text, band
assert "Re-adjudicates" in text, band
assert "sparse certified" in text, band
# The adjudication must claim only what the artifacts support:
# recomputes failed to land the pair in band — NOT a proven
# periodic map (sol round-1 blocker 3). The refusal scans the
# WHOLE production sources, not just the constant, so a stray
# comment cannot reintroduce the overclaim (sol round 2).
assert "oscillat" not in text.lower(), band
import populace.build.us_runtime.ssi_take_up as ssi_module

for source_path in (Path(builder.__file__), Path(ssi_module.__file__)):
assert "oscillat" not in source_path.read_text().lower(), source_path


def test_ssi_delivery_fences_are_passed_on_the_dense_arm_only() -> None:
"""The sparse certified arm must keep hard enforcement: structurally,
main()'s single _enforce_ssi_take_up_delivery call may pass the fence
table only under args.dense_default_dataset, with None otherwise (the
#443 AST-guard pattern)."""
import ast

builder = _load_builder_module()
tree = ast.parse(Path(builder.__file__).read_text())
main_fn = next(
n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == "main"
)
calls = [
n
for n in ast.walk(main_fn)
if isinstance(n, ast.Call)
and getattr(n.func, "id", "") == "_enforce_ssi_take_up_delivery"
]
assert len(calls) == 1, "exactly one delivery-enforcement call site"
fence_kwargs = [kw for kw in calls[0].keywords if kw.arg == "enforcement_fences"]
assert len(fence_kwargs) == 1, "the call site must pass enforcement_fences"
value = fence_kwargs[0].value
assert isinstance(value, ast.IfExp), "fences must be arm-conditional"
assert isinstance(value.test, ast.Attribute)
assert getattr(value.test.value, "id", "") == "args"
assert value.test.attr == "dense_default_dataset"
assert getattr(value.body, "id", "") == "US_DENSE_SSI_TAKE_UP_ENFORCEMENT_FENCES"
assert isinstance(value.orelse, ast.Constant) and value.orelse.value is None


def test_delivery_gate_result_reaches_the_manifest_gates_block() -> None:
"""A release built with fences must be distinguishable from one whose
bands passed enforcement: the delivery gate result (effective enforced
set + fenced rows with adjudication text) must ride the manifest gates
block, and main() must thread it into _build_manifests."""
import ast

builder = _load_builder_module()
tree = ast.parse(Path(builder.__file__).read_text())
main_fn = next(
n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == "main"
)
manifest_calls = [
n
for n in ast.walk(main_fn)
if isinstance(n, ast.Call) and getattr(n.func, "id", "") == "_build_manifests"
]
assert manifest_calls, "main() must call _build_manifests"
assert all(
any(kw.arg == "ssi_take_up_delivery_gate_result" for kw in call.keywords)
for call in manifest_calls
), "every _build_manifests call must thread the delivery gate result"
build_fn = next(
n
for n in ast.walk(tree)
if isinstance(n, ast.FunctionDef) and n.name == "_build_manifests"
)
carrier_dicts = [
node
for node in ast.walk(build_fn)
if isinstance(node, ast.Dict)
and any(
isinstance(key, ast.Constant) and key.value == "ssi_take_up_delivery"
for key in node.keys
)
]
assert len(carrier_dicts) >= 2, (
"BOTH manifest writers (build_manifest.json gates block AND "
"release_manifest.json build section) must record the "
"ssi_take_up_delivery receipt — release_manifest.json alone has "
f"to distinguish fenced from enforced delivery; found "
f"{len(carrier_dicts)} carrier dict(s)"
)


def test_final_household_weight_evidence_writes_only_on_gate_failure_path() -> None:
"""populace#568 review blocker 2: the evidence pair must be written on
the batched gate-failure path ONLY — green runs carry weights in the
Expand Down Expand Up @@ -4656,9 +4760,11 @@ def fake_final_medicaid_diagnostics(
fake_final_ssi_diagnostics,
)

def fake_ssi_delivery_gate(diagnostics, *, targets):
def fake_ssi_delivery_gate(diagnostics, *, targets, enforcement_fences=None):
captured["ssi_delivery_gate_called"] = True
captured["ssi_delivery_gate_targets"] = dict(targets)
# The sparse e2e paths must never see dense fences (populace#566/#567).
captured["ssi_delivery_gate_enforcement_fences"] = enforcement_fences
# The integrity and retirement cases pass delivery to isolate their
# own early failure. Other modes retain the populace#547 delivery
# cofailure and its written retry basis.
Expand Down Expand Up @@ -5096,6 +5202,9 @@ def fake_release_gate_failures(*args, **kwargs):
assert final_basis.kind == "current_frame"
assert final_basis.band("65_plus").candidate_capacity == pytest.approx(1_000.0)
assert captured["ssi_delivery_gate_called"] is True
# This e2e harness runs the sparse arm: the dense-only enforcement
# fences must never reach the gate here (populace#566/#567).
assert captured["ssi_delivery_gate_enforcement_fences"] is None
assert captured["ssi_delivery_gate_targets"] == fake_band_targets
# The frozen-assignment digest invalidates the materialization cache on
# any retry whose flags differ (populace#507/#508 split-brain fix).
Expand Down Expand Up @@ -9440,14 +9549,16 @@ def test_enforce_ssi_delivery_returns_batch_failures_and_writes_the_basis(
release_dir = tmp_path / "release"
release_dir.mkdir()

failures = builder._enforce_ssi_take_up_delivery(
failures, gate_result = builder._enforce_ssi_take_up_delivery(
diagnostics,
targets=_SSI_BAND_TARGETS,
release_dir=release_dir,
telemetry=None,
)

assert failures
# The returned gate result is the manifest receipt for this run.
assert not gate_result.passed
assert all(failure.startswith("SSI take-up deliver") for failure in failures)
assert any("--ssi-take-up-prior-weight-basis" in failure for failure in failures)
written_path = release_dir / "us_ssi_take_up.json"
Expand Down Expand Up @@ -9476,14 +9587,17 @@ def test_enforce_ssi_delivery_passes_in_tolerance_and_writes_nothing(
release_dir = tmp_path / "release"
release_dir.mkdir()

failures = builder._enforce_ssi_take_up_delivery(
failures, gate_result = builder._enforce_ssi_take_up_delivery(
diagnostics,
targets=_SSI_BAND_TARGETS,
release_dir=release_dir,
telemetry=None,
)

assert failures == []
assert gate_result.passed
# No fences on this sparse-shaped call: full enforcement documented.
assert gate_result.details["adjudication_fenced_band_keys"] == []
assert not (release_dir / "us_ssi_take_up.json").exists()


Expand All @@ -9505,14 +9619,15 @@ def test_enforce_ssi_delivery_survives_unwritable_retry_artifact(
release_dir = tmp_path / "release"
release_dir.mkdir()

failures = builder._enforce_ssi_take_up_delivery(
failures, gate_result = builder._enforce_ssi_take_up_delivery(
diagnostics,
targets=_SSI_BAND_TARGETS,
release_dir=release_dir,
telemetry=None,
)

assert failures
assert not gate_result.passed
assert failures[0].startswith("SSI take-up delivery failed:")
assert any("could NOT be written" in failure for failure in failures)
# json.dumps runs before write_text, so no partial artifact exists.
Expand Down
Loading
Loading