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..339ba2e0 --- /dev/null +++ b/src/populace_dynamics/axiom_benefit_bridge.py @@ -0,0 +1,1706 @@ +"""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 +# 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" +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.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.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 + ): + 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" + ) + # 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] + 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", + } + ) + 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 = [] + 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, + } + return ( + request, + outputs, + 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.""" + + 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 + # 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 = [ + 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") + 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, ...]: + 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 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()) + + 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], + "artifact_sources": { + role: list(names) for role, names in self.artifact_sources + }, + "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", + # 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",), + ), + ), + ) + + +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.sources_for(prepared.artifact_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") + request = _require_consistent(prepared) + 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(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 84c88dd1..20562c50 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,298 | +| `unit` | 2,341 | | `artifact` | 2,672 | | `integration_psid` | 848 | | `reproduction_legacy` | 520 | | `oracle_policyengine` | 159 | -| **Total** | **6,497** | +| **Total** | **6,540** | diff --git a/tests/estimates/test_birth_evidence_artifact.py b/tests/estimates/test_birth_evidence_artifact.py index e31d58bf..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( @@ -309,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..e5e3aa2f --- /dev/null +++ b/tests/test_axiom_benefit_bridge.py @@ -0,0 +1,1152 @@ +"""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 dataclasses +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_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: + 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_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( + "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 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() + + +# --------------------------------------------------------------------------- +# 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" + 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(): + 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 + 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 d5a41003..743e3ea2 100644 --- a/tests/tier_counts.json +++ b/tests/tier_counts.json @@ -1,7 +1,7 @@ { "schema_version": 1, "counts": { - "unit": 2298, + "unit": 2341, "artifact": 2672, "integration_psid": 848, "reproduction_legacy": 520,