From ef53849f93a02d024bed60c86b0bb6fdd8e281e1 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Tue, 22 Sep 2026 14:33:22 -0400 Subject: [PATCH 1/2] Bridge retained earnings histories into actual Axiom AIME/PIA execution Add populace_dynamics.axiom_benefit_bridge, an opt-in transport from retained cohort histories to the real axiom-rules-engine `run-lifetime` binary. It builds lifetime-request/v2 requests in the retained SSA Case A runner's format, runs the pinned engine (commit 49f2225, the binary that reproduced Case A) as a subprocess against pinned compiled artifacts, parses and validates the typed response, and returns per-person results carrying provenance: engine, artifact and candidate-module SHA-256s, request and stdout digests, selected rule versions, and the fact that the RuleSpec candidates are unaccepted. The bridge performs no benefit arithmetic; indexing, top-N selection, averaging and the PIA formula run inside the engine. The bridge refuses rather than guesses: - every elapsed-window year through the year before entitlement must be supplied; gaps raise BenefitBridgeRefusal listing the years (a leading gap may be accepted only by an exact, recorded acknowledgement); - pre-2014 years come only from an explicit CallerSuppliedHistory whose synthetic flag travels onto every row; otherwise the gap is recorded; - projected labor income is used as creditable earnings only under an explicit ProjectedAmountDeclaration matching the history's unit; - missing birth years, wage-index years, contribution-base checks, and mortality removal before entitlement all refuse; - the PIA output is requested only for the candidate's 2026 first-eligibility cohort; other people get AIME from the AIME-only artifact compiled from the same 415(b) module. The engine parses decimals with rust_decimal's from_str_exact, which rejects exact expansions of most fractional binary64 values, so the binary64-to-decimal encoding is an explicit caller choice recorded per row. Tests: fake-engine unit tests for request construction, refusals, response validation, binding tampering and the subprocess path, plus cohort tests on invented people produced by actual engine steps. Integration tests (skipped without the engine) rebuild SSA's published Case A and Case B requests byte-for-byte through the bridge and reproduce AIME $5,825 / PIA $2,609.80 and AIME $11,463 with identical stdout. Register the module in POST_REVIEW_SOURCE_EXCLUSIONS, its pinned tuple and the reachability guard. Remove a byte-identical duplicate of test_source_reachability_includes_implicit_package_initializers left by the stack reconciliation (ruff F811; the second copy already shadowed the first, so collection is unchanged). Tier manifest recollected: unit 2,246, artifact 2,672, integration_psid 848, reproduction_legacy 520, oracle_policyengine 159; total 6,445. Co-Authored-By: Claude Opus 5.5 --- scripts/first_estimates_birth_evidence.py | 3 + src/populace_dynamics/axiom_benefit_bridge.py | 1554 +++++++++++++++++ tests/README-tiers.md | 4 +- .../estimates/test_birth_evidence_artifact.py | 47 +- tests/test_axiom_benefit_bridge.py | 952 ++++++++++ tests/tier_counts.json | 2 +- 6 files changed, 2519 insertions(+), 43 deletions(-) create mode 100644 src/populace_dynamics/axiom_benefit_bridge.py create mode 100644 tests/test_axiom_benefit_bridge.py diff --git a/scripts/first_estimates_birth_evidence.py b/scripts/first_estimates_birth_evidence.py index 5967bea3..7c2ad5c1 100644 --- a/scripts/first_estimates_birth_evidence.py +++ b/scripts/first_estimates_birth_evidence.py @@ -188,6 +188,9 @@ Path("src/populace_dynamics/closed_cohort_history.py"), Path("src/populace_dynamics/assembled_history_observer.py"), Path("src/populace_dynamics/compact_cohort_history.py"), + # The opt-in Axiom benefit bridge runs an external engine on retained + # histories and is outside the historical projection call graph. + Path("src/populace_dynamics/axiom_benefit_bridge.py"), ) POST_REVIEW_SHARED_SOURCE_BLOBS = { Path( diff --git a/src/populace_dynamics/axiom_benefit_bridge.py b/src/populace_dynamics/axiom_benefit_bridge.py new file mode 100644 index 00000000..8d63c6a9 --- /dev/null +++ b/src/populace_dynamics/axiom_benefit_bridge.py @@ -0,0 +1,1554 @@ +"""Run retained earnings histories through the actual Axiom AIME/PIA engine. + +The bridge transports facts; it does not calculate benefits. It builds +``axiom-rules-engine/lifetime-request/v2`` requests in the format of the +retained SSA Case A runner, executes the real ``run-lifetime`` binary as a +subprocess against pinned compiled artifacts, and parses the typed response. +Wage indexing, computation-year selection, averaging and the ordinary PIA +formula all happen inside the engine. + +The bound RuleSpec modules are unaccepted encoder candidates. Numerical +agreement with a published example is diagnostic only. When a person lacks +an input the candidates need, the bridge raises +:class:`BenefitBridgeRefusal` instead of guessing. It never pads missing +years with zeros, never treats projected labor income as creditable +earnings without an explicit caller declaration, and requests the PIA output +only for the candidate's supplied 2026 first-eligibility cohort. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import subprocess +from collections.abc import Callable, Iterable, Mapping, Sequence +from dataclasses import dataclass, field +from decimal import Decimal, InvalidOperation +from numbers import Integral +from pathlib import Path + +from .closed_cohort_history import ClosedCohortEarningsHistory + +AIME_OUTPUT = "us:statutes/42/415/b#average_indexed_monthly_earnings" +PIA_OUTPUT = "us:statutes/42/415/a#ordinary_pia_before_cola" +AIME_INPUT_PREFIX = "us:statutes/42/415/b#input." +PIA_COHORT_INPUT = ( + "us:statutes/42/415/a#input.supplied_2026_first_eligibility_cohort" +) +# The 415(a) candidate gates its formula on a supplied 2026 +# first-eligibility flag and imports only the 2026 bend points. +PIA_CANDIDATE_ELIGIBILITY_YEAR = 2026 +# The 415(b) candidate counts computation base years after its 1950 +# ``elapsed_years_baseline_year`` parameter. +FIRST_COMPUTATION_BASE_YEAR = 1951 +REQUEST_SCHEMA = "axiom-rules-engine/lifetime-request/v2" +RESPONSE_SCHEMA = "axiom-rules-engine/lifetime-response/v2" +ARTIFACT_AIME_PIA = "aime_pia" +ARTIFACT_AIME_ONLY = "aime_only" +CANDIDATE_STATUS = ( + "unaccepted encoder candidates: 42 USC 415(b) AIME and a conditional " + "2026-cohort 415(a) ordinary PIA; the combined graph's import-proof " + "metadata is stale and proof validation has not passed" +) + +# Case A supplied these conditions as scoped assumptions, not findings. +SCOPED_ASSUMPTIONS: Mapping[str, bool] = { + "individual_is_entitled_to_old_age_insurance_benefit": True, + "individual_is_entitled_to_disability_insurance_benefit": False, + "individual_previously_entitled_to_disability_insurance_benefit": False, + ( + "at_least_twelve_consecutive_months_without_disability_or_old_age_" + "benefit_before_current_eligibility" + ): False, + "computation_base_year_is_entirely_in_period_of_disability": False, + "elapsed_year_is_partly_in_period_of_disability": False, +} + +# Reviewed identities of the engine and candidate files that reproduced the +# published SSA Case A AIME ($5,825) and initial PIA ($2,609.80). +REVIEWED_ENGINE_SOURCE_COMMIT = "49f2225c2788b242f961e1eaee41f75cf703f685" +REVIEWED_PINS: Mapping[str, str] = { + "engine": ( + "e41161754874789a84d609f243e1db92948fd19806555bc942ab2048946c8dea" + ), + "engine_provenance_receipt": ( + "469d13fd0ecfa9995f128f0129e695f47111d3baebd1e31466c2f1e0162ecd3c" + ), + "engine_binding_manifest": ( + "0b6515e5e6f2650899c757a3683c8a9faa1819f05bef857a942b278ddda196c0" + ), + ARTIFACT_AIME_PIA: ( + "10d32e4284f64f63a62c4e256ed059cb49e2d1b57118c7b8004bd9076c46f3a8" + ), + ARTIFACT_AIME_ONLY: ( + "ed1fdb444c41e88944e76edb9fe66d6dc9849876ae8265d9a860eb8c76fb5643" + ), + "aime_pia_compile_receipt": ( + "b5dc18fe554da11c065181d3b111bb8762487c7b6a2b0d5c80edfd26a387466a" + ), + "module:us/statutes/42/415/b.yaml": ( + "24ddf3b6810f68a77ae0f9295f98994ed4a3d24a46a94609045296661a477318" + ), + "module:us/statutes/42/415/a.yaml": ( + "81487dbc6115d7513f39e7de714f8bf6886cbb06562a5784336f93617c616dda" + ), + "module:us/policies/ssa/pia-bend-points/2026.yaml": ( + "dd39d3ceda711c0eca7e2a752561c368efd59d64dbf35343b16d1cd5e7c882e0" + ), + "aime_only_source:us/statutes/42/415/b.yaml": ( + "24ddf3b6810f68a77ae0f9295f98994ed4a3d24a46a94609045296661a477318" + ), +} +ENGINE_ROOT_ENV = "POPULACE_DYNAMICS_AXIOM_ENGINE_ROOT" +EVIDENCE_DIR_ENV = "POPULACE_DYNAMICS_AXIOM_EVIDENCE_DIR" +DEFAULT_ENGINE_ROOT = "~/TheAxiomFoundation/_worktrees/" + ( + "dynamics-lazy-lifetime-20260922" +) +DEFAULT_EVIDENCE_DIR = "~/microcosm-launch-evidence/" + ( + "dynasim-parity-20260909/parallel-oasdi-20260920/encoder-recovery" +) + +# The engine parses decimal inputs with rust_decimal's +# ``Decimal::from_str_exact`` (axiom-rules-engine src/lifetime_api.rs), which +# rejects the long exact expansions of most fractional binary64 values. +# Converting projected binary64 amounts is therefore an explicit caller +# choice, recorded on every row. +BINARY64_EXACT = "exact_binary64" +BINARY64_SHORTEST_ROUND_TRIP = "shortest_round_trip_binary64" +BINARY64_ENCODINGS = (BINARY64_EXACT, BINARY64_SHORTEST_ROUND_TRIP) + +_DECIMAL_TEXT = re.compile(r"-?(?:0|[1-9][0-9]*)(?:\.[0-9]+)?") +_SHA256 = re.compile(r"[0-9a-f]{64}") +_SUBPROCESS_ENV = {"PATH": "/usr/bin:/bin"} + + +class BenefitBridgeRefusal(ValueError): + """The rule needs an input this person lacks; nothing was guessed.""" + + def __init__( + self, + subject: str, + reasons: Sequence[str], + *, + gaps: Mapping[str, Sequence[int]] | None = None, + ) -> None: + self.subject = subject + self.reasons = tuple(reasons) + self.gaps = { + name: tuple(years) for name, years in (gaps or {}).items() + } + if not self.reasons: + raise ValueError("a refusal needs at least one reason") + super().__init__(f"{subject}: " + "; ".join(self.reasons)) + + +class AxiomExecutionError(RuntimeError): + """The engine failed or returned a response that does not validate.""" + + +class BindingMismatch(RuntimeError): + """A bound engine, artifact or candidate file is absent or changed.""" + + +def _sha256_file(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _sha256_bytes(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _integer(value: object, label: str) -> int: + if isinstance(value, bool) or not isinstance(value, Integral): + raise TypeError(f"{label} must be an integer without coercion") + return int(value) + + +def _text(value: object, label: str) -> str: + if type(value) is not str or not value.strip(): + raise ValueError(f"{label} must be a nonempty string") + return value + + +def _amount(value: object, label: str) -> Decimal: + """Accept exact decimal inputs only; binary floats need explicit bits.""" + if isinstance(value, bool): + raise TypeError(f"{label} cannot be a boolean") + if isinstance(value, Decimal): + amount = value + elif isinstance(value, Integral): + amount = Decimal(int(value)) + elif type(value) is str and _DECIMAL_TEXT.fullmatch(value): + amount = Decimal(value) + else: + raise TypeError( + f"{label} must be a Decimal, integer or plain decimal string" + ) + if not amount.is_finite() or amount < 0: + raise ValueError(f"{label} must be finite and nonnegative") + return amount + + +def _decimal_text(amount: Decimal) -> str: + """Plain decimal text accepted by the engine, never exponent notation.""" + text = "0" if amount == 0 else format(amount, "f") + if not _DECIMAL_TEXT.fullmatch(text): + raise ValueError(f"amount {amount!r} has no plain decimal form") + return text + + +def binary64_decimal(amount_hex: str, encoding: str) -> Decimal: + """Decimal for binary64 source bits under an explicit encoding. + + ``exact_binary64`` is the exact value of the bits. The + ``shortest_round_trip_binary64`` encoding is Python's shortest decimal + that converts back to the same bits; it can differ from the exact value + by less than half a unit in the last binary64 place. + """ + if type(amount_hex) is not str: + raise TypeError("binary64 amount must be float.hex() text") + value = float.fromhex(amount_hex) + if value.hex() != amount_hex: + raise ValueError("binary64 amount must be canonical float.hex() text") + if encoding == BINARY64_EXACT: + return Decimal(value) + if encoding == BINARY64_SHORTEST_ROUND_TRIP: + shortest = Decimal(repr(value)) + if shortest == shortest.to_integral_value(): + shortest = shortest.quantize(Decimal(1)) + if float(shortest) != value: + raise ValueError("shortest decimal does not round-trip") + return shortest + raise ValueError(f"unsupported binary64 encoding {encoding!r}") + + +def year_ranges(years: Iterable[int]) -> str: + """Compress sorted years into ``1974-2013, 2023`` style text.""" + ordered = sorted(set(years)) + if not ordered: + return "none" + spans, start, previous = [], ordered[0], ordered[0] + for year in ordered[1:]: + if year != previous + 1: + spans.append((start, previous)) + start = year + previous = year + spans.append((start, previous)) + return ", ".join(str(a) if a == b else f"{a}-{b}" for a, b in spans) + + +def _year_period(year: int) -> dict[str, str]: + return { + "period_kind": "custom", + "name": "calendar_year", + "start": f"{year}-01-01", + "end": f"{year}-12-31", + } + + +@dataclass(frozen=True) +class AnnualSeries: + """Exact published decimal strings by calendar year, never floats.""" + + label: str + values: tuple[tuple[int, str], ...] + source_sha256: str | None = None + + def __post_init__(self) -> None: + _text(self.label, "series label") + if self.source_sha256 is not None and ( + type(self.source_sha256) is not str + or not _SHA256.fullmatch(self.source_sha256) + ): + raise ValueError("series source digest must be SHA-256 hex") + rows = [] + for year, value in self.values: + year = _integer(year, "series year") + if type(value) is not str or not _DECIMAL_TEXT.fullmatch(value): + raise ValueError(f"{year} value must be a decimal string") + if Decimal(value) <= 0: + raise ValueError(f"{year} value must be positive") + rows.append((year, value)) + rows.sort() + if len({year for year, _ in rows}) != len(rows): + raise ValueError("series years must be unique") + object.__setattr__(self, "values", tuple(rows)) + + @classmethod + def from_mapping( + cls, + label: str, + values: Mapping[int, str], + *, + source_sha256: str | None = None, + ) -> AnnualSeries: + return cls(label, tuple(values.items()), source_sha256) + + def get(self, year: int) -> str | None: + return dict(self.values).get(year) + + def document(self) -> dict[str, object]: + years = [year for year, _ in self.values] + return { + "label": self.label, + "source_sha256": self.source_sha256, + "years": year_ranges(years), + } + + +@dataclass(frozen=True) +class EarningsRow: + """One annual nominal amount as transported, with its provenance.""" + + year: int + amount: Decimal + source: str + synthetic: bool + amount_hex: str | None = None + encoding: str | None = None + + def __post_init__(self) -> None: + object.__setattr__(self, "year", _integer(self.year, "row year")) + object.__setattr__(self, "amount", _amount(self.amount, "amount")) + _text(self.source, "row source") + if type(self.synthetic) is not bool: + raise TypeError("synthetic must be an explicit bool") + if self.amount_hex is None: + if self.encoding is not None: + raise ValueError("an encoding needs binary64 source bits") + elif binary64_decimal(self.amount_hex, self.encoding) != self.amount: + raise ValueError("amount must equal its encoded binary64 bits") + + def document(self) -> dict[str, object]: + return { + "year": self.year, + "amount": _decimal_text(self.amount), + "source": self.source, + "synthetic": self.synthetic, + "amount_hex": self.amount_hex, + "encoding": self.encoding, + } + + +@dataclass(frozen=True) +class CallerSuppliedHistory: + """An explicit annual history supplied by the caller, labeled at source. + + Use it for pre-projection years the retained projection does not + contain. Invented rows must set ``synthetic=True``; the flag travels + into every result. + """ + + label: str + synthetic: bool + amounts: tuple[tuple[int, Decimal], ...] + + def __post_init__(self) -> None: + _text(self.label, "supplied history label") + if type(self.synthetic) is not bool: + raise TypeError("synthetic must be an explicit bool") + rows = sorted( + (_integer(year, "supplied year"), _amount(amount, "amount")) + for year, amount in self.amounts + ) + if len({year for year, _ in rows}) != len(rows): + raise ValueError("supplied years must be unique") + object.__setattr__(self, "amounts", tuple(rows)) + + @classmethod + def from_mapping( + cls, + label: str, + amounts: Mapping[int, object], + *, + synthetic: bool, + ) -> CallerSuppliedHistory: + return cls(label, synthetic, tuple(amounts.items())) + + def rows(self) -> tuple[EarningsRow, ...]: + prefix = ( + "synthetic_caller_supplied" + if self.synthetic + else ("caller_supplied") + ) + return tuple( + EarningsRow(year, amount, f"{prefix}:{self.label}", self.synthetic) + for year, amount in self.amounts + ) + + +@dataclass(frozen=True) +class LeadingGapAcknowledgement: + """Explicit acceptance that leading elapsed-window years are absent. + + Only for sources (such as SSA's published worked examples) that list no + earlier rows. The years are recorded as unsupplied; the engine receives + no rows for them, so they cannot enter its top-N selection. + """ + + years: tuple[int, ...] + statement: str + + def __post_init__(self) -> None: + years = tuple(sorted(_integer(y, "gap year") for y in self.years)) + if not years or years != tuple(range(years[0], years[-1] + 1)): + raise ValueError("acknowledged years must be one contiguous run") + _text(self.statement, "acknowledgement statement") + object.__setattr__(self, "years", years) + + +@dataclass(frozen=True) +class OrdinaryRetirementDeterminations: + """Calendar facts the candidates take as inputs but do not derive.""" + + birth_year: int + first_old_age_entitlement_year: int + year_attained_age_21: int + year_attained_age_62: int + indexing_year: int + death_or_later_sentinel_year: int + derivations: tuple[str, ...] = () + + def __post_init__(self) -> None: + for name in ( + "birth_year", + "first_old_age_entitlement_year", + "year_attained_age_21", + "year_attained_age_62", + "indexing_year", + "death_or_later_sentinel_year", + ): + object.__setattr__(self, name, _integer(getattr(self, name), name)) + reasons = [] + if self.year_attained_age_62 != self.year_attained_age_21 + 41: + reasons.append("ages 21 and 62 must be attained 41 years apart") + if self.first_old_age_entitlement_year < self.year_attained_age_62: + reasons.append( + "ordinary old-age entitlement year " + f"{self.first_old_age_entitlement_year} precedes the year " + f"of attaining 62 ({self.year_attained_age_62})" + ) + if self.indexing_year >= self.first_old_age_entitlement_year: + reasons.append("indexing year must precede entitlement") + if ( + self.death_or_later_sentinel_year + < self.first_old_age_entitlement_year + ): + reasons.append( + "death-or-later sentinel precedes entitlement; this bridge " + "supports living ordinary old-age entitlement only" + ) + if reasons: + raise BenefitBridgeRefusal( + f"birth year {self.birth_year}", reasons + ) + object.__setattr__(self, "derivations", tuple(self.derivations)) + + @classmethod + def from_birth_year( + cls, birth_year: int, entitlement_year: int + ) -> OrdinaryRetirementDeterminations: + """Apply the retained Case A/B calendar conventions to a birth year.""" + birth_year = _integer(birth_year, "birth year") + entitlement_year = _integer(entitlement_year, "entitlement year") + return cls( + birth_year, + entitlement_year, + birth_year + 21, + birth_year + 62, + birth_year + 60, + entitlement_year, + ( + "year_individual_attained_age_21 = birth year + 21", + "year_individual_attained_age_62 = birth year + 62", + "indexing_year_for_aime = year of attaining 62 minus 2; " + "the 415(b) candidate defers indexing-year selection, and " + "this matches the retained Case A (2024) and Case B (2019) " + "inputs", + "year_of_death_or_later_sentinel = first entitlement year, " + "a later-boundary sentinel as in the Case A/B runners, not " + "a death assertion", + "birth year only: SSA treats an age as attained on the day " + "before the birthday (retained POMS RS 00615.015 extract), " + "so a January 1 birth would attain each age one calendar " + "year earlier; exact birth dates are unavailable", + ), + ) + + def document(self) -> dict[str, object]: + return { + "birth_year": self.birth_year, + "first_old_age_entitlement_year": ( + self.first_old_age_entitlement_year + ), + "year_attained_age_21": self.year_attained_age_21, + "year_attained_age_62": self.year_attained_age_62, + "indexing_year": self.indexing_year, + "death_or_later_sentinel_year": self.death_or_later_sentinel_year, + "derivations": list(self.derivations), + } + + +@dataclass(frozen=True) +class PreparedRequest: + """A validated lifetime request and everything recorded about it.""" + + person_id: str + request: Mapping[str, object] + wire: bytes + artifact_role: str + outputs: tuple[str, ...] + rows: tuple[EarningsRow, ...] + determinations: OrdinaryRetirementDeterminations + recorded_gaps: tuple[Mapping[str, object], ...] + wage_index: AnnualSeries + notes: tuple[str, ...] = () + + @property + def pia_requested(self) -> bool: + return PIA_OUTPUT in self.outputs + + @property + def calculation_year(self) -> int: + return self.determinations.first_old_age_entitlement_year + + @property + def request_sha256(self) -> str: + return _sha256_bytes(self.wire) + + +def _window_start(determinations: OrdinaryRetirementDeterminations) -> int: + return max( + FIRST_COMPUTATION_BASE_YEAR, determinations.year_attained_age_21 + 1 + ) + + +def _missing_wage_index_years( + years: Iterable[int], indexing_year: int, wage_index: AnnualSeries +) -> tuple[int, ...]: + needed = {year for year in years if year <= indexing_year} + needed.add(indexing_year) + return tuple(sorted(y for y in needed if wage_index.get(y) is None)) + + +def prepare_request( + person_id: str, + rows: Sequence[EarningsRow], + determinations: OrdinaryRetirementDeterminations, + wage_index: AnnualSeries, + *, + contribution_base: AnnualSeries | None = None, + leading_gap: LeadingGapAcknowledgement | None = None, + notes: Sequence[str] = (), +) -> PreparedRequest: + """Build one person's request exactly as the Case A runner did. + + Every year from the start of the elapsed-year window (the year after + attaining 21, and after 1950) through the year before entitlement must be + supplied, unless a leading run of those years is explicitly acknowledged. + Computation base years before the elapsed-year window may be absent; + they are recorded, never filled. + """ + _text(person_id, "person id") + if type(determinations) is not OrdinaryRetirementDeterminations: + raise TypeError("explicit OrdinaryRetirementDeterminations required") + if type(wage_index) is not AnnualSeries: + raise TypeError("explicit wage-index AnnualSeries required") + rows = tuple(sorted(rows, key=lambda row: row.year)) + if any(type(row) is not EarningsRow for row in rows): + raise TypeError("EarningsRow values required") + entitlement = determinations.first_old_age_entitlement_year + index_year = determinations.indexing_year + reasons: list[str] = [] + gaps: dict[str, tuple[int, ...]] = {} + years = [row.year for row in rows] + if len(set(years)) != len(years): + raise ValueError("each year may be supplied once") + outside = [ + y for y in years if not FIRST_COMPUTATION_BASE_YEAR <= y < entitlement + ] + if outside: + reasons.append( + f"years {year_ranges(outside)} are not computation base years " + f"before the {entitlement} entitlement year" + ) + window = range(_window_start(determinations), entitlement) + supplied = set(years) + missing = [y for y in window if y not in supplied] + acknowledged: tuple[int, ...] = () + if leading_gap is not None: + if type(leading_gap) is not LeadingGapAcknowledgement: + raise TypeError("LeadingGapAcknowledgement required") + first = min(supplied) if supplied else entitlement + leading = tuple(y for y in missing if y < first) + if leading_gap.years != leading or not leading: + reasons.append( + "acknowledged years " + f"{year_ranges(leading_gap.years)} do not equal the " + f"unsupplied leading elapsed-window years " + f"{year_ranges(leading)}" + ) + else: + acknowledged = leading + missing = [y for y in missing if y not in acknowledged] + if missing: + reasons.append( + f"required years {year_ranges(missing)} are absent; the " + f"elapsed-year window runs {window.start}-{window.stop - 1} and " + "no year may be padded" + ) + gaps["missing_required_years"] = tuple(missing) + if years: + holes = [ + y for y in range(min(years), max(years) + 1) if y not in supplied + ] + holes = [y for y in holes if y not in missing] + if holes: + reasons.append(f"interior years {year_ranges(holes)} are absent") + gaps["missing_interior_years"] = tuple(holes) + else: + reasons.append("no earnings rows were supplied") + missing_index = _missing_wage_index_years( + (supplied | set(window)) - set(acknowledged), index_year, wage_index + ) + if missing_index: + reasons.append( + "national average wage index values are absent for " + f"{year_ranges(missing_index)} (indexing year {index_year})" + ) + gaps["missing_wage_index_years"] = missing_index + recorded: list[dict[str, object]] = [] + if contribution_base is not None: + if type(contribution_base) is not AnnualSeries: + raise TypeError("contribution base must be an AnnualSeries") + unchecked, above = [], [] + for row in rows: + if row.amount == 0: + continue + ceiling = contribution_base.get(row.year) + if ceiling is None: + unchecked.append(row.year) + elif row.amount > Decimal(ceiling): + above.append(row.year) + if unchecked: + reasons.append( + "no contribution and benefit base is available to check " + f"nonzero amounts in {year_ranges(unchecked)}" + ) + gaps["unchecked_creditable_ceiling_years"] = tuple(unchecked) + if above: + reasons.append( + f"amounts in {year_ranges(above)} exceed the contribution and " + "benefit base; the bridge does not limit creditable earnings" + ) + gaps["above_creditable_ceiling_years"] = tuple(above) + else: + recorded.append( + { + "kind": "creditable_ceiling_not_checked", + "effect": "no contribution and benefit base series was " + "supplied, so amounts were not checked against it", + } + ) + if reasons: + raise BenefitBridgeRefusal(person_id, reasons, gaps=gaps) + + first_supplied = min(years) + early = range( + max(FIRST_COMPUTATION_BASE_YEAR, determinations.birth_year), + min(first_supplied, window.start), + ) + if len(early): + recorded.append( + { + "kind": "computation_base_years_before_elapsed_window_" + "not_supplied", + "years": year_ranges(early), + "effect": "no engine rows; these years cannot enter the " + "top-N selection", + } + ) + if acknowledged: + recorded.append( + { + "kind": "acknowledged_unsupplied_elapsed_window_years", + "years": year_ranges(acknowledged), + "statement": leading_gap.statement, + "effect": "no engine rows; these years cannot enter the " + "top-N selection", + } + ) + pia = determinations.year_attained_age_62 == PIA_CANDIDATE_ELIGIBILITY_YEAR + outputs = (AIME_OUTPUT, PIA_OUTPUT) if pia else (AIME_OUTPUT,) + batches = [] + for row in rows: + values: dict[str, tuple[str, object]] = { + name: ("bool", value) for name, value in SCOPED_ASSUMPTIONS.items() + } + values.update( + { + "computation_base_year": ("integer", row.year), + "creditable_earnings_for_computation_base_year": ( + "decimal", + _decimal_text(row.amount), + ), + "national_average_wage_index_for_indexing_year": ( + "decimal", + wage_index.get(index_year), + ), + "indexing_year_for_aime": ("integer", index_year), + "year_individual_attained_age_21": ( + "integer", + determinations.year_attained_age_21, + ), + "year_individual_attained_age_62": ( + "integer", + determinations.year_attained_age_62, + ), + "year_of_death_or_later_sentinel": ( + "integer", + determinations.death_or_later_sentinel_year, + ), + "year_of_first_old_age_entitlement": ("integer", entitlement), + } + ) + if row.year <= index_year: + # Case A/B omit the per-year index after the indexing year. + values["national_average_wage_index_for_computation_base_year"] = ( + "decimal", + wage_index.get(row.year), + ) + inputs = { + AIME_INPUT_PREFIX + name: {"kind": kind, "values": [value]} + for name, (kind, value) in values.items() + } + if pia: + inputs[PIA_COHORT_INPUT] = {"kind": "bool", "values": [True]} + batches.append( + {"row_count": 1, "entity_ids": [person_id], "inputs": inputs} + ) + calculation = _year_period(entitlement) + request = { + "schema": REQUEST_SCHEMA, + "entity": "Person", + "arithmetic": "decimal", + "periods": [_year_period(row.year) for row in rows], + "batches": batches, + "calculation_period": calculation, + "outputs": list(outputs), + "output_period": calculation, + } + wire = ( + json.dumps(request, sort_keys=True, separators=(",", ":")) + "\n" + ).encode() + return PreparedRequest( + person_id, + request, + wire, + ARTIFACT_AIME_PIA if pia else ARTIFACT_AIME_ONLY, + outputs, + rows, + determinations, + tuple(recorded), + wage_index, + tuple(notes), + ) + + +@dataclass(frozen=True) +class BoundFile: + """A file whose SHA-256 must match before and after every execution.""" + + role: str + path: Path + sha256: str + + def __post_init__(self) -> None: + _text(self.role, "bound role") + object.__setattr__(self, "path", Path(self.path)) + if type(self.sha256) is not str or not _SHA256.fullmatch(self.sha256): + raise ValueError("bound digest must be lowercase SHA-256 hex") + + def document(self) -> dict[str, str]: + return { + "role": self.role, + "path": str(self.path), + "sha256": self.sha256, + } + + +@dataclass(frozen=True) +class AxiomEngineBinding: + """The engine binary, compiled artifacts and candidate sources in use.""" + + engine: BoundFile + artifacts: tuple[BoundFile, ...] + supporting: tuple[BoundFile, ...] + engine_source_commit: str + candidate_status: str = CANDIDATE_STATUS + candidates_accepted: bool = False + engine_manifest_role: str | None = None + timeout_seconds: float = 60.0 + + def __post_init__(self) -> None: + roles = [ + f.role for f in (self.engine, *self.artifacts, *self.supporting) + ] + if len(set(roles)) != len(roles): + raise ValueError("bound roles must be unique") + if {a.role for a in self.artifacts} != { + ARTIFACT_AIME_PIA, + ARTIFACT_AIME_ONLY, + }: + raise ValueError("aime_pia and aime_only artifacts are required") + if type(self.candidates_accepted) is not bool: + raise TypeError("candidates_accepted must be an explicit bool") + _text(self.engine_source_commit, "engine source commit") + + @property + def files(self) -> tuple[BoundFile, ...]: + return (self.engine, *self.artifacts, *self.supporting) + + def artifact(self, role: str) -> BoundFile: + for artifact in self.artifacts: + if artifact.role == role: + return artifact + raise KeyError(role) + + def missing_files(self) -> tuple[Path, ...]: + return tuple(f.path for f in self.files if not f.path.is_file()) + + def verify(self) -> None: + """Fail closed on any absent or changed file, or manifest drift.""" + problems = [] + for bound in self.files: + if bound.path.is_symlink() or not bound.path.is_file(): + problems.append(f"{bound.role} is absent: {bound.path}") + elif _sha256_file(bound.path) != bound.sha256: + problems.append(f"{bound.role} digest changed: {bound.path}") + if not problems and self.engine_manifest_role is not None: + manifest = json.loads( + next( + f + for f in self.supporting + if f.role == self.engine_manifest_role + ).path.read_text() + ) + receipt = next( + ( + f.sha256 + for f in self.supporting + if f.role == "engine_provenance_receipt" + ), + None, + ) + if ( + manifest.get("binary_sha256") != self.engine.sha256 + or manifest.get("source_commit") != self.engine_source_commit + or manifest.get("official_no_build_verified") is not True + or (receipt and manifest.get("receipt_sha256") != receipt) + ): + problems.append("engine binding manifest does not bind engine") + if problems: + raise BindingMismatch("; ".join(problems)) + + def document(self) -> dict[str, object]: + return { + "engine": self.engine.document(), + "engine_source_commit": self.engine_source_commit, + "artifacts": [a.document() for a in self.artifacts], + "supporting": [f.document() for f in self.supporting], + "candidates_accepted": self.candidates_accepted, + "candidate_status": self.candidate_status, + } + + +def reviewed_case_a_binding( + engine_root: str | os.PathLike[str] | None = None, + evidence_dir: str | os.PathLike[str] | None = None, +) -> AxiomEngineBinding: + """Bind the exact engine and candidates that reproduced Case A. + + ``evidence_dir`` is the retained ``encoder-recovery`` directory. The + locations default to this machine's layout and may be overridden by + ``POPULACE_DYNAMICS_AXIOM_ENGINE_ROOT`` and + ``POPULACE_DYNAMICS_AXIOM_EVIDENCE_DIR``; the digests may not. + """ + root = Path( + engine_root or os.environ.get(ENGINE_ROOT_ENV) or DEFAULT_ENGINE_ROOT + ).expanduser() + evidence = Path( + evidence_dir + or os.environ.get(EVIDENCE_DIR_ENV) + or DEFAULT_EVIDENCE_DIR + ).expanduser() + case_a = evidence / "ssa-aime-pia-first-comparison-20260922" + integrated = case_a / "integrated-experiment" + rulespec = integrated / "rulespec-us/us" + replay = evidence / "replay-415-b-nested-lifetime-20260922" + binary = root / "target/release/axiom-rules-engine" + + def bound(role: str, path: Path) -> BoundFile: + return BoundFile(role, path, REVIEWED_PINS[role]) + + return AxiomEngineBinding( + engine=bound("engine", binary), + artifacts=( + bound(ARTIFACT_AIME_PIA, integrated / "aime-pia.compiled.json"), + bound(ARTIFACT_AIME_ONLY, replay / "compile/415-b.compiled.json"), + ), + supporting=( + bound( + "engine_provenance_receipt", + binary.with_suffix(".provenance.json"), + ), + bound( + "engine_binding_manifest", case_a / "lazy-engine-binding.json" + ), + bound( + "aime_pia_compile_receipt", integrated / "compile-receipt.json" + ), + bound( + "module:us/statutes/42/415/b.yaml", + rulespec / "statutes/42/415/b.yaml", + ), + bound( + "module:us/statutes/42/415/a.yaml", + rulespec / "statutes/42/415/a.yaml", + ), + bound( + "module:us/policies/ssa/pia-bend-points/2026.yaml", + rulespec / "policies/ssa/pia-bend-points/2026.yaml", + ), + bound( + "aime_only_source:us/statutes/42/415/b.yaml", + replay / "rulespec-us/us/statutes/42/415/b.yaml", + ), + ), + engine_source_commit=REVIEWED_ENGINE_SOURCE_COMMIT, + engine_manifest_role="engine_binding_manifest", + ) + + +def _unique_object(pairs: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, value in pairs: + if key in result: + raise ValueError("duplicate JSON key") + result[key] = value + return result + + +def _reject_constant(value: str) -> None: + raise ValueError(f"nonfinite JSON constant {value}") + + +@dataclass(frozen=True) +class ParsedResponse: + engine_version: str + values: Mapping[str, Decimal] + selected_versions: tuple[Mapping[str, object], ...] + + +def parse_lifetime_response( + request: Mapping[str, object], stdout: bytes +) -> ParsedResponse: + """Validate a one-person response against its request, then read it.""" + try: + response = json.loads( + stdout, + object_pairs_hook=_unique_object, + parse_constant=_reject_constant, + ) + except (ValueError, UnicodeDecodeError) as exc: + raise AxiomExecutionError( + f"response is not valid JSON: {exc}" + ) from exc + fields = { + "schema", + "engine_version", + "artifact_format_version", + "arithmetic", + "entity", + "row_count", + "entity_ids", + "periods", + "calculation_period", + "reference_period", + "output_period", + "selected_versions", + "outputs", + } + if not isinstance(response, dict) or set(response) != fields: + raise AxiomExecutionError("response fields do not match schema v2") + batch = request["batches"][-1] + expected = { + "schema": RESPONSE_SCHEMA, + "artifact_format_version": 2, + "arithmetic": "decimal", + "entity": request["entity"], + "row_count": batch["row_count"], + "entity_ids": batch["entity_ids"], + "periods": request["periods"], + "calculation_period": request["calculation_period"], + "reference_period": request["output_period"], + "output_period": request["output_period"], + } + for name, wanted in expected.items(): + if response[name] != wanted or type(response[name]) is not type( + wanted + ): + raise AxiomExecutionError( + f"response {name} does not match request" + ) + version = response["engine_version"] + if type(version) is not str or not version: + raise AxiomExecutionError("response engine_version is invalid") + outputs = response["outputs"] + requested = list(request["outputs"]) + if not isinstance(outputs, dict) or set(outputs) != set(requested): + raise AxiomExecutionError("response outputs differ from the request") + selected = response["selected_versions"] + if not isinstance(selected, list) or not selected: + raise AxiomExecutionError("response selected_versions are missing") + calculation_start = request["calculation_period"]["start"] + derived = {} + for item in selected: + if not isinstance(item, dict) or item.get("kind") not in ( + "derived", + "unversioned_derived", + "parameter", + ): + raise AxiomExecutionError("response selected version is invalid") + if item["kind"] == "derived": + start, end = item.get("effective_from"), item.get("effective_to") + if ( + type(start) is not str + or start > calculation_start + or ( + end is not None + and (type(end) is not str or end < calculation_start) + ) + ): + raise AxiomExecutionError( + "selected version does not cover the calculation date" + ) + if item["kind"] != "parameter" and item.get("id") is not None: + derived[item["id"]] = item.get("name") + values = {} + for reference in requested: + output = outputs[reference] + if not isinstance(output, dict) or set(output) != { + "id", + "name", + "dtype", + "unit", + "column", + }: + raise AxiomExecutionError(f"output {reference} is malformed") + column = output["column"] + if ( + output["id"] != reference + or derived.get(reference) != output["name"] + or output["dtype"] != "decimal" + or output["unit"] != "USD" + or not isinstance(column, dict) + or set(column) != {"kind", "values"} + or column["kind"] != "decimal" + or not isinstance(column["values"], list) + or len(column["values"]) != 1 + ): + raise AxiomExecutionError(f"output {reference} is malformed") + text = column["values"][0] + if type(text) is not str or not _DECIMAL_TEXT.fullmatch(text): + raise AxiomExecutionError(f"output {reference} is not a decimal") + try: + values[reference] = Decimal(text) + except InvalidOperation as exc: + raise AxiomExecutionError(f"output {reference} invalid") from exc + return ParsedResponse(version, values, tuple(selected)) + + +@dataclass(frozen=True) +class AxiomBenefitResult: + """Engine outputs for one person, with inputs and execution provenance.""" + + person_id: str + aime: Decimal + ordinary_pia_before_cola: Decimal | None + pia_status: str + prepared: PreparedRequest + engine_version: str + stdout_sha256: str + stderr: str + binding: AxiomEngineBinding + selected_versions: tuple[Mapping[str, object], ...] = field(repr=False) + stdout: bytes = field(default=b"", repr=False, compare=False) + + def document(self) -> dict[str, object]: + prepared = self.prepared + artifact = self.binding.artifact(prepared.artifact_role) + return { + "person_id": self.person_id, + "aime": _decimal_text(self.aime), + "ordinary_pia_before_cola": ( + None + if self.ordinary_pia_before_cola is None + else _decimal_text(self.ordinary_pia_before_cola) + ), + "pia_status": self.pia_status, + "calculation_year": prepared.calculation_year, + "determinations": prepared.determinations.document(), + "scoped_assumptions": dict(SCOPED_ASSUMPTIONS), + "earnings_rows": [row.document() for row in prepared.rows], + "synthetic_rows": sum(row.synthetic for row in prepared.rows), + "recorded_gaps": [dict(gap) for gap in prepared.recorded_gaps], + "notes": list(prepared.notes), + "wage_index": prepared.wage_index.document(), + "provenance": { + "engine_path": str(self.binding.engine.path), + "engine_sha256": self.binding.engine.sha256, + "engine_source_commit": self.binding.engine_source_commit, + "engine_version": self.engine_version, + "artifact_role": prepared.artifact_role, + "artifact_path": str(artifact.path), + "artifact_sha256": artifact.sha256, + "module_sha256": { + f.role: f.sha256 + for f in self.binding.supporting + if "module:" in f.role or "source:" in f.role + }, + "candidates_accepted": self.binding.candidates_accepted, + "candidate_status": self.binding.candidate_status, + "outputs_requested": list(prepared.outputs), + "request_sha256": prepared.request_sha256, + "stdout_sha256": self.stdout_sha256, + "stderr": self.stderr, + "selected_versions": [dict(v) for v in self.selected_versions], + "policy_arithmetic_in_bridge": False, + }, + } + + +Runner = Callable[[Sequence[str], bytes, float], tuple[int, bytes, bytes]] + + +def subprocess_runner( + argv: Sequence[str], wire: bytes, timeout: float +) -> tuple[int, bytes, bytes]: + """Run the engine exactly as the Case A runner did.""" + process = subprocess.run( + list(argv), + input=wire, + capture_output=True, + timeout=timeout, + env=dict(_SUBPROCESS_ENV), + check=False, + ) + return process.returncode, process.stdout, process.stderr + + +def execute_request( + prepared: PreparedRequest, + binding: AxiomEngineBinding, + *, + runner: Runner | None = None, +) -> AxiomBenefitResult: + """Verify the binding, run the engine once, verify again, then parse.""" + if type(prepared) is not PreparedRequest: + raise TypeError("PreparedRequest required") + if type(binding) is not AxiomEngineBinding: + raise TypeError("AxiomEngineBinding required") + binding.verify() + artifact = binding.artifact(prepared.artifact_role) + argv = ( + str(binding.engine.path), + "run-lifetime", + "--artifact", + str(artifact.path), + ) + try: + code, stdout, stderr = (runner or subprocess_runner)( + argv, prepared.wire, binding.timeout_seconds + ) + except subprocess.TimeoutExpired as exc: + raise AxiomExecutionError( + f"{prepared.person_id}: engine timed out" + ) from exc + binding.verify() + error_text = stderr.decode("utf-8", errors="replace").strip() + if code != 0: + message = error_text or stdout.decode("utf-8", "replace").strip() + try: + message = json.loads(message)["message"] + except (ValueError, KeyError, TypeError): + pass + raise AxiomExecutionError( + f"{prepared.person_id}: engine exited {code}: {message}" + ) + parsed = parse_lifetime_response(prepared.request, stdout) + pia = parsed.values.get(PIA_OUTPUT) + status = ( + "computed_by_2026_cohort_candidate" + if prepared.pia_requested + else "not_requested: the 415(a) candidate covers only the 2026 " + "first-eligibility cohort; this person attains 62 in " + f"{prepared.determinations.year_attained_age_62}" + ) + return AxiomBenefitResult( + prepared.person_id, + parsed.values[AIME_OUTPUT], + pia, + status, + prepared, + parsed.engine_version, + _sha256_bytes(stdout), + error_text, + binding, + parsed.selected_versions, + stdout, + ) + + +@dataclass(frozen=True) +class PersonDemographics: + """Birth year and sex taken from a retained mortality-step record.""" + + dynamics_person_key: int + birth_year: int + sex: str + derivation: str + + +def cohort_demographics( + history: ClosedCohortEarningsHistory, +) -> dict[int, PersonDemographics]: + """Read ages and sexes that the first retained mortality step recorded. + + The step records ages from the frame for the year before its target + year (the 2014 frame for the 2015 step). The engine's convention is + ``age = year - birth_year``, so each birth year is that year minus age. + """ + if type(history) is not ClosedCohortEarningsHistory: + raise TypeError("ClosedCohortEarningsHistory required") + if not history.transitions: + raise BenefitBridgeRefusal( + "cohort", + [ + "the retained history has no mortality step, so no ages or " + "birth years are recorded" + ], + ) + step = history.transitions[0].mortality + frame_year = step.target_year - 1 + return { + row.dynamics_person_key: PersonDemographics( + row.dynamics_person_key, + frame_year - row.age, + row.sex, + f"{frame_year} minus age {row.age} recorded for the " + f"{frame_year} frame by the mortality step targeting " + f"{step.target_year}", + ) + for row in step.rows + } + + +@dataclass(frozen=True) +class ProjectedAmountDeclaration: + """The caller's explicit statement that projected amounts may be used. + + Forward histories hold labor income whose coverage is not materialized, + in the history's own unit. The candidates need creditable earnings in + USD. Nothing is converted unless the caller declares it here; the + statement is recorded verbatim with every result. + """ + + history_unit: str + treat_as_usd_creditable_earnings: bool + projected_amounts_synthetic: bool + statement: str + binary64_encoding: str = BINARY64_EXACT + + def __post_init__(self) -> None: + _text(self.history_unit, "history unit") + _text(self.statement, "declaration statement") + if self.binary64_encoding not in BINARY64_ENCODINGS: + raise ValueError("unsupported binary64 encoding") + for name in ( + "treat_as_usd_creditable_earnings", + "projected_amounts_synthetic", + ): + if type(getattr(self, name)) is not bool: + raise TypeError(f"{name} must be an explicit bool") + + +def _source_identity(history: ClosedCohortEarningsHistory, key: int) -> str: + identity = history.baseline.identity_map.reverse_rows([key])[0] + return f"{identity.logical_type}:{identity.canonical_value}" + + +def _removal_year( + history: ClosedCohortEarningsHistory, key: int +) -> int | None: + for transition in history.transitions: + step = transition.mortality + if key in step.pre_keys and key not in step.post_keys: + return step.target_year + return None + + +def prepare_cohort_person( + history: ClosedCohortEarningsHistory, + dynamics_person_key: int, + *, + demographics: Mapping[int, PersonDemographics], + entitlement_age: int, + declaration: ProjectedAmountDeclaration | None, + wage_index: AnnualSeries, + contribution_base: AnnualSeries | None = None, + pre_projection: CallerSuppliedHistory | None = None, +) -> PreparedRequest: + """Convert one retained person into a request, or refuse with reasons.""" + if type(history) is not ClosedCohortEarningsHistory: + raise TypeError("ClosedCohortEarningsHistory required") + key = _integer(dynamics_person_key, "dynamics person key") + entitlement_age = _integer(entitlement_age, "entitlement age") + person_id = f"dynamics-person-{key}" + matches = [h for h in history.histories if h.roster_keys == (key,)] + if len(matches) != 1: + raise ValueError(f"{person_id} is outside the original cohort") + person = matches[0] + first_projected = person.observations[0].year + demographic = demographics.get(key) + if demographic is None or type(demographic.birth_year) is not int: + raise BenefitBridgeRefusal( + person_id, ["no birth year is available for this person"] + ) + reasons: list[str] = [] + gaps: dict[str, tuple[int, ...]] = {} + if declaration is None: + reasons.append( + "projected amounts are forward labor income (coverage " + f"{person.coverage_status}, unit {person.unit}); the candidate " + "needs creditable earnings and no ProjectedAmountDeclaration " + "was supplied" + ) + else: + if type(declaration) is not ProjectedAmountDeclaration: + raise TypeError("ProjectedAmountDeclaration required") + if declaration.history_unit != person.unit: + reasons.append( + f"declaration covers unit {declaration.history_unit}, but " + f"the history unit is {person.unit}" + ) + if not declaration.treat_as_usd_creditable_earnings: + reasons.append( + "the declaration does not permit projected amounts to be " + "treated as USD creditable earnings" + ) + entitlement = demographic.birth_year + entitlement_age + try: + determinations = OrdinaryRetirementDeterminations.from_birth_year( + demographic.birth_year, entitlement + ) + except BenefitBridgeRefusal as refusal: + raise BenefitBridgeRefusal( + person_id, [*reasons, *refusal.reasons], gaps=gaps + ) from refusal + removal = _removal_year(history, key) + if removal is not None and removal <= entitlement: + reasons.append( + f"removed by the simulated mortality step targeting {removal}, " + f"not after the {entitlement} entitlement year; only living " + "ordinary old-age entitlement is supported" + ) + projected = [o for o in person.observations if o.year < entitlement] + if not projected: + reasons.append( + f"no projected year ({first_projected}-{person.last_year}) " + f"precedes the {entitlement} entitlement year, so the " + "projection contributes no computation base year" + ) + unavailable = [o.year for o in projected if o.amount_hex is None] + if unavailable: + reasons.append( + f"projected amounts for {year_ranges(unavailable)} are " + "unavailable (outside the forward-earnings domain)" + ) + gaps["unavailable_projected_years"] = tuple(unavailable) + supplement_rows: tuple[EarningsRow, ...] = () + if pre_projection is not None: + if type(pre_projection) is not CallerSuppliedHistory: + raise TypeError("CallerSuppliedHistory required") + late = [y for y, _ in pre_projection.amounts if y >= first_projected] + if late: + raise ValueError( + f"pre-projection years must precede {first_projected}: " + f"{year_ranges(late)}" + ) + supplement_rows = pre_projection.rows() + window = range(_window_start(determinations), entitlement) + supplied = {row.year for row in supplement_rows} | { + o.year for o in projected + } + before = [y for y in window if y < first_projected and y not in supplied] + after = [y for y in window if y > person.last_year] + if before: + reasons.append( + f"pre-projection years {year_ranges(before)} are required (the " + f"elapsed-year window starts {window.start}) but absent: the " + f"retained history begins in {first_projected} and no " + "caller-supplied history covers them" + ) + gaps["missing_pre_projection_years"] = tuple(before) + if after: + reasons.append( + f"years {year_ranges(after)} are required before the " + f"{entitlement} entitlement year but absent: the retained " + f"history ends in {person.last_year}" + ) + gaps["missing_post_projection_years"] = tuple(after) + missing_index = _missing_wage_index_years( + window, determinations.indexing_year, wage_index + ) + if missing_index: + reasons.append( + "national average wage index values are absent for " + f"{year_ranges(missing_index)} (indexing year " + f"{determinations.indexing_year})" + ) + gaps["missing_wage_index_years"] = missing_index + if reasons: + raise BenefitBridgeRefusal(person_id, reasons, gaps=gaps) + source = f"projected:{person.realization_id}" + rows = supplement_rows + tuple( + EarningsRow( + o.year, + binary64_decimal(o.amount_hex, declaration.binary64_encoding), + source, + declaration.projected_amounts_synthetic, + o.amount_hex, + declaration.binary64_encoding, + ) + for o in projected + ) + notes = [ + f"source identity {_source_identity(history, key)}; sex " + f"{demographic.sex}; birth year {demographic.derivation}", + f"projected amounts transported as USD creditable earnings under " + f"declaration: {declaration.statement}", + f"projected binary64 amounts encoded as " + f"{declaration.binary64_encoding}", + ] + excluded = [o.year for o in person.observations if o.year >= entitlement] + if excluded: + notes.append( + f"projected years {year_ranges(excluded)} are not computation " + f"base years before the {entitlement} entitlement and were not " + "transported" + ) + if removal is not None: + notes.append( + f"removed by the simulated mortality step targeting {removal}, " + "after entitlement" + ) + return prepare_request( + person_id, + rows, + determinations, + wage_index, + contribution_base=contribution_base, + notes=notes, + ) + + +@dataclass(frozen=True) +class PersonOutcome: + """Executed result, explicit refusal or engine failure for one person.""" + + dynamics_person_key: int + source_identity: str + birth_year: int | None + sex: str | None + status: str + result: AxiomBenefitResult | None = None + reasons: tuple[str, ...] = () + gaps: Mapping[str, tuple[int, ...]] = field(default_factory=dict) + + def document(self) -> dict[str, object]: + return { + "dynamics_person_key": self.dynamics_person_key, + "source_identity": self.source_identity, + "birth_year": self.birth_year, + "sex": self.sex, + "status": self.status, + "reasons": list(self.reasons), + "gaps": { + name: year_ranges(years) for name, years in self.gaps.items() + }, + "result": None if self.result is None else self.result.document(), + } + + +def run_cohort( + history: ClosedCohortEarningsHistory, + *, + entitlement_age: int, + declaration: ProjectedAmountDeclaration | None, + wage_index: AnnualSeries, + binding: AxiomEngineBinding, + contribution_base: AnnualSeries | None = None, + pre_projection: Mapping[int, CallerSuppliedHistory] | None = None, + runner: Runner | None = None, +) -> tuple[PersonOutcome, ...]: + """Run every original cohort member; refusals never stop the others. + + A changed or absent bound file raises :class:`BindingMismatch` and stops + the whole run. + """ + demographics = cohort_demographics(history) + outcomes = [] + for person in history.histories: + key = person.roster_keys[0] + demographic = demographics.get(key) + common = { + "dynamics_person_key": key, + "source_identity": _source_identity(history, key), + "birth_year": ( + None if demographic is None else demographic.birth_year + ), + "sex": None if demographic is None else demographic.sex, + } + try: + prepared = prepare_cohort_person( + history, + key, + demographics=demographics, + entitlement_age=entitlement_age, + declaration=declaration, + wage_index=wage_index, + contribution_base=contribution_base, + pre_projection=(pre_projection or {}).get(key), + ) + result = execute_request(prepared, binding, runner=runner) + except BenefitBridgeRefusal as refusal: + outcomes.append( + PersonOutcome( + **common, + status="refused", + reasons=refusal.reasons, + gaps=refusal.gaps, + ) + ) + except AxiomExecutionError as error: + outcomes.append( + PersonOutcome(**common, status="failed", reasons=(str(error),)) + ) + else: + outcomes.append( + PersonOutcome(**common, status="executed", result=result) + ) + return tuple(outcomes) diff --git a/tests/README-tiers.md b/tests/README-tiers.md index b5747465..98bd5100 100644 --- a/tests/README-tiers.md +++ b/tests/README-tiers.md @@ -38,9 +38,9 @@ pytest --collect-only -q -m oracle_policyengine | tail -1 | Tier | Tests at HEAD | |---|---:| -| `unit` | 2,207 | +| `unit` | 2,246 | | `artifact` | 2,672 | | `integration_psid` | 848 | | `reproduction_legacy` | 520 | | `oracle_policyengine` | 159 | -| **Total** | **6,406** | +| **Total** | **6,445** | diff --git a/tests/estimates/test_birth_evidence_artifact.py b/tests/estimates/test_birth_evidence_artifact.py index 7b2880f0..6a675c37 100644 --- a/tests/estimates/test_birth_evidence_artifact.py +++ b/tests/estimates/test_birth_evidence_artifact.py @@ -108,6 +108,7 @@ def test_post_review_sources_are_outside_historical_reducer_identity(): Path("src/populace_dynamics/closed_cohort_history.py"), Path("src/populace_dynamics/assembled_history_observer.py"), Path("src/populace_dynamics/compact_cohort_history.py"), + Path("src/populace_dynamics/axiom_benefit_bridge.py"), ) assert reducer.POST_REVIEW_SHARED_SOURCE_BLOBS == { Path( @@ -221,46 +222,6 @@ def test_source_reachability_includes_implicit_package_initializers( assert "sample.untracked_extension" not in reachable -@pytest.mark.parametrize( - "statement", - [ - "import sample.leaf", - "from sample.leaf import function", - "from sample import leaf", - "import sample.untracked_extension", - ], -) -def test_source_reachability_includes_implicit_package_initializers( - tmp_path, statement -): - package = tmp_path / "__init__.py" - leaf = tmp_path / "leaf.py" - hidden = tmp_path / "hidden.py" - consumer = tmp_path / "consumer.py" - package.write_text("from . import hidden\n") - leaf.write_text("def function(): pass\n") - hidden.write_text("") - consumer.write_text(statement + "\n") - modules = { - "consumer": consumer, - "sample": package, - "sample.leaf": leaf, - "sample.hidden": hidden, - } - reachable = set() - pending = ["consumer"] - while pending: - name = pending.pop() - if name in reachable: - continue - reachable.add(name) - pending.extend( - _internal_imports(name, modules[name], modules) - reachable - ) - assert {"sample", "sample.hidden"}.issubset(reachable) - assert "sample.untracked_extension" not in reachable - - def test_post_review_exclusions_are_unreachable_from_birth_evidence(): module_paths = _repository_module_paths() root_module = "scripts.first_estimates_birth_evidence" @@ -349,6 +310,12 @@ def test_post_review_exclusions_are_unreachable_from_birth_evidence(): "opt-in identity and history modules became reachable from the " f"birth-evidence reducer: {sorted(history_modules & reachable)}" ) + benefit_bridge = "populace_dynamics.axiom_benefit_bridge" + assert benefit_bridge in module_paths + assert benefit_bridge not in reachable, ( + "the opt-in Axiom benefit bridge became reachable from the " + "birth-evidence reducer" + ) assert "populace_dynamics.engine.steps" in reachable diff --git a/tests/test_axiom_benefit_bridge.py b/tests/test_axiom_benefit_bridge.py new file mode 100644 index 00000000..b6ea6f77 --- /dev/null +++ b/tests/test_axiom_benefit_bridge.py @@ -0,0 +1,952 @@ +"""Axiom benefit bridge: invented inputs with a fake engine, plus real Case A. + +Every earnings amount, wage index and cohort member in the unit tests is +INVENTED for mechanics only. The integration tests run the actual +axiom-rules-engine binary on SSA's published 2026 Case A and Case B inputs +and skip when that engine or its retained evidence is absent. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import stat +import sys +from decimal import Decimal +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest + +import populace_dynamics.axiom_benefit_bridge as bridge +from populace_dynamics.closed_cohort_history import ClosedCohortEarningsHistory +from populace_dynamics.engine.earnings_domain import EarningsDomainAdapter +from populace_dynamics.engine.steps import AgeSexMortalityModel +from populace_dynamics.forward_earnings_history import ForwardEarningsHistory +from populace_dynamics.person_identity import PersonIdentity, PersonIdentityMap +from tests.test_closed_cohort_history import step +from tests.test_m6_engine_forward_earnings import _generator + +# INVENTED series: plain increasing values, not SSA's published index. +INVENTED_AWI = bridge.AnnualSeries.from_mapping( + "INVENTED wage index for tests", + {year: f"{10000 + 500 * (year - 1951)}.25" for year in range(1951, 2025)}, +) +INVENTED_BASE = bridge.AnnualSeries.from_mapping( + "INVENTED contribution base for tests", + {year: "100000" for year in range(1951, 2027)}, +) +INVENTED_DECLARATION = bridge.ProjectedAmountDeclaration( + "XTS", + True, + True, + "INVENTED test amounts treated as USD creditable earnings", + bridge.BINARY64_SHORTEST_ROUND_TRIP, +) + + +def _history(start, stop, *, amount="1000", synthetic=True): + return bridge.CallerSuppliedHistory.from_mapping( + "INVENTED test history", + {year: amount for year in range(start, stop)}, + synthetic=synthetic, + ) + + +def _case_a_like(): + """INVENTED 1964 cohort shaped like SSA Case A (1986-2025 rows).""" + determinations = bridge.OrdinaryRetirementDeterminations.from_birth_year( + 1964, 2026 + ) + return _history(1986, 2026).rows(), determinations + + +def _fake_response(request, values, *, engine_version="fake-engine-0"): + outputs, selected = {}, [] + for reference in request["outputs"]: + name = reference.split("#", 1)[1] + outputs[reference] = { + "id": reference, + "name": name, + "dtype": "decimal", + "unit": "USD", + "column": {"kind": "decimal", "values": [values[reference]]}, + } + selected.append( + { + "kind": "derived", + "name": name, + "id": reference, + "version_index": 0, + "effective_from": "1979-01-01", + "effective_to": None, + } + ) + return { + "schema": bridge.RESPONSE_SCHEMA, + "engine_version": engine_version, + "artifact_format_version": 2, + "arithmetic": "decimal", + "entity": "Person", + "row_count": 1, + "entity_ids": request["batches"][-1]["entity_ids"], + "periods": request["periods"], + "calculation_period": request["calculation_period"], + "reference_period": request["output_period"], + "output_period": request["output_period"], + "selected_versions": selected, + "outputs": outputs, + } + + +class RecordingRunner: + """Fake engine: records each call and answers from the request.""" + + def __init__(self, values=None, *, mutate=None, code=0, stderr=b""): + self.calls = [] + self.values = values or { + bridge.AIME_OUTPUT: "1234", + bridge.PIA_OUTPUT: "567.8", + } + self.mutate = mutate + self.code = code + self.stderr = stderr + + def __call__(self, argv, wire, timeout): + self.calls.append((tuple(argv), wire, timeout)) + request = json.loads(wire) + response = _fake_response(request, self.values) + if self.mutate is not None: + response = self.mutate(response) + body = ( + response + if isinstance(response, bytes) + else (json.dumps(response).encode()) + ) + return self.code, b"" if self.code else body, self.stderr + + +def _sha(path): + return hashlib.sha256(Path(path).read_bytes()).hexdigest() + + +@pytest.fixture +def fake_binding(tmp_path): + files = {} + for role, text in ( + ("engine", "#!/bin/sh\nexit 3\n"), + (bridge.ARTIFACT_AIME_PIA, '{"invented": "aime_pia"}'), + (bridge.ARTIFACT_AIME_ONLY, '{"invented": "aime_only"}'), + ("module:invented.yaml", "format: invented\n"), + ): + path = tmp_path / role.replace(":", "_") + path.write_text(text) + files[role] = bridge.BoundFile(role, path, _sha(path)) + return bridge.AxiomEngineBinding( + engine=files["engine"], + artifacts=( + files[bridge.ARTIFACT_AIME_PIA], + files[bridge.ARTIFACT_AIME_ONLY], + ), + supporting=(files["module:invented.yaml"],), + engine_source_commit="INVENTED-commit", + ) + + +def test_request_reproduces_the_case_a_wire_format(): + rows, determinations = _case_a_like() + prepared = bridge.prepare_request( + "invented-a", rows, determinations, INVENTED_AWI + ) + request = prepared.request + assert prepared.artifact_role == bridge.ARTIFACT_AIME_PIA + assert prepared.outputs == (bridge.AIME_OUTPUT, bridge.PIA_OUTPUT) + assert request["schema"] == "axiom-rules-engine/lifetime-request/v2" + assert request["outputs"] == [bridge.AIME_OUTPUT, bridge.PIA_OUTPUT] + assert request["calculation_period"] == { + "period_kind": "custom", + "name": "calendar_year", + "start": "2026-01-01", + "end": "2026-12-31", + } + assert request["output_period"] == request["calculation_period"] + assert [p["start"][:4] for p in request["periods"]] == [ + str(year) for year in range(1986, 2026) + ] + assert ( + prepared.wire + == ( + json.dumps(request, sort_keys=True, separators=(",", ":")) + "\n" + ).encode() + ) + per_year = ( + bridge.AIME_INPUT_PREFIX + + "national_average_wage_index_for_computation_base_year" + ) + for year, batch in zip(range(1986, 2026), request["batches"], strict=True): + inputs = batch["inputs"] + assert batch["row_count"] == 1 + assert batch["entity_ids"] == ["invented-a"] + assert inputs[bridge.PIA_COHORT_INPUT] == { + "kind": "bool", + "values": [True], + } + prefix = bridge.AIME_INPUT_PREFIX + assert inputs[prefix + "computation_base_year"] == { + "kind": "integer", + "values": [year], + } + assert inputs[ + prefix + "creditable_earnings_for_computation_base_year" + ]["values"] == ["1000"] + assert inputs[prefix + "indexing_year_for_aime"]["values"] == [2024] + assert inputs[prefix + "year_individual_attained_age_21"][ + "values" + ] == [1985] + assert inputs[prefix + "year_individual_attained_age_62"][ + "values" + ] == [2026] + assert inputs[prefix + "year_of_death_or_later_sentinel"][ + "values" + ] == [2026] + assert inputs[ + prefix + "national_average_wage_index_for_indexing_year" + ]["values"] == [INVENTED_AWI.get(2024)] + # Case A omits the per-year index only after the indexing year. + assert (per_year in inputs) is (year <= 2024) + for name, value in bridge.SCOPED_ASSUMPTIONS.items(): + assert inputs[prefix + name] == {"kind": "bool", "values": [value]} + assert len(inputs) == (16 if year <= 2024 else 15) + kinds = [gap["kind"] for gap in prepared.recorded_gaps] + assert kinds == [ + "creditable_ceiling_not_checked", + "computation_base_years_before_elapsed_window_not_supplied", + ] + assert prepared.recorded_gaps[1]["years"] == "1964-1985" + + +def test_pia_is_requested_only_for_the_candidate_2026_cohort(): + determinations = bridge.OrdinaryRetirementDeterminations.from_birth_year( + 1959, 2026 + ) + prepared = bridge.prepare_request( + "invented-b", + _history(1981, 2026).rows(), + determinations, + INVENTED_AWI, + ) + assert prepared.outputs == (bridge.AIME_OUTPUT,) + assert prepared.artifact_role == bridge.ARTIFACT_AIME_ONLY + assert all( + bridge.PIA_COHORT_INPUT not in batch["inputs"] + for batch in prepared.request["batches"] + ) + + +def test_ordinary_determinations_follow_the_case_a_b_conventions(): + determinations = bridge.OrdinaryRetirementDeterminations.from_birth_year( + 1959, 2026 + ) + assert ( + determinations.year_attained_age_21, + determinations.year_attained_age_62, + determinations.indexing_year, + determinations.death_or_later_sentinel_year, + ) == (1980, 2021, 2019, 2026) + assert any("January 1" in text for text in determinations.derivations) + with pytest.raises(bridge.BenefitBridgeRefusal, match="precedes"): + bridge.OrdinaryRetirementDeterminations.from_birth_year(1964, 2025) + with pytest.raises(TypeError): + bridge.OrdinaryRetirementDeterminations.from_birth_year(1964.0, 2026) + + +def test_too_few_years_refuse_and_list_the_gap(): + _, determinations = _case_a_like() + with pytest.raises(bridge.BenefitBridgeRefusal) as caught: + bridge.prepare_request( + "invented-short", + _history(2000, 2026).rows(), + determinations, + INVENTED_AWI, + ) + assert caught.value.gaps["missing_required_years"] == tuple( + range(1986, 2000) + ) + assert "1986-1999" in str(caught.value) + + +def test_interior_and_trailing_gaps_are_never_padded(): + rows, determinations = _case_a_like() + holed = [row for row in rows if row.year not in (2000, 2025)] + with pytest.raises(bridge.BenefitBridgeRefusal) as caught: + bridge.prepare_request( + "invented-holes", holed, determinations, INVENTED_AWI + ) + assert caught.value.gaps["missing_required_years"] == (2000, 2025) + + +def test_rows_at_or_after_entitlement_are_not_computation_base_years(): + rows, determinations = _case_a_like() + extra = rows + _history(2026, 2027).rows() + with pytest.raises(bridge.BenefitBridgeRefusal, match="2026"): + bridge.prepare_request( + "invented-late", extra, determinations, INVENTED_AWI + ) + with pytest.raises(ValueError, match="once"): + bridge.prepare_request( + "invented-dup", rows + rows[:1], determinations, INVENTED_AWI + ) + + +def test_missing_wage_index_years_refuse(): + rows, determinations = _case_a_like() + thinned = bridge.AnnualSeries( + "INVENTED thinned index", + tuple(item for item in INVENTED_AWI.values if item[0] != 1990), + ) + with pytest.raises(bridge.BenefitBridgeRefusal) as caught: + bridge.prepare_request("invented-awi", rows, determinations, thinned) + assert caught.value.gaps["missing_wage_index_years"] == (1990,) + + +def test_leading_gap_needs_an_exact_explicit_acknowledgement(): + determinations = bridge.OrdinaryRetirementDeterminations.from_birth_year( + 1959, 2026 + ) + rows = _history(1986, 2026).rows() + with pytest.raises(bridge.BenefitBridgeRefusal) as caught: + bridge.prepare_request("invented", rows, determinations, INVENTED_AWI) + assert caught.value.gaps["missing_required_years"] == tuple( + range(1981, 1986) + ) + wrong = bridge.LeadingGapAcknowledgement((1982, 1983), "INVENTED") + with pytest.raises(bridge.BenefitBridgeRefusal, match="do not equal"): + bridge.prepare_request( + "invented", rows, determinations, INVENTED_AWI, leading_gap=wrong + ) + exact = bridge.LeadingGapAcknowledgement( + tuple(range(1981, 1986)), "INVENTED source lists no earlier rows" + ) + prepared = bridge.prepare_request( + "invented", rows, determinations, INVENTED_AWI, leading_gap=exact + ) + recorded = {gap["kind"]: gap for gap in prepared.recorded_gaps} + acknowledged = recorded["acknowledged_unsupplied_elapsed_window_years"] + assert acknowledged["years"] == "1981-1985" + assert acknowledged["statement"] == exact.statement + assert recorded[ + "computation_base_years_before_elapsed_window_not_supplied" + ]["years"] == ("1959-1980") + with pytest.raises(ValueError, match="contiguous"): + bridge.LeadingGapAcknowledgement((1981, 1983), "INVENTED") + + +def test_contribution_base_is_a_check_never_a_cap(): + rows, determinations = _case_a_like() + prepared = bridge.prepare_request( + "invented", + rows, + determinations, + INVENTED_AWI, + contribution_base=INVENTED_BASE, + ) + assert "creditable_ceiling_not_checked" not in { + gap["kind"] for gap in prepared.recorded_gaps + } + high = [ + ( + bridge.EarningsRow(r.year, Decimal("100000.01"), r.source, True) + if r.year == 1990 + else r + ) + for r in rows + ] + with pytest.raises(bridge.BenefitBridgeRefusal) as caught: + bridge.prepare_request( + "invented", + high, + determinations, + INVENTED_AWI, + contribution_base=INVENTED_BASE, + ) + assert caught.value.gaps["above_creditable_ceiling_years"] == (1990,) + partial = bridge.AnnualSeries( + "INVENTED partial base", + tuple(item for item in INVENTED_BASE.values if item[0] >= 2000), + ) + with pytest.raises(bridge.BenefitBridgeRefusal) as caught: + bridge.prepare_request( + "invented", + rows, + determinations, + INVENTED_AWI, + contribution_base=partial, + ) + assert caught.value.gaps["unchecked_creditable_ceiling_years"] == tuple( + range(1986, 2000) + ) + + +def test_amounts_are_exact_and_binary64_encoding_is_explicit(): + with pytest.raises(TypeError): + bridge.EarningsRow(2000, 0.1, "INVENTED", True) + with pytest.raises(ValueError): + bridge.EarningsRow(2000, Decimal("-1"), "INVENTED", True) + bits = (0.1).hex() + exact = bridge.binary64_decimal(bits, bridge.BINARY64_EXACT) + shortest = bridge.binary64_decimal( + bits, bridge.BINARY64_SHORTEST_ROUND_TRIP + ) + assert exact == Decimal(0.1) and shortest == Decimal("0.1") + assert bridge.binary64_decimal( + (21550.0).hex(), bridge.BINARY64_SHORTEST_ROUND_TRIP + ) == Decimal("21550") + row = bridge.EarningsRow( + 2000, + shortest, + "INVENTED", + True, + bits, + bridge.BINARY64_SHORTEST_ROUND_TRIP, + ) + assert row.document()["amount"] == "0.1" + with pytest.raises(ValueError, match="encoded"): + bridge.EarningsRow( + 2000, + exact, + "INVENTED", + True, + bits, + bridge.BINARY64_SHORTEST_ROUND_TRIP, + ) + with pytest.raises(ValueError, match="encoding"): + bridge.EarningsRow( + 2000, shortest, "INVENTED", True, None, bridge.BINARY64_EXACT + ) + with pytest.raises(ValueError, match="unsupported"): + bridge.binary64_decimal(bits, "rounded_to_cents") + + +def test_execute_parses_outputs_and_records_provenance(fake_binding): + rows, determinations = _case_a_like() + prepared = bridge.prepare_request( + "invented-a", rows, determinations, INVENTED_AWI + ) + runner = RecordingRunner() + result = bridge.execute_request(prepared, fake_binding, runner=runner) + ((argv, wire, timeout),) = runner.calls + assert argv == ( + str(fake_binding.engine.path), + "run-lifetime", + "--artifact", + str(fake_binding.artifact(bridge.ARTIFACT_AIME_PIA).path), + ) + assert wire == prepared.wire and timeout == 60.0 + assert result.aime == Decimal("1234") + assert result.ordinary_pia_before_cola == Decimal("567.8") + document = result.document() + provenance = document["provenance"] + assert document["aime"] == "1234" + assert document["ordinary_pia_before_cola"] == "567.8" + assert provenance["engine_sha256"] == fake_binding.engine.sha256 + assert provenance["artifact_sha256"] == ( + fake_binding.artifact(bridge.ARTIFACT_AIME_PIA).sha256 + ) + assert provenance["module_sha256"] == { + "module:invented.yaml": fake_binding.supporting[0].sha256 + } + assert provenance["candidates_accepted"] is False + assert "unaccepted" in provenance["candidate_status"] + assert provenance["request_sha256"] == prepared.request_sha256 + assert provenance["policy_arithmetic_in_bridge"] is False + assert document["synthetic_rows"] == 40 + + +def test_aime_only_people_use_the_aime_only_artifact(fake_binding): + determinations = bridge.OrdinaryRetirementDeterminations.from_birth_year( + 1959, 2026 + ) + prepared = bridge.prepare_request( + "invented-b", _history(1981, 2026).rows(), determinations, INVENTED_AWI + ) + runner = RecordingRunner() + result = bridge.execute_request(prepared, fake_binding, runner=runner) + assert runner.calls[0][0][-1] == str( + fake_binding.artifact(bridge.ARTIFACT_AIME_ONLY).path + ) + assert result.ordinary_pia_before_cola is None + assert result.pia_status.startswith("not_requested") + + +def _without_output(response): + response["outputs"].pop(bridge.PIA_OUTPUT) + return response + + +def _set(path, value): + def mutate(response): + target = response + for key in path[:-1]: + target = target[key] + target[path[-1]] = value + return response + + return mutate + + +@pytest.mark.parametrize( + "mutate", + [ + _set(("schema",), "axiom-rules-engine/lifetime-response/v1"), + _set(("entity_ids",), ["someone-else"]), + _set(("row_count",), True), + _set(("periods",), []), + _set(("artifact_format_version",), 3), + _set(("engine_version",), ""), + _set(("extra",), 1), + _without_output, + _set(("outputs", bridge.AIME_OUTPUT, "column", "values"), [1234]), + _set(("outputs", bridge.AIME_OUTPUT, "column", "values"), ["1e3"]), + _set(("outputs", bridge.AIME_OUTPUT, "unit"), "XTS"), + _set(("selected_versions", 0, "effective_from"), "2030-01-01"), + _set(("selected_versions",), []), + lambda response: json.dumps(response) + .replace('"schema"', '"schema": "x", "schema"', 1) + .encode(), + ], +) +def test_malformed_responses_are_rejected(fake_binding, mutate): + rows, determinations = _case_a_like() + prepared = bridge.prepare_request( + "invented-a", rows, determinations, INVENTED_AWI + ) + with pytest.raises(bridge.AxiomExecutionError): + bridge.execute_request( + prepared, fake_binding, runner=RecordingRunner(mutate=mutate) + ) + + +def test_engine_failures_carry_the_engine_message(fake_binding): + rows, determinations = _case_a_like() + prepared = bridge.prepare_request( + "invented-a", rows, determinations, INVENTED_AWI + ) + stderr = json.dumps( + { + "category": "invalid_request", + "field": None, + "message": "INVENTED engine complaint", + "schema": "axiom-rules-engine/lifetime-error/v1", + } + ).encode() + with pytest.raises(bridge.AxiomExecutionError, match="INVENTED engine"): + bridge.execute_request( + prepared, + fake_binding, + runner=RecordingRunner(code=1, stderr=stderr), + ) + + +def test_changed_bound_files_block_or_invalidate_execution(fake_binding): + rows, determinations = _case_a_like() + prepared = bridge.prepare_request( + "invented-a", rows, determinations, INVENTED_AWI + ) + module = fake_binding.supporting[0].path + original = module.read_text() + module.write_text("changed\n") + runner = RecordingRunner() + with pytest.raises(bridge.BindingMismatch, match="module:invented"): + bridge.execute_request(prepared, fake_binding, runner=runner) + assert runner.calls == [] + module.write_text(original) + + def tampering(argv, wire, timeout): + Path(argv[-1]).write_text('{"changed": true}') + return RecordingRunner()(argv, wire, timeout) + + with pytest.raises(bridge.BindingMismatch, match="aime_pia"): + bridge.execute_request(prepared, fake_binding, runner=tampering) + + +def test_subprocess_path_runs_an_executable_fake_engine(tmp_path): + """The default runner really spawns the bound binary with stdin wire.""" + script = tmp_path / "fake-axiom-rules-engine" + received = tmp_path / "received.json" + script.write_text( + f"#!{sys.executable}\n" + "import json, sys\n" + "sys.path.insert(0, " + repr(str(Path(__file__).parent.parent)) + ")\n" + "from tests.test_axiom_benefit_bridge import _fake_response\n" + "wire = sys.stdin.buffer.read()\n" + f"open({str(received)!r}, 'wb').write(wire)\n" + "assert sys.argv[1:3] == ['run-lifetime', '--artifact']\n" + "values = {'us:statutes/42/415/b#average_indexed_monthly_earnings':" + " '42', 'us:statutes/42/415/a#ordinary_pia_before_cola': '37.8'}\n" + "print(json.dumps(_fake_response(json.loads(wire), values)))\n" + ) + script.chmod(script.stat().st_mode | stat.S_IXUSR) + files = {} + for role in (bridge.ARTIFACT_AIME_PIA, bridge.ARTIFACT_AIME_ONLY): + path = tmp_path / role + path.write_text("{}") + files[role] = bridge.BoundFile(role, path, _sha(path)) + binding = bridge.AxiomEngineBinding( + engine=bridge.BoundFile("engine", script, _sha(script)), + artifacts=tuple(files.values()), + supporting=(), + engine_source_commit="INVENTED-commit", + ) + rows, determinations = _case_a_like() + prepared = bridge.prepare_request( + "invented-a", rows, determinations, INVENTED_AWI + ) + result = bridge.execute_request(prepared, binding) + assert received.read_bytes() == prepared.wire + assert (result.aime, result.ordinary_pia_before_cola) == ( + Decimal("42"), + Decimal("37.8"), + ) + + +def test_reviewed_binding_pins_cannot_be_overridden(tmp_path, monkeypatch): + monkeypatch.setenv(bridge.ENGINE_ROOT_ENV, str(tmp_path / "engine")) + monkeypatch.setenv(bridge.EVIDENCE_DIR_ENV, str(tmp_path / "evidence")) + binding = bridge.reviewed_case_a_binding() + assert binding.engine.path == ( + tmp_path / "engine/target/release/axiom-rules-engine" + ) + assert binding.engine.sha256 == bridge.REVIEWED_PINS["engine"] + assert binding.engine_source_commit == ( + bridge.REVIEWED_ENGINE_SOURCE_COMMIT + ) + assert binding.candidates_accepted is False + assert len(binding.missing_files()) == len(binding.files) + with pytest.raises(bridge.BindingMismatch, match="absent"): + binding.verify() + + +# --------------------------------------------------------------------------- +# Retained cohort histories (INVENTED people produced by actual engine steps) +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def invented_cohort(): + """Four INVENTED people; key 10 is removed by mortality in 2015.""" + mapping = PersonIdentityMap.from_identities( + [ + PersonIdentity("int64", -(2**63)), + PersonIdentity("string", "001"), + *(PersonIdentity("uint64", 2**64 - 1 - i) for i in range(19)), + ] + ) + generator = EarningsDomainAdapter(_generator()) + initial = pd.DataFrame( + { + "person_id": np.array([20, 10, 1, 0], dtype="int64"), + "year": np.full(4, 2014, dtype="int64"), + "age": np.array([62, 36, 61, 30], dtype="int64"), + "sex": ["female", "male", "female", "female"], + } + ) + frame = generator.materialize_initial_frame(initial) + baseline = ForwardEarningsHistory.start( + mapping, + frame, + realization_id="invented-cohort", + generator_digest="a" * 64, + source_contract_digest="b" * 64, + lineage_digest="c" * 64, + unit="XTS", + price_basis="nominal", + ) + model = AgeSexMortalityModel( + ((0, 34), (35, 120)), + { + ("0-34", "female"): 0.0, + ("0-34", "male"): 0.0, + ("35+", "female"): 0.0, + ("35+", "male"): 1.0, + }, + ) + cohort = ClosedCohortEarningsHistory.start(baseline, draw_index=3) + for year in range(2015, 2023): + frame, mortality, _, _ = step(frame, year, generator, baseline, model) + cohort = cohort.append( + mortality=mortality, earnings_frame=frame, lineage_digest="e" * 64 + ) + return cohort + + +def _zero_supplements(demographics): + return { + key: bridge.CallerSuppliedHistory.from_mapping( + "SYNTHETIC explicit zeros", + {year: 0 for year in range(demo.birth_year + 22, 2014)}, + synthetic=True, + ) + for key, demo in demographics.items() + if demo.birth_year + 22 < 2014 + } + + +def test_demographics_come_from_the_retained_2014_ages(invented_cohort): + demographics = bridge.cohort_demographics(invented_cohort) + assert {key: (d.birth_year, d.sex) for key, d in demographics.items()} == { + 0: (1984, "female"), + 1: (1953, "female"), + 10: (1978, "male"), + 20: (1952, "female"), + } + empty = ClosedCohortEarningsHistory.start( + invented_cohort.baseline, draw_index=3 + ) + with pytest.raises(bridge.BenefitBridgeRefusal, match="no ages"): + bridge.cohort_demographics(empty) + + +def _prepare(cohort, key, **overrides): + demographics = bridge.cohort_demographics(cohort) + options = { + "demographics": demographics, + "entitlement_age": 70, + "declaration": INVENTED_DECLARATION, + "wage_index": INVENTED_AWI, + "contribution_base": INVENTED_BASE, + "pre_projection": _zero_supplements(demographics).get(key), + } + options.update(overrides) + return bridge.prepare_cohort_person(cohort, key, **options) + + +def test_projected_labor_income_needs_an_explicit_declaration( + invented_cohort, +): + with pytest.raises(bridge.BenefitBridgeRefusal, match="Declaration"): + _prepare(invented_cohort, 20, declaration=None) + usd = bridge.ProjectedAmountDeclaration("USD", True, True, "INVENTED") + with pytest.raises(bridge.BenefitBridgeRefusal, match="unit is XTS"): + _prepare(invented_cohort, 20, declaration=usd) + refused = bridge.ProjectedAmountDeclaration("XTS", False, True, "INVENTED") + with pytest.raises(bridge.BenefitBridgeRefusal, match="does not permit"): + _prepare(invented_cohort, 20, declaration=refused) + + +def test_missing_birth_year_refuses(invented_cohort): + demographics = bridge.cohort_demographics(invented_cohort) + demographics.pop(20) + with pytest.raises(bridge.BenefitBridgeRefusal, match="birth year"): + _prepare(invented_cohort, 20, demographics=demographics) + + +def test_pre_projection_gap_is_recorded_not_filled(invented_cohort): + with pytest.raises(bridge.BenefitBridgeRefusal) as caught: + _prepare(invented_cohort, 20, pre_projection=None) + assert caught.value.gaps == { + "missing_pre_projection_years": tuple(range(1974, 2014)) + } + with pytest.raises(bridge.BenefitBridgeRefusal) as caught: + _prepare(invented_cohort, 1, entitlement_age=75) + assert caught.value.gaps["missing_post_projection_years"] == tuple( + range(2023, 2028) + ) + with pytest.raises(ValueError, match="precede 2014"): + _prepare(invented_cohort, 20, pre_projection=_history(2010, 2015)) + + +def test_explicit_supplement_carries_projected_bits(invented_cohort): + prepared = _prepare(invented_cohort, 20) + person = next( + h for h in invented_cohort.histories if h.roster_keys == (20,) + ) + assert prepared.person_id == "dynamics-person-20" + assert prepared.calculation_year == 2022 + assert prepared.outputs == (bridge.AIME_OUTPUT,) + projected = [r for r in prepared.rows if r.amount_hex is not None] + synthetic = [r for r in prepared.rows if r.amount_hex is None] + assert [r.year for r in synthetic] == list(range(1974, 2014)) + assert all(r.synthetic and r.amount == 0 for r in synthetic) + assert [r.year for r in projected] == list(range(2014, 2022)) + for row, observation in zip(projected, person.observations, strict=False): + assert row.amount_hex == observation.amount_hex + assert row.amount == Decimal(float.fromhex(observation.amount_hex)) + assert row.encoding == bridge.BINARY64_SHORTEST_ROUND_TRIP + assert row.source == "projected:invented-cohort" + assert any( + "2022 are not computation base years" in n for n in prepared.notes + ) + assert any("uint64" in n or "int64" in n for n in prepared.notes) + + +def test_mortality_removal_and_unavailable_amounts_refuse(invented_cohort): + with pytest.raises(bridge.BenefitBridgeRefusal, match="mortality step"): + _prepare(invented_cohort, 10) + with pytest.raises(bridge.BenefitBridgeRefusal) as caught: + _prepare(invented_cohort, 0) + assert caught.value.gaps["unavailable_projected_years"] == tuple( + range(2014, 2023) + ) + with pytest.raises(bridge.BenefitBridgeRefusal, match="no projected"): + _prepare(invented_cohort, 20, entitlement_age=62) + + +def test_run_cohort_records_every_person(invented_cohort, fake_binding): + demographics = bridge.cohort_demographics(invented_cohort) + runner = RecordingRunner() + outcomes = bridge.run_cohort( + invented_cohort, + entitlement_age=70, + declaration=INVENTED_DECLARATION, + wage_index=INVENTED_AWI, + binding=fake_binding, + contribution_base=INVENTED_BASE, + pre_projection=_zero_supplements(demographics), + runner=runner, + ) + status = {o.dynamics_person_key: o.status for o in outcomes} + assert status == { + 0: "refused", + 1: "executed", + 10: "refused", + 20: "executed", + } + assert len(runner.calls) == 2 + executed = next(o for o in outcomes if o.dynamics_person_key == 1) + document = executed.document() + assert document["source_identity"] == "string:001" + assert document["result"]["aime"] == "1234" + assert document["result"]["ordinary_pia_before_cola"] is None + refused = next(o for o in outcomes if o.dynamics_person_key == 0) + assert refused.document()["gaps"]["unavailable_projected_years"] == ( + "2014-2022" + ) + Path(fake_binding.engine.path).write_text("changed") + with pytest.raises(bridge.BindingMismatch): + bridge.run_cohort( + invented_cohort, + entitlement_age=70, + declaration=INVENTED_DECLARATION, + wage_index=INVENTED_AWI, + binding=fake_binding, + pre_projection=_zero_supplements(demographics), + runner=runner, + ) + + +# --------------------------------------------------------------------------- +# Actual Axiom engine: SSA's published 2026 Case A and Case B +# --------------------------------------------------------------------------- + +CASE_A_TRANSPORT_SHA256 = ( + "7471fb22622078762b5d2064c6b09df7220ebb580a2bbadac8382f9de87b873e" +) +CASE_B_TRANSPORT_SHA256 = ( + "daf051a82bff8f27f0e107fa4ec58bb8c3f7dd7540356216422d9766e818042c" +) +RETAINED_CASE_A_REQUEST_SHA256 = ( + "f5bf1279d6e47d23dd247af0e03007e22d562d78d64463887dfc315902e47ea4" +) +RETAINED_CASE_A_STDOUT_SHA256 = ( + "fafc8d3151ff6a0d2707dfe44c25b912fcd1765a7c90122a547aa96614a57e92" +) +RETAINED_CASE_B_REQUEST_SHA256 = ( + "d738bb71b77b753d4408fafa65fb716d1dc67f321e73011954ed2488dff15811" +) +RETAINED_CASE_B_STDOUT_SHA256 = ( + "ad87a6c2269624c3c57bfc58d4251dfd2578a796bdc88d3295e9907555389bfb" +) + + +def _published_case(name, digest): + evidence = Path( + os.environ.get(bridge.EVIDENCE_DIR_ENV) or bridge.DEFAULT_EVIDENCE_DIR + ).expanduser() + binding = bridge.reviewed_case_a_binding() + transport = evidence / "ssa-aime-pia-first-comparison-20260922" / name + missing = [str(p) for p in binding.missing_files()] + if not transport.is_file(): + missing.append(str(transport)) + if missing: + pytest.skip( + "actual Axiom engine or evidence absent: " + ", ".join(missing) + ) + assert _sha(transport) == digest + source = json.loads(transport.read_text()) + rows = source["annual_observations"] + history = bridge.CallerSuppliedHistory.from_mapping( + source["benchmark"], + {row["year"]: row["nominal_covered_earnings_dollars"] for row in rows}, + synthetic=False, + ) + wage_index = bridge.AnnualSeries.from_mapping( + "SSA AWI transported with " + source["benchmark"], + { + row["year"]: row["awi_decimal"] + for row in rows + if row["awi_decimal"] + }, + source_sha256=digest, + ) + return binding, source, history, wage_index + + +def test_actual_engine_reproduces_published_case_a_through_the_bridge(): + binding, source, history, wage_index = _published_case( + "primary-source-2026-case-a/source-data-transport.json", + CASE_A_TRANSPORT_SHA256, + ) + determinations = bridge.OrdinaryRetirementDeterminations.from_birth_year( + int(source["cohort_facts"]["birth_year"]), + int(source["cohort_facts"]["calculation_year"]), + ) + prepared = bridge.prepare_request( + "ssa_2026_case_a", history.rows(), determinations, wage_index + ) + # Byte-identical to the retained request that first matched Case A. + assert prepared.request_sha256 == RETAINED_CASE_A_REQUEST_SHA256 + result = bridge.execute_request(prepared, binding) + assert result.aime == Decimal("5825") + assert result.ordinary_pia_before_cola == Decimal("2609.80") + assert result.stdout_sha256 == RETAINED_CASE_A_STDOUT_SHA256 + assert hashlib.sha256(result.stdout).hexdigest() == result.stdout_sha256 + assert result.stderr == "" + provenance = result.document()["provenance"] + assert provenance["engine_sha256"] == bridge.REVIEWED_PINS["engine"] + assert provenance["artifact_sha256"] == ( + bridge.REVIEWED_PINS[bridge.ARTIFACT_AIME_PIA] + ) + assert provenance["candidates_accepted"] is False + assert provenance["engine_version"] == "0.2.2" + + +def test_actual_engine_reproduces_published_case_b_aime_only_path(): + binding, source, history, wage_index = _published_case( + "primary-source-2026-case-b/source-data-transport-v2.json", + CASE_B_TRANSPORT_SHA256, + ) + determinations = bridge.OrdinaryRetirementDeterminations.from_birth_year( + int(source["cohort_facts"]["birth_year"]), + int(source["cohort_facts"]["calculation_year"]), + ) + with pytest.raises(bridge.BenefitBridgeRefusal, match="1981-1985"): + bridge.prepare_request( + "ssa_2026_case_b", history.rows(), determinations, wage_index + ) + prepared = bridge.prepare_request( + "ssa_2026_case_b", + history.rows(), + determinations, + wage_index, + leading_gap=bridge.LeadingGapAcknowledgement( + tuple(range(1981, 1986)), + "SSA's published Case B table lists no earnings before 1986", + ), + ) + assert prepared.request_sha256 == RETAINED_CASE_B_REQUEST_SHA256 + result = bridge.execute_request(prepared, binding) + assert result.aime == Decimal("11463") + assert result.ordinary_pia_before_cola is None + assert result.stdout_sha256 == RETAINED_CASE_B_STDOUT_SHA256 diff --git a/tests/tier_counts.json b/tests/tier_counts.json index 42a79468..4b0ec327 100644 --- a/tests/tier_counts.json +++ b/tests/tier_counts.json @@ -1,7 +1,7 @@ { "schema_version": 1, "counts": { - "unit": 2207, + "unit": 2246, "artifact": 2672, "integration_psid": 848, "reproduction_legacy": 520, From 8caf372f1733dd92fed4964134950144db3edcfc Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Tue, 22 Sep 2026 15:05:09 -0400 Subject: [PATCH 2/2] Close benefit-bridge refusal and provenance gaps found in review Independent review of ef53849f found four defects; each now has a regression test. - OrdinaryRetirementDeterminations accepted any indexing year before entitlement. Under the bridge's scoped assumptions (no disability, no death before entitlement) 42 USC 415(b)(3)(A)(ii)(I) with 415(a)(3)(B)(i) fixes it at the year of attaining 62 minus 2. On the real pinned engine, INVENTED flat earnings for a 1953 birth year entitled in 2023 gave AIME 5,806 when indexed to 2013 and 7,835 when indexed to 2021, with no refusal. The constructor now refuses any other indexing year, an age-21 year other than birth year + 21 (or + 20 for a January 1 birth, per the retained POMS RS 00615.015 extract), and attaining 62 before 1979 (415(a)(3)(A)). - prepare_request sent rows dated before the birth year to the engine, where they entered the top-N selection. It now refuses them and lists the years under gaps["years_before_birth"]. - execute_request ran any PreparedRequest, so a hand-built or dataclasses.replace-d request could document rows, determinations or a wage index that were not in the wire next to the wire's digest. It now rebuilds the request from the recorded inputs and raises ValueError on any divergence in wire, request, outputs or artifact role. - AIME-only results listed the 415(a) and 2026 bend-point modules in provenance.module_sha256, although the AIME-only program holds only 415(b). AxiomEngineBinding gains artifact_sources; the reviewed binding attributes 415(a), 415(b) and the bend points to the AIME/PIA program and the 415(b) source alone to the AIME-only program. Bindings without recorded sources keep the previous list-everything behavior. Case A and Case B requests remain byte-identical to the retained requests, and the real-engine tests still reproduce AIME 5,825 / PIA 2,609.80 and AIME 11,463. Re-running the r3 fixture passes with this code gave the same status, reasons, gaps, AIME, request and stdout digests for all 60 person outcomes; only AIME-only module attribution changed. Tier manifest: unit 2,250; total 6,449. Co-Authored-By: Claude Opus 5.5 --- src/populace_dynamics/axiom_benefit_bridge.py | 186 ++++++++++++++-- tests/README-tiers.md | 4 +- tests/test_axiom_benefit_bridge.py | 200 ++++++++++++++++++ tests/tier_counts.json | 2 +- 4 files changed, 372 insertions(+), 20 deletions(-) diff --git a/src/populace_dynamics/axiom_benefit_bridge.py b/src/populace_dynamics/axiom_benefit_bridge.py index 8d63c6a9..339ba2e0 100644 --- a/src/populace_dynamics/axiom_benefit_bridge.py +++ b/src/populace_dynamics/axiom_benefit_bridge.py @@ -43,6 +43,11 @@ # The 415(b) candidate counts computation base years after its 1950 # ``elapsed_years_baseline_year`` parameter. FIRST_COMPUTATION_BASE_YEAR = 1951 +# 42 USC 415(a)(3)(A): the AIME-based paragraph (1) formula applies only to +# an individual not eligible for old-age benefits before January 1979, and +# 415(a)(3)(B)(i) makes old-age eligibility begin in the month of attaining +# 62. The candidates' rule versions are likewise effective from 1979. +FIRST_AIME_ELIGIBILITY_YEAR = 1979 REQUEST_SCHEMA = "axiom-rules-engine/lifetime-request/v2" RESPONSE_SCHEMA = "axiom-rules-engine/lifetime-response/v2" ARTIFACT_AIME_PIA = "aime_pia" @@ -424,14 +429,38 @@ def __post_init__(self) -> None: reasons = [] if self.year_attained_age_62 != self.year_attained_age_21 + 41: reasons.append("ages 21 and 62 must be attained 41 years apart") + if self.year_attained_age_21 - self.birth_year not in (20, 21): + reasons.append( + f"age 21 attained in {self.year_attained_age_21} does not " + f"fit birth year {self.birth_year}: an age is attained on " + "the day before the birthday, so it must be the birth year " + "plus 21 (plus 20 only for a January 1 birth)" + ) + if self.year_attained_age_62 < FIRST_AIME_ELIGIBILITY_YEAR: + reasons.append( + f"attaining 62 in {self.year_attained_age_62} makes this " + "person eligible for old-age benefits before January 1979; " + "42 USC 415(a)(3)(A) limits the AIME-based formula to later " + "eligibility" + ) if self.first_old_age_entitlement_year < self.year_attained_age_62: reasons.append( "ordinary old-age entitlement year " f"{self.first_old_age_entitlement_year} precedes the year " f"of attaining 62 ({self.year_attained_age_62})" ) - if self.indexing_year >= self.first_old_age_entitlement_year: - reasons.append("indexing year must precede entitlement") + if self.indexing_year != self.year_attained_age_62 - 2: + # 415(b)(3)(A)(ii)(I): the second year before the earliest of + # death, old-age eligibility or disability eligibility. The + # scoped assumptions exclude disability and death before + # entitlement, so old-age eligibility (attaining 62) governs. + reasons.append( + f"indexing year {self.indexing_year} is not the second " + "year before old-age eligibility in " + f"{self.year_attained_age_62}; with no disability and no " + "death before entitlement, 42 USC 415(b)(3)(A)(ii)(I) fixes " + f"it at {self.year_attained_age_62 - 2}" + ) if ( self.death_or_later_sentinel_year < self.first_old_age_entitlement_year @@ -574,6 +603,14 @@ def prepare_request( f"years {year_ranges(outside)} are not computation base years " f"before the {entitlement} entitlement year" ) + # The engine would rank such rows in its top-N selection. + before_birth = [y for y in years if y < determinations.birth_year] + if before_birth: + reasons.append( + f"years {year_ranges(before_birth)} precede the " + f"{determinations.birth_year} birth year" + ) + gaps["years_before_birth"] = tuple(before_birth) window = range(_window_start(determinations), entitlement) supplied = set(years) missing = [y for y in window if y not in supplied] @@ -680,6 +717,38 @@ def prepare_request( "top-N selection", } ) + request, outputs, artifact_role = _build_request( + person_id, rows, determinations, wage_index + ) + return PreparedRequest( + person_id, + request, + _wire(request), + artifact_role, + outputs, + rows, + determinations, + tuple(recorded), + wage_index, + tuple(notes), + ) + + +def _wire(request: Mapping[str, object]) -> bytes: + return ( + json.dumps(request, sort_keys=True, separators=(",", ":")) + "\n" + ).encode() + + +def _build_request( + person_id: str, + rows: tuple[EarningsRow, ...], + determinations: OrdinaryRetirementDeterminations, + wage_index: AnnualSeries, +) -> tuple[dict[str, object], tuple[str, ...], str]: + """The Case A wire request for already validated rows (no refusals).""" + entitlement = determinations.first_old_age_entitlement_year + index_year = determinations.indexing_year pia = determinations.year_attained_age_62 == PIA_CANDIDATE_ELIGIBILITY_YEAR outputs = (AIME_OUTPUT, PIA_OUTPUT) if pia else (AIME_OUTPUT,) batches = [] @@ -740,23 +809,51 @@ def prepare_request( "outputs": list(outputs), "output_period": calculation, } - wire = ( - json.dumps(request, sort_keys=True, separators=(",", ":")) + "\n" - ).encode() - return PreparedRequest( - person_id, + return ( request, - wire, - ARTIFACT_AIME_PIA if pia else ARTIFACT_AIME_ONLY, outputs, - rows, - determinations, - tuple(recorded), - wage_index, - tuple(notes), + ARTIFACT_AIME_PIA if pia else ARTIFACT_AIME_ONLY, ) +def _require_consistent(prepared: PreparedRequest) -> dict[str, object]: + """Rebuild the wire from the recorded inputs; refuse any divergence. + + A result documents ``prepared.rows``, determinations and wage index next + to ``request_sha256``. Rebuilding the request from those records makes + the digest bind them, so a hand-built or ``dataclasses.replace``-d + PreparedRequest cannot report rows other than the ones sent. + """ + if ( + type(prepared.determinations) is not OrdinaryRetirementDeterminations + or type(prepared.wage_index) is not AnnualSeries + or any(type(row) is not EarningsRow for row in prepared.rows) + or [row.year for row in prepared.rows] + != sorted({row.year for row in prepared.rows}) + ): + raise ValueError( + f"{prepared.person_id}: prepared request records are malformed" + ) + request, outputs, artifact_role = _build_request( + prepared.person_id, + prepared.rows, + prepared.determinations, + prepared.wage_index, + ) + wire = _wire(request) + if ( + prepared.wire != wire + or _wire(prepared.request) != wire + or tuple(prepared.outputs) != outputs + or prepared.artifact_role != artifact_role + ): + raise ValueError( + f"{prepared.person_id}: prepared request does not match its " + "recorded rows, determinations and wage index" + ) + return request + + @dataclass(frozen=True) class BoundFile: """A file whose SHA-256 must match before and after every execution.""" @@ -791,6 +888,10 @@ class AxiomEngineBinding: candidates_accepted: bool = False engine_manifest_role: str | None = None timeout_seconds: float = 60.0 + # Supporting roles each compiled artifact was built from, per artifact + # role. Empty means unrecorded: results then list every bound + # candidate file rather than attributing modules to one artifact. + artifact_sources: tuple[tuple[str, tuple[str, ...]], ...] = () def __post_init__(self) -> None: roles = [ @@ -806,6 +907,24 @@ def __post_init__(self) -> None: if type(self.candidates_accepted) is not bool: raise TypeError("candidates_accepted must be an explicit bool") _text(self.engine_source_commit, "engine source commit") + sources = tuple( + (role, tuple(names)) for role, names in self.artifact_sources + ) + if sources: + supporting = {f.role for f in self.supporting} + named = [role for role, _ in sources] + if sorted(named) != sorted(a.role for a in self.artifacts): + raise ValueError("artifact sources must name each artifact") + if any( + not names + or len(set(names)) != len(names) + or not set(names) <= supporting + for _, names in sources + ): + raise ValueError( + "artifact sources must be distinct bound supporting roles" + ) + object.__setattr__(self, "artifact_sources", sources) @property def files(self) -> tuple[BoundFile, ...]: @@ -817,6 +936,19 @@ def artifact(self, role: str) -> BoundFile: return artifact raise KeyError(role) + def sources_for(self, role: str) -> tuple[BoundFile, ...]: + """Candidate files an artifact was compiled from, when recorded.""" + by_role = {f.role: f for f in self.supporting} + for artifact_role, names in self.artifact_sources: + if artifact_role == role: + return tuple(by_role[name] for name in names) + self.artifact(role) + return tuple( + f + for f in self.supporting + if "module:" in f.role or "source:" in f.role + ) + def missing_files(self) -> tuple[Path, ...]: return tuple(f.path for f in self.files if not f.path.is_file()) @@ -860,6 +992,9 @@ def document(self) -> dict[str, object]: "engine_source_commit": self.engine_source_commit, "artifacts": [a.document() for a in self.artifacts], "supporting": [f.document() for f in self.supporting], + "artifact_sources": { + role: list(names) for role, names in self.artifact_sources + }, "candidates_accepted": self.candidates_accepted, "candidate_status": self.candidate_status, } @@ -929,6 +1064,23 @@ def bound(role: str, path: Path) -> BoundFile: ), engine_source_commit=REVIEWED_ENGINE_SOURCE_COMMIT, engine_manifest_role="engine_binding_manifest", + # The AIME/PIA program holds the 415(a), 415(b) and 2026 bend-point + # modules (its compile receipt overlays 415(a) and 415(b) onto the + # rule-spec root); the AIME-only program holds 415(b) alone. + artifact_sources=( + ( + ARTIFACT_AIME_PIA, + ( + "module:us/statutes/42/415/b.yaml", + "module:us/statutes/42/415/a.yaml", + "module:us/policies/ssa/pia-bend-points/2026.yaml", + ), + ), + ( + ARTIFACT_AIME_ONLY, + ("aime_only_source:us/statutes/42/415/b.yaml",), + ), + ), ) @@ -1117,8 +1269,7 @@ def document(self) -> dict[str, object]: "artifact_sha256": artifact.sha256, "module_sha256": { f.role: f.sha256 - for f in self.binding.supporting - if "module:" in f.role or "source:" in f.role + for f in self.binding.sources_for(prepared.artifact_role) }, "candidates_accepted": self.binding.candidates_accepted, "candidate_status": self.binding.candidate_status, @@ -1161,6 +1312,7 @@ def execute_request( raise TypeError("PreparedRequest required") if type(binding) is not AxiomEngineBinding: raise TypeError("AxiomEngineBinding required") + request = _require_consistent(prepared) binding.verify() artifact = binding.artifact(prepared.artifact_role) argv = ( @@ -1188,7 +1340,7 @@ def execute_request( raise AxiomExecutionError( f"{prepared.person_id}: engine exited {code}: {message}" ) - parsed = parse_lifetime_response(prepared.request, stdout) + parsed = parse_lifetime_response(request, stdout) pia = parsed.values.get(PIA_OUTPUT) status = ( "computed_by_2026_cohort_candidate" diff --git a/tests/README-tiers.md b/tests/README-tiers.md index 98bd5100..3150bcba 100644 --- a/tests/README-tiers.md +++ b/tests/README-tiers.md @@ -38,9 +38,9 @@ pytest --collect-only -q -m oracle_policyengine | tail -1 | Tier | Tests at HEAD | |---|---:| -| `unit` | 2,246 | +| `unit` | 2,250 | | `artifact` | 2,672 | | `integration_psid` | 848 | | `reproduction_legacy` | 520 | | `oracle_policyengine` | 159 | -| **Total** | **6,445** | +| **Total** | **6,449** | diff --git a/tests/test_axiom_benefit_bridge.py b/tests/test_axiom_benefit_bridge.py index b6ea6f77..e5e3aa2f 100644 --- a/tests/test_axiom_benefit_bridge.py +++ b/tests/test_axiom_benefit_bridge.py @@ -8,6 +8,7 @@ from __future__ import annotations +import dataclasses import hashlib import json import os @@ -262,6 +263,71 @@ def test_ordinary_determinations_follow_the_case_a_b_conventions(): bridge.OrdinaryRetirementDeterminations.from_birth_year(1964.0, 2026) +def test_determinations_refuse_inconsistent_calendar_facts(): + """Regression: the public constructor once accepted any earlier + indexing year. Indexing to 2021 instead of 2013 for a 1953 birth year + entitled at 70 raised the real engine's AIME on INVENTED flat earnings + from 5,806 to 7,835 without a refusal.""" + birth, entitlement = 1953, 2023 + statutory = bridge.OrdinaryRetirementDeterminations.from_birth_year( + birth, entitlement + ) + assert statutory.indexing_year == 2013 + with pytest.raises(bridge.BenefitBridgeRefusal) as caught: + bridge.OrdinaryRetirementDeterminations( + birth, entitlement, 1974, 2015, 2021, entitlement + ) + (reason,) = caught.value.reasons + assert "415(b)(3)(A)(ii)(I)" in reason and "2013" in reason + with pytest.raises(bridge.BenefitBridgeRefusal, match="2013"): + bridge.OrdinaryRetirementDeterminations( + birth, entitlement, 1974, 2015, 2012, entitlement + ) + # Only birth year + 21, or + 20 for a January 1 birth, fits age 21. + january_first = bridge.OrdinaryRetirementDeterminations( + birth, entitlement, 1973, 2014, 2012, entitlement + ) + assert january_first.year_attained_age_21 == birth + 20 + with pytest.raises(bridge.BenefitBridgeRefusal, match="does not fit"): + bridge.OrdinaryRetirementDeterminations( + birth, entitlement, 1976, 2017, 2015, entitlement + ) + # 415(a)(3)(A): no AIME-based formula for eligibility before 1979. + with pytest.raises(bridge.BenefitBridgeRefusal, match="January 1979"): + bridge.OrdinaryRetirementDeterminations.from_birth_year(1916, 1986) + assert ( + bridge.OrdinaryRetirementDeterminations.from_birth_year( + 1917, 1987 + ).year_attained_age_62 + == bridge.FIRST_AIME_ELIGIBILITY_YEAR + ) + + +def test_rows_before_the_birth_year_refuse(): + """Regression: INVENTED rows dated before birth once reached the + engine's top-N selection.""" + determinations = bridge.OrdinaryRetirementDeterminations.from_birth_year( + 1960, 2022 + ) + rows = _history(1955, 2022).rows() + with pytest.raises(bridge.BenefitBridgeRefusal) as caught: + bridge.prepare_request( + "invented-prebirth", rows, determinations, INVENTED_AWI + ) + assert caught.value.gaps == { + "years_before_birth": tuple(range(1955, 1960)) + } + assert "1955-1959 precede the 1960 birth year" in str(caught.value) + # Rows from the birth year on remain acceptable computation base years. + prepared = bridge.prepare_request( + "invented-from-birth", + [row for row in rows if row.year >= 1960], + determinations, + INVENTED_AWI, + ) + assert prepared.rows[0].year == 1960 + + def test_too_few_years_refuse_and_list_the_gap(): _, determinations = _case_a_like() with pytest.raises(bridge.BenefitBridgeRefusal) as caught: @@ -528,6 +594,116 @@ def test_malformed_responses_are_rejected(fake_binding, mutate): ) +def test_prepared_records_must_match_the_wire_they_document(fake_binding): + """Regression: a PreparedRequest whose rows, wire or outputs were + swapped still executed, and its result documented rows other than the + ones sent next to the sent request's digest.""" + rows, determinations = _case_a_like() + prepared = bridge.prepare_request( + "invented-a", rows, determinations, INVENTED_AWI + ) + other = bridge.prepare_request( + "invented-a", + _history(1986, 2026, amount="2000").rows(), + determinations, + INVENTED_AWI, + ) + tampered = [ + dataclasses.replace(prepared, rows=other.rows), + dataclasses.replace(prepared, wire=other.wire), + dataclasses.replace(prepared, request=other.request), + dataclasses.replace(prepared, outputs=(bridge.AIME_OUTPUT,)), + dataclasses.replace(prepared, artifact_role=bridge.ARTIFACT_AIME_ONLY), + dataclasses.replace(prepared, wage_index=INVENTED_BASE), + dataclasses.replace( + prepared, + determinations=( + bridge.OrdinaryRetirementDeterminations.from_birth_year( + 1964, 2027 + ) + ), + ), + dataclasses.replace(prepared, rows=prepared.rows[::-1]), + dataclasses.replace(prepared, rows=(*prepared.rows, None)), + ] + runner = RecordingRunner() + for candidate in tampered: + with pytest.raises(ValueError, match="invented-a: prepared request"): + bridge.execute_request(candidate, fake_binding, runner=runner) + assert runner.calls == [] + result = bridge.execute_request(prepared, fake_binding, runner=runner) + assert runner.calls[0][1] == prepared.wire + assert result.document()["earnings_rows"][0]["amount"] == "1000" + + +def test_results_attribute_modules_to_the_executed_artifact(fake_binding): + """Regression: AIME-only results listed the 415(a) and bend-point + modules, which the AIME-only program does not contain.""" + extra = fake_binding.supporting[0].path.with_name("aime_only_source") + extra.write_text("format: invented aime-only source\n") + supporting = ( + *fake_binding.supporting, + bridge.BoundFile("aime_only_source:invented.yaml", extra, _sha(extra)), + ) + binding = dataclasses.replace( + fake_binding, + supporting=supporting, + artifact_sources=( + (bridge.ARTIFACT_AIME_PIA, ("module:invented.yaml",)), + (bridge.ARTIFACT_AIME_ONLY, ("aime_only_source:invented.yaml",)), + ), + ) + only = bridge.prepare_request( + "invented-b", + _history(1981, 2026).rows(), + bridge.OrdinaryRetirementDeterminations.from_birth_year(1959, 2026), + INVENTED_AWI, + ) + rows, determinations = _case_a_like() + both = bridge.prepare_request( + "invented-a", rows, determinations, INVENTED_AWI + ) + runner = RecordingRunner() + modules = { + prepared.artifact_role: bridge.execute_request( + prepared, binding, runner=runner + ).document()["provenance"]["module_sha256"] + for prepared in (only, both) + } + assert modules == { + bridge.ARTIFACT_AIME_ONLY: { + "aime_only_source:invented.yaml": _sha(extra) + }, + bridge.ARTIFACT_AIME_PIA: { + "module:invented.yaml": fake_binding.supporting[0].sha256 + }, + } + assert binding.document()["artifact_sources"] == { + bridge.ARTIFACT_AIME_PIA: ["module:invented.yaml"], + bridge.ARTIFACT_AIME_ONLY: ["aime_only_source:invented.yaml"], + } + # Unrecorded sources keep listing every bound candidate file. + unrecorded = dataclasses.replace(binding, artifact_sources=()) + assert set( + bridge.execute_request(only, unrecorded, runner=runner).document()[ + "provenance" + ]["module_sha256"] + ) == {"module:invented.yaml", "aime_only_source:invented.yaml"} + for bad in ( + ((bridge.ARTIFACT_AIME_PIA, ("module:invented.yaml",)),), + ( + (bridge.ARTIFACT_AIME_PIA, ("module:invented.yaml",)), + (bridge.ARTIFACT_AIME_ONLY, ("module:absent.yaml",)), + ), + ( + (bridge.ARTIFACT_AIME_PIA, ()), + (bridge.ARTIFACT_AIME_ONLY, ("module:invented.yaml",)), + ), + ): + with pytest.raises(ValueError, match="artifact sources"): + dataclasses.replace(binding, artifact_sources=bad) + + def test_engine_failures_carry_the_engine_message(fake_binding): rows, determinations = _case_a_like() prepared = bridge.prepare_request( @@ -623,6 +799,16 @@ def test_reviewed_binding_pins_cannot_be_overridden(tmp_path, monkeypatch): bridge.REVIEWED_ENGINE_SOURCE_COMMIT ) assert binding.candidates_accepted is False + assert dict(binding.artifact_sources) == { + bridge.ARTIFACT_AIME_PIA: ( + "module:us/statutes/42/415/b.yaml", + "module:us/statutes/42/415/a.yaml", + "module:us/policies/ssa/pia-bend-points/2026.yaml", + ), + bridge.ARTIFACT_AIME_ONLY: ( + "aime_only_source:us/statutes/42/415/b.yaml", + ), + } assert len(binding.missing_files()) == len(binding.files) with pytest.raises(bridge.BindingMismatch, match="absent"): binding.verify() @@ -920,6 +1106,14 @@ def test_actual_engine_reproduces_published_case_a_through_the_bridge(): ) assert provenance["candidates_accepted"] is False assert provenance["engine_version"] == "0.2.2" + assert provenance["module_sha256"] == { + role: bridge.REVIEWED_PINS[role] + for role in ( + "module:us/statutes/42/415/b.yaml", + "module:us/statutes/42/415/a.yaml", + "module:us/policies/ssa/pia-bend-points/2026.yaml", + ) + } def test_actual_engine_reproduces_published_case_b_aime_only_path(): @@ -950,3 +1144,9 @@ def test_actual_engine_reproduces_published_case_b_aime_only_path(): assert result.aime == Decimal("11463") assert result.ordinary_pia_before_cola is None assert result.stdout_sha256 == RETAINED_CASE_B_STDOUT_SHA256 + provenance = result.document()["provenance"] + assert provenance["artifact_sha256"] == ( + bridge.REVIEWED_PINS[bridge.ARTIFACT_AIME_ONLY] + ) + role = "aime_only_source:us/statutes/42/415/b.yaml" + assert provenance["module_sha256"] == {role: bridge.REVIEWED_PINS[role]} diff --git a/tests/tier_counts.json b/tests/tier_counts.json index 4b0ec327..710245e1 100644 --- a/tests/tier_counts.json +++ b/tests/tier_counts.json @@ -1,7 +1,7 @@ { "schema_version": 1, "counts": { - "unit": 2246, + "unit": 2250, "artifact": 2672, "integration_psid": 848, "reproduction_legacy": 520,