From 51f115102d9a742fab4bf29666a6d7550dbb1fa9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Fri, 14 Aug 2026 14:08:52 +0200 Subject: [PATCH 1/4] UK source stages on the shared manifest: source_stages.json + StagePlan assembly Copy the frozen HMRC manifest into canonical uk/source_stages.json, split at operation granularity into the two runtime stage names, with incumbent-name citations rephrased; the frozen original stays byte-identical behind a sha-pinned lockstep test. Add the ten UK operation kinds plus the generic materialize_rules_engine_predictors to the shared allowlist. The UK national driver assembles via country_stage_plan(load_country_spec("uk"), ...) with UKNationalStage kept as a one-release deprecated shim; StageRecords are surfaced, including through checkpoint resume, and the UK frame validation still runs after every stage. Shared rules-engine seam in frame/rules.py (country assertion + predictor materializer), a thin deferred-import PolicyEngine-UK adapter, and the microcosm-frame [uk] extra. Workstream E1 of the UK migration (#145, #665). Co-Authored-By: Claude Fable 5 --- .../677-uk-source-stages-manifest.added.md | 1 + .../src/microcosm/build/source_manifest.py | 11 + .../microcosm/build/uk/country_package.json | 1 + .../src/microcosm/build/uk/source_stages.json | 744 ++++++++++++++++++ .../build/uk_runtime/national_build.py | 117 ++- .../build/uk_runtime/source_runtime.py | 113 +++ .../tests/test_country_spec.py | 1 + .../tests/test_uk_national_build.py | 112 +++ .../tests/test_uk_national_build_driver.py | 21 +- .../tests/test_uk_source_runtime.py | 139 ++++ .../tests/test_uk_source_stages.py | 147 ++++ packages/microcosm-frame/pyproject.toml | 1 + .../frame/adapters/policyengine_uk.py | 160 ++++ .../frame/adapters/policyengine_us.py | 2 + .../src/microcosm/frame/rules.py | 89 ++- .../tests/test_policyengine_uk_adapter.py | 28 + tools/build_uk_national_dataset.py | 26 +- 17 files changed, 1673 insertions(+), 40 deletions(-) create mode 100644 changelog.d/677-uk-source-stages-manifest.added.md create mode 100644 packages/microcosm-build/src/microcosm/build/uk/source_stages.json create mode 100644 packages/microcosm-build/src/microcosm/build/uk_runtime/source_runtime.py create mode 100644 packages/microcosm-build/tests/test_uk_source_runtime.py create mode 100644 packages/microcosm-build/tests/test_uk_source_stages.py create mode 100644 packages/microcosm-frame/src/microcosm/frame/adapters/policyengine_uk.py create mode 100644 packages/microcosm-frame/tests/test_policyengine_uk_adapter.py diff --git a/changelog.d/677-uk-source-stages-manifest.added.md b/changelog.d/677-uk-source-stages-manifest.added.md new file mode 100644 index 00000000..1d57d5d2 --- /dev/null +++ b/changelog.d/677-uk-source-stages-manifest.added.md @@ -0,0 +1 @@ +Declare the UK canonical `source_stages.json`, assemble UK national stages through shared `country_stage_plan`/`StagePlan`, and add `materialize_rules_engine_predictors` plus a PolicyEngine-UK rules-engine adapter. diff --git a/packages/microcosm-build/src/microcosm/build/source_manifest.py b/packages/microcosm-build/src/microcosm/build/source_manifest.py index ef646178..54216698 100644 --- a/packages/microcosm-build/src/microcosm/build/source_manifest.py +++ b/packages/microcosm-build/src/microcosm/build/source_manifest.py @@ -47,6 +47,7 @@ "attribute_self_employed_health_premiums", "calibrate_binary_assignment", "calibrate_binary_assignment_joint_targets", + "classify_hmrc_income_facts_with_reviewed_fences", "convert_interest_to_structural_mortgage_inputs", "compute_ratio", "declare_income_reference_offset", @@ -82,7 +83,10 @@ "fit_vehicle_model", "fit_weighted_imputer", "fit_weighted_qrf", + "fit_weighted_qrf_stage1", + "fit_weighted_qrf_stage2", "fold_into", + "gate_distributional_effective_mass", "head_carry", "join", "impute_retirement_contributions_to_puf_support", @@ -97,13 +101,20 @@ "impute_workers_compensation_to_puf_support", "impute_weeks_unemployed_to_puf_support", "map_columns", + "materialize_hmrc_income_bands_fail_closed", + "materialize_rules_engine_predictors", "read_table", "read_tables", "read_acs_rent_donor", + "replace_zero_weight_spi_support", + "retain_adjudicated_frs_hmrc_leaves", "replace_sentinels", "split_component_by_share", + "strict_read_private_table", "support_clip", "uprate", + "verify_certified_candidate", + "verify_pinned_hmrc_source_pair", "zero_when_false", } ) diff --git a/packages/microcosm-build/src/microcosm/build/uk/country_package.json b/packages/microcosm-build/src/microcosm/build/uk/country_package.json index d339827e..8eccaad0 100644 --- a/packages/microcosm-build/src/microcosm/build/uk/country_package.json +++ b/packages/microcosm-build/src/microcosm/build/uk/country_package.json @@ -11,6 +11,7 @@ "hmrc_income_release_gate_report.json", "hmrc_income_replay_report.json", "hmrc_income_source_stages.json", + "source_stages.json", "input_mass_reviewed_exclusions.json", "national_staging_build_record.json", "qrf_tail_reviewed_exclusions.json", diff --git a/packages/microcosm-build/src/microcosm/build/uk/source_stages.json b/packages/microcosm-build/src/microcosm/build/uk/source_stages.json new file mode 100644 index 00000000..2a8e2654 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/uk/source_stages.json @@ -0,0 +1,744 @@ +{ + "version": 1, + "country": "uk", + "policy": "The UK HMRC/SPI income family is source-manifest-defined. Private donor data must be supplied locally, every artifact must be SHA-256 verified at runtime, retained FRS constituents and published bands fail closed, and the current replay keeps importance-kind weights because all 208 banded facts require an unavailable full FRS total-income measure.", + "stages": [ + { + "stage": "frs_hmrc_retained_leaves", + "survey": "Family Resources Survey 2023-24", + "source": "Department for Work and Pensions Family Resources Survey 2023-24 raw adult.tab and benefits.tab, caller-supplied local input", + "grain": "person", + "artifacts": [], + "operations": [ + { + "kind": "verify_certified_candidate", + "artifact": "base_candidate", + "runtime_sha256_required": true, + "fail_on_mismatch": true + }, + { + "kind": "retain_adjudicated_frs_hmrc_leaves", + "population": "certified_microcosm_uk_candidate_base_channel", + "source_vintage": "2023-24", + "mapped_build_period": 2023, + "annualization": "weekly raw FRS amounts * (365.25 / 7)", + "status": "adjudicated_partial_replay", + "retained_full_constituents": { + "hmrc_spi_pay": { + "spi_concept": "PAY", + "scope": "full", + "raw_sources": [ + "ADULT.INEARNS" + ], + "formula": "max(0, ADULT.INEARNS) * (365.25 / 7)" + }, + "hmrc_spi_unemployment_benefit_income": { + "spi_concept": "UBISJA", + "scope": "full", + "raw_sources": [ + "BENEFITS.BENEFIT=14:BENAMT", + "BENEFITS.BENEFIT=19:BENAMT" + ], + "formula": "sum(BENAMT where BENEFIT in {14, 19}) * (365.25 / 7)" + }, + "hmrc_spi_incapacity_benefit_income": { + "spi_concept": "INCPBEN", + "scope": "full", + "raw_sources": [ + "BENEFITS.BENEFIT=17:BENAMT" + ], + "formula": "sum(BENAMT where BENEFIT == 17) * (365.25 / 7)", + "observed_support": "structural zero in the audited 2023-24 FRS; retained so future vintages flow" + } + }, + "retained_named_subsets": { + "ossben_identifiable_subset": { + "spi_concept": "OSSBEN", + "raw_sources": [ + "BENEFITS.BENEFIT=13:BENAMT", + "BENEFITS.BENEFIT=16,VAR2 in {1,3}:BENAMT" + ], + "formula": "sum(BENAMT where BENEFIT == 13 or (BENEFIT == 16 and VAR2 in {1, 3})) * (365.25 / 7)", + "scope": "identifiable_subset" + }, + "srp_regular_code5": { + "spi_concept": "SRP", + "raw_sources": [ + "BENEFITS.BENEFIT=5:BENAMT" + ], + "formula": "sum(BENAMT where BENEFIT == 5) * (365.25 / 7)", + "scope": "regular_code5_subset" + } + }, + "source_absent_full_constituents": [ + "EPB", + "EXPS", + "TAXTERM", + "MOTHINC", + "OTHERINC" + ], + "full_concepts_forbidden_on_frs": [ + "hmrc_spi_employment_benefits", + "hmrc_spi_employment_expenses", + "hmrc_spi_taxable_termination_pay", + "hmrc_spi_miscellaneous_employment_income", + "hmrc_spi_other_income", + "hmrc_spi_other_social_security_income", + "hmrc_spi_state_pension_income" + ], + "forbid_proxy_substitution": [ + "employment_income", + "miscellaneous_income" + ], + "fail_on_missing_retained_constituent": true, + "fail_on_full_concept_alias": true + } + ], + "outputs": [ + "hmrc_spi_pay", + "hmrc_spi_unemployment_benefit_income", + "hmrc_spi_incapacity_benefit_income", + "ossben_identifiable_subset", + "srp_regular_code5" + ], + "notes": "Retains the adjudicated source-faithful FRS HMRC leaf columns before the SPI income rebuild: full PAY, UBISJA, and INCPBEN, plus explicitly named OSSBEN and SRP subsets. The runtime verifies the certified candidate before retaining these leaves." + }, + { + "stage": "hmrc_spi_income", + "survey": "Survey of Personal Incomes Public Use Tape 2022-23 and HMRC Personal Incomes Tables 3.6/3.7 2023-24", + "source": "https://assets.publishing.service.gov.uk/media/69f1f12d2fae53a03709682f/Collated_Tables_3_1_to_3_11_2324.ods", + "grain": "person", + "artifacts": [ + { + "role": "qrf_donor", + "kind": "private_microdata", + "format": "tab_delimited", + "survey": "Survey of Personal Incomes Public Use Tape 2022-23", + "vintage": "2022-23", + "tax_year_start": 2022, + "ukds_study_number": "SN 9422", + "doi": "10.5255/UKDA-SN-9422-1", + "filename": "put2223uk.tab", + "sha256": "5ef829461060c91a2a47be59ad541d9b519fc3976d66ca80d4920f711bb96f66", + "size_bytes": 141323762, + "reviewed_source": "PolicyEngine licensed UKDS mirror (private Hugging Face repository), spi_2022_23.zip", + "access": "private_local_input", + "locator": "caller-supplied local input", + "runtime_sha256_required": true + }, + { + "role": "published_fact_surface", + "kind": "administrative_table", + "format": "ods", + "survey": "HMRC Personal Incomes Tables 3.6 and 3.7", + "publication": "https://www.gov.uk/government/statistics/personal-incomes-statistics-for-the-tax-year-2023-to-2024", + "vintage": "2023-24", + "tax_year_start": 2023, + "locator": "https://assets.publishing.service.gov.uk/media/69f1f12d2fae53a03709682f/Collated_Tables_3_1_to_3_11_2324.ods", + "sha256": "ad063b06b2bdeef8600dbbb09d48153337a4966f8c7eea50df7a2e0304ebd73e", + "size_bytes": 166693, + "mime_type": "application/vnd.oasis.opendocument.spreadsheet", + "sheets": [ + "Table_3_6", + "Table_3_7" + ], + "mapped_build_period": 2023, + "period_mapping": "tax_year_start", + "runtime_sha256_required": true + } + ], + "operations": [ + { + "kind": "verify_pinned_hmrc_source_pair", + "artifact_roles": [ + "qrf_donor", + "published_fact_surface" + ], + "require_before_source_read": true, + "runtime_sha256_required": true, + "fail_on_mismatch": true + }, + { + "kind": "replace_zero_weight_spi_support", + "existing_channel": "spi", + "require_existing_weight": 0, + "replacement_strata": [ + "clone_index", + "household_is_capital_gains_clone", + "region" + ], + "spi_prior_national_household_mass_share": 0.5, + "output_weight_kind": "importance", + "preserve_total_household_mass": true, + "require_mass_change_record": true, + "mass_change_reason": "Allocate 50% of certified UK national household prior mass to the rebuilt 2022-23 SPI support channel; total national mass is conserved.", + "fail_on_live_existing_spi_mass": true + }, + { + "kind": "strict_read_private_table", + "artifact_role": "qrf_donor", + "filename": "put2223uk.tab", + "delimiter": "\t", + "weight": "FACT", + "required_columns": [ + "AGERANGE", + "GORCODE", + "SEX", + "FACT", + "PAY", + "EPB", + "EXPS", + "TAXTERM", + "INCPBEN", + "OSSBEN", + "UBISJA", + "MOTHINC", + "OTHERINC", + "PROFITS", + "CAPALL", + "LOSSBF", + "SRP", + "INCBBS", + "DIVIDENDS", + "PENSION", + "INCPROP", + "OTHERINV", + "GIFTAID", + "GIFTINV", + "TEI", + "TII", + "TI" + ], + "runtime_sha256_required": true, + "fail_on_missing_file": true, + "fail_on_missing_columns": true, + "fail_on_invalid_weight": true + }, + { + "kind": "fit_weighted_qrf_stage1", + "training_artifact_role": "qrf_donor", + "predictors": [ + "age", + "gender", + "region" + ], + "categorical_predictors": [ + "gender", + "region" + ], + "source_sampling_weight": "FACT", + "sample_size": 100000, + "sample_with_replacement": true, + "post_sample_fit_weight": "uniform", + "fit_weight_kind": "design", + "double_apply_source_weight": false, + "source_columns": { + "self_employment_income": [ + "PROFITS", + "CAPALL", + "LOSSBF" + ], + "savings_interest_income": [ + "INCBBS" + ], + "dividend_income": [ + "DIVIDENDS" + ], + "private_pension_income": [ + "PENSION" + ], + "property_income": [ + "INCPROP" + ], + "other_investment_income": [ + "OTHERINV" + ], + "gift_aid": [ + "GIFTAID" + ], + "charitable_investment_gifts": [ + "GIFTINV" + ], + "hmrc_spi_pay": [ + "PAY" + ], + "hmrc_spi_employment_benefits": [ + "EPB" + ], + "hmrc_spi_employment_expenses": [ + "EXPS" + ], + "hmrc_spi_incapacity_benefit_income": [ + "INCPBEN" + ], + "hmrc_spi_other_social_security_income": [ + "OSSBEN" + ], + "hmrc_spi_taxable_termination_pay": [ + "TAXTERM" + ], + "hmrc_spi_unemployment_benefit_income": [ + "UBISJA" + ], + "hmrc_spi_miscellaneous_employment_income": [ + "MOTHINC" + ], + "hmrc_spi_other_income": [ + "OTHERINC" + ], + "hmrc_spi_state_pension_income": [ + "SRP" + ] + }, + "derived_policyengine_outputs": { + "employment_income": { + "source_columns": [ + "PAY", + "EPB", + "TAXTERM" + ], + "formula": "hmrc_spi_pay + hmrc_spi_employment_benefits + hmrc_spi_taxable_termination_pay", + "derive_after_draw": true + } + }, + "outputs": [ + "self_employment_income", + "savings_interest_income", + "dividend_income", + "private_pension_income", + "property_income", + "other_investment_income", + "gift_aid", + "charitable_investment_gifts", + "hmrc_spi_pay", + "hmrc_spi_employment_benefits", + "hmrc_spi_employment_expenses", + "hmrc_spi_incapacity_benefit_income", + "hmrc_spi_other_social_security_income", + "hmrc_spi_taxable_termination_pay", + "hmrc_spi_unemployment_benefit_income", + "hmrc_spi_miscellaneous_employment_income", + "hmrc_spi_other_income", + "hmrc_spi_state_pension_income" + ], + "joint_draw": true, + "savings_interest_source_semantics": "INCBBS is taxable bank/building-society interest before reconstruction to the PolicyEngine gross input", + "employment_income_source_semantics": "PolicyEngine input = PAY + EPB + TAXTERM, matching the pinned enhanced-FRS pipeline; it is not the Table 3.6 measure", + "hmrc_employed_income_source_semantics": "Derived after each draw as max(0, PAY + EPB - EXPS) + INCPBEN + OSSBEN + TAXTERM + UBISJA + MOTHINC, using normalized leaves identically on FRS and SPI channels", + "self_employment_income_source_semantics": "max(0, PROFITS - CAPALL - LOSSBF)", + "assessable_income_source_semantics": "QRF draws leaves only; TEI, TII, and TI are deterministic post-draw accounting aggregates and TI equals TEI + TII exactly", + "source_ti_identity_fields": [ + "TI", + "TEI", + "TII" + ], + "source_leaf_reconciliation": { + "documentation_url": "https://doc.ukdataservice.ac.uk/doc/9422/mrdoc/pdf/9422_put_2223_full_documentation.pdf", + "composite_indicator": "AGERANGE == -1", + "formulas": { + "TEI": "max(0, PAY + EPB - EXPS) + INCPBEN + OSSBEN + TAXTERM + UBISJA + MOTHINC + OTHERINC + SRP + PENSION + max(0, PROFITS - CAPALL - LOSSBF)", + "TII": "OTHERINV + DIVIDENDS + INCPROP + INCBBS", + "TI": "TEI + TII" + }, + "maximum_absolute_difference_gbp": { + "ordinary": { + "TEI": 15, + "TII": 10, + "TI": 20 + }, + "composite": { + "TEI": 180, + "TII": 10, + "TI": 180 + } + }, + "rationale": "The official PUT rounds source fields, averages documented composite records, then rounds remaining income fields to GBP 5. These are the observed envelopes in the exact sha-pinned donor; post-draw synthetic identities remain exact." + }, + "ti_identity_absolute_tolerance_gbp": 5, + "stochastic_aggregates_forbidden": [ + "hmrc_spi_employed_income", + "hmrc_spi_total_earned_income", + "hmrc_spi_total_investment_income", + "hmrc_spi_assessable_income" + ], + "require_all_predictors": true, + "require_all_outputs": true + }, + { + "kind": "fit_weighted_qrf_stage2", + "training_population": "certified_microcosm_uk_candidate_base_channel", + "target_population": "rebuilt_spi_support_channel", + "predictors": [ + "age", + "gender", + "region", + "employment_income", + "self_employment_income", + "savings_interest_income", + "dividend_income", + "private_pension_income", + "property_income" + ], + "reviewed_absent_predictors": { + "other_investment_income": "This remains a stage-1 SPI draw and an official HMRC fact component, but it is not an FRS-only stage-2 predictor: the incumbent UK data build's frs_only.py defines exactly six income predictors and the certified Microcosm UK base candidate has no other_investment_income column." + }, + "categorical_predictors": [ + "gender", + "region" + ], + "weight": "household_weight", + "weight_mapping": "household_to_person", + "outputs": [ + "employee_pension_contributions", + "employer_pension_contributions", + "personal_pension_contributions", + "pension_contributions_via_salary_sacrifice", + "tax_free_savings_income", + "universal_credit_reported", + "pension_credit_reported", + "child_benefit_reported", + "housing_benefit_reported", + "income_support_reported", + "working_tax_credit_reported", + "child_tax_credit_reported", + "attendance_allowance_reported", + "state_pension_reported", + "dla_sc_reported", + "dla_m_reported", + "pip_m_reported", + "pip_dl_reported", + "sda_reported", + "carers_allowance_reported", + "iidb_reported", + "afcs_reported", + "bsp_reported", + "winter_fuel_allowance_reported", + "council_tax_benefit_reported", + "jsa_contrib_reported", + "jsa_income_reported", + "esa_contrib_reported", + "esa_income_reported" + ], + "reviewed_absent_outputs": { + "incapacity_benefit_reported": "Absent/all-default on the pinned enhanced-FRS export and certified Microcosm UK base; not a populated loader layer.", + "maternity_allowance_reported": "Absent from the pinned enhanced-FRS export and certified Microcosm UK base; no training source can be materialized for this stage." + }, + "postprocess": { + "gross_savings_interest_income": "stage1 INCBBS draw + stage2 tax_free_savings_income", + "refresh_disability_categories": [ + "aa_category", + "dla_sc_category", + "dla_m_category", + "pip_m_category", + "pip_dl_category" + ], + "refresh_disability_flags": [ + "is_disabled_for_benefits", + "is_enhanced_disabled_for_benefits", + "is_severely_disabled_for_benefits" + ] + }, + "joint_draw": true, + "require_all_predictors": true, + "require_all_materializable_outputs": true, + "require_all_outputs": false + }, + { + "kind": "materialize_hmrc_income_bands_fail_closed", + "artifact_role": "published_fact_surface", + "mapped_build_period": 2023, + "period_mapping": "tax_year_start", + "column_index_base": 0, + "data_row_start_index": 5, + "stop_label": "All ranges", + "count_unit_multiplier": 1000, + "amount_unit_multiplier": 1000000, + "component_columns": { + "employment_income": { + "sheet": "Table_3_6", + "count_column_index": 4, + "amount_column_index": 5 + }, + "self_employment_income": { + "sheet": "Table_3_6", + "count_column_index": 1, + "amount_column_index": 2 + }, + "state_pension": { + "sheet": "Table_3_6", + "count_column_index": 7, + "amount_column_index": 8 + }, + "private_pension_income": { + "sheet": "Table_3_6", + "count_column_index": 10, + "amount_column_index": 11 + }, + "property_income": { + "sheet": "Table_3_7", + "count_column_index": 1, + "amount_column_index": 2 + }, + "savings_interest_income": { + "sheet": "Table_3_7", + "count_column_index": 4, + "amount_column_index": 5 + }, + "dividend_income": { + "sheet": "Table_3_7", + "count_column_index": 7, + "amount_column_index": 8 + }, + "other_investment_income": { + "sheet": "Table_3_7", + "count_column_index": 10, + "amount_column_index": 11 + } + }, + "required_band_lower_bounds_gbp": [ + 12570, + 15000, + 20000, + 30000, + 40000, + 50000, + 70000, + 100000, + 150000, + 200000, + 300000, + 500000, + 1000000 + ], + "required_measures": [ + "count", + "amount" + ], + "fail_on_missing_sheet": true, + "fail_on_missing_component": true, + "fail_on_missing_band": true, + "fail_on_non_numeric_value": true + }, + { + "kind": "classify_hmrc_income_facts_with_reviewed_fences", + "target_operation": "materialize_hmrc_income_bands_fail_closed", + "components": [ + "employment_income", + "self_employment_income", + "state_pension", + "private_pension_income", + "property_income", + "savings_interest_income", + "dividend_income", + "other_investment_income" + ], + "breakdown_dependency": "hmrc_spi_assessable_income", + "frs_breakdown_status": "unavailable_full_measure", + "input_weight_kind": "importance", + "output_weight_kind": "importance", + "calibration_permitted": false, + "required_fact_count": 208, + "outcome_counts": { + "exact_pass": 0, + "exact_fail": 0, + "directional_pass": 0, + "directional_fail": 0, + "excluded_with_fence": 208 + }, + "classification_rationale": "Every published fact uses non-overlapping total-income bands. The FRS channel cannot materialize full TEI, and omitted income can move a person between bands, so neither an exact fact nor a per-band directional bound is valid.", + "reviewed_fences": [ + { + "fence_id": "frs_epb_source_absent", + "constituents": [ + "EPB" + ], + "raw_sources_searched": [ + "JOB.EXPBEN01-EXPBEN13", + "JOB.CARVAL", + "JOB.CARAMT", + "JOB.FUELAMT", + "JOB.VCHAMT", + "JOB.CHVAMT" + ], + "finding": "Missing. EXPBEN* are receipt flags, and the amount fields cover only selected benefits; they cannot produce complete taxable expenses payments and benefits.", + "mass_implication": "12.9485464% of certified-candidate FRS effective person mass has at least one receipt flag, but this is not monetary support.", + "rationale": "Receipt flags and selected benefit amounts cannot be promoted to the SPI EPB monetary concept without an imputation or proxy.", + "dependent_fence_ids": [] + }, + { + "fence_id": "frs_exps_source_absent", + "constituents": [ + "EXPS" + ], + "raw_sources_searched": [ + "JOB.EXPBEN04/EXPBEN05", + "JOB.MILEAMT/JOB.MOTAMT", + "JOB.UMILEAMT/JOB.UMOTAMT", + "JOB.DEDUC1-DEDUC9", + "JOB.UDEDUC1-UDEDUC9" + ], + "finding": "Missing. These fields describe reimbursements or payroll deductions, not the complete tax-deductible employment-expense amount required by SPI.", + "mass_implication": "5.1302528% of certified-candidate FRS effective person mass has an adjacent reimbursement flag; the true EXPS mass is not estimable.", + "rationale": "The nearby fields do not measure the required deductible amount, and EXPS enters the employed-income identity with a negative sign.", + "dependent_fence_ids": [] + }, + { + "fence_id": "frs_taxterm_source_absent", + "constituents": [ + "TAXTERM" + ], + "raw_sources_searched": [ + "ADULT.REDAMT", + "ADULT and JOB taxable-termination split search" + ], + "finding": "Missing. REDAMT is gross redundancy pay and has neither the taxable amount nor non-redundancy termination pay.", + "mass_implication": "0.3746084% of certified-candidate FRS effective person mass has positive gross redundancy pay; taxable mass is unknown.", + "rationale": "Gross redundancy pay cannot be relabeled as taxable termination pay.", + "dependent_fence_ids": [] + }, + { + "fence_id": "frs_mothinc_source_absent", + "constituents": [ + "MOTHINC" + ], + "raw_sources_searched": [ + "ODDJOB.OJAMT/ODDJOB.OJNOW", + "ADULT.ALLPAY2", + "ADULT.ROYYR2-ROYYR4", + "JOB.OWNOTHER" + ], + "finding": "Missing. The fields are heterogeneous and belong to distinct income concepts; assigning their union to SPI miscellaneous employment income would be a proxy.", + "mass_implication": "Odd-job-only effective person mass is 0.1724207%; the broader unresolved miscellaneous pool is 1.4650566%.", + "rationale": "The FRS instrument cannot separate the SPI miscellaneous-employment concept source-faithfully.", + "dependent_fence_ids": [] + }, + { + "fence_id": "frs_otherinc_source_absent", + "constituents": [ + "OTHERINC" + ], + "raw_sources_searched": [ + "ADULT, ODDJOB, and JOB miscellaneous fields", + "PENSION", + "ACCOUNTS", + "ASSETS", + "BENEFITS" + ], + "finding": "Missing. No person-level raw FRS variable has SPI OTHERINC semantics, and the miscellaneous pool cannot be split between MOTHINC and OTHERINC from source evidence.", + "mass_implication": "No separable mass estimate exists; the unresolved miscellaneous pool is 1.4650566% of certified-candidate FRS effective person mass.", + "rationale": "A union of heterogeneous residual fields would be a new proxy, not a retained source constituent.", + "dependent_fence_ids": [] + }, + { + "fence_id": "frs_ossben_identifiable_subset", + "constituents": [ + "OSSBEN", + "ossben_identifiable_subset" + ], + "raw_sources_searched": [ + "BENEFITS.BENAMT", + "BENEFITS.BENEFIT", + "BENEFITS.VAR2", + "BENEFITS codes 13, 16, 6, and 30" + ], + "finding": "Incomplete. Carer's Allowance and contribution-based ESA form an identifiable subset, but code 6 mixes tax treatments and code 30 is an undifferentiated catch-all, so the complete taxable family cannot be emitted.", + "mass_implication": "1.8045088% of certified-candidate FRS effective person mass carries the identifiable lower-bound subset; it is not full OSSBEN support.", + "rationale": "The retained column must remain explicitly named as a subset and cannot satisfy the full SPI concept.", + "dependent_fence_ids": [] + }, + { + "fence_id": "frs_srp_regular_code5_subset", + "constituents": [ + "SRP", + "srp_regular_code5" + ], + "raw_sources_searched": [ + "BENEFITS.BENAMT where BENEFIT == 5", + "BENEFITS codes 6 and 9" + ], + "finding": "Incomplete. Code 5 supplies regular State Pension, but the FRS source does not identify the full SPI combination of State Pension lump sums and widow's pension; code 6 mixes benefits and code 9 is tax-free War Widow's Pension.", + "mass_implication": "18.1567916% of certified-candidate FRS effective person mass carries regular code-5 State Pension; it is not complete SRP support.", + "rationale": "The retained column must remain explicitly named as a subset and cannot be reported as the full published state-pension measure.", + "dependent_fence_ids": [] + }, + { + "fence_id": "full_frs_tei_band_unavailable", + "constituents": [ + "EPB", + "EXPS", + "TAXTERM", + "MOTHINC", + "OTHERINC", + "OSSBEN", + "SRP" + ], + "raw_sources_searched": [], + "finding": "The complete FRS TEI measure cannot be materialized from retained source constituents, so exact HMRC total-income band assignment is unavailable on the FRS channel.", + "mass_implication": "Every one of the 208 published facts is banded by total income and therefore depends on this unavailable like-for-like measure.", + "rationale": "A component-level subset does not imply a per-band lower bound: omitted income can move a taxpayer into or out of any non-overlapping published band. Biased partial bands are not emitted as estimates.", + "dependent_fence_ids": [ + "frs_epb_source_absent", + "frs_exps_source_absent", + "frs_taxterm_source_absent", + "frs_mothinc_source_absent", + "frs_otherinc_source_absent", + "frs_ossben_identifiable_subset", + "frs_srp_regular_code5_subset" + ] + } + ], + "fact_fence_id": "full_frs_tei_band_unavailable", + "blocked_dependency": "hmrc_spi_assessable_income", + "fail_on_unfenced_exclusion": true, + "fail_on_fact_count_mismatch": true, + "forbid_biased_estimate_or_delta": true + }, + { + "kind": "gate_distributional_effective_mass", + "columns": [ + "gift_aid", + "charitable_investment_gifts" + ], + "weight": "household_weight", + "weight_mapping": "household_to_person", + "support_channel_column": "person_support_channel", + "required_support_channel": "spi", + "mass_share_denominator": "all_person_effective_mass", + "minimum_nondefault_mass_share": 1e-06, + "fail_below_floor": true + } + ], + "official_table_components": [ + "employment_income", + "self_employment_income", + "state_pension", + "private_pension_income", + "property_income", + "savings_interest_income", + "dividend_income", + "other_investment_income" + ], + "donor_relief_outputs": [ + "gift_aid", + "charitable_investment_gifts" + ], + "outputs": [ + "employment_income", + "self_employment_income", + "state_pension", + "private_pension_income", + "property_income", + "savings_interest_income", + "dividend_income", + "other_investment_income", + "gift_aid", + "charitable_investment_gifts", + "hmrc_spi_employed_income", + "hmrc_spi_total_earned_income", + "hmrc_spi_total_investment_income", + "hmrc_spi_assessable_income" + ], + "notes": "Current-source adjudicated replay contract: the private 2022-23 SPI donor and public 2023-24 HMRC ODS are pinned by reviewed SHA-256 and size and verified together before either is opened. The QRF draws source leaves; HMRC employed income, TEI, TII, and TI are deterministic post-draw aggregates on the SPI channel, with TI exactly equal to TEI + TII. PolicyEngine employment_income remains the narrow PAY + EPB + TAXTERM input on SPI rows. Stage 2 mirrors the incumbent UK data build's frs_only.py exactly: its income predictors are employment, self-employment, savings interest, dividends, private pension, and property income. Other investment income remains a stage-1 SPI draw and official HMRC fact component, but is excluded from stage 2 because the certified FRS candidate does not carry it. The FRS channel retains source-faithful full PAY, UBISJA, and INCPBEN plus explicitly named ossben_identifiable_subset and srp_regular_code5; EPB, EXPS, TAXTERM, MOTHINC, OTHERINC, full OSSBEN, and full SRP remain forbidden. Because the missing legs prevent a complete FRS TEI measure, none of the 208 non-overlapping total-income-band facts is exact or directional. Every fact is an excluded-with-fence record, no calibration is performed, and weights remain importance-kind. Gift Aid restoration still requires the rebuilt positive-mass SPI channel to clear the reviewed 1ppm effective-mass floor." + } + ] +} diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/national_build.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/national_build.py index fd491949..7059a3fa 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/national_build.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/national_build.py @@ -43,6 +43,8 @@ GatePhaseReport, ) from microcosm.build.gates import GateResult +from microcosm.build.plan import Stage as PlanStage +from microcosm.build.plan import StagePlan, StageRecord from microcosm.build.uk_runtime.battery_bindings import UK_GATE_REGISTRY from microcosm.build.uk_runtime.national_frame import ( UKStagingProvenance, @@ -102,7 +104,14 @@ @dataclass(frozen=True) class UKNationalStage: - """One named, deterministic ``Frame -> Frame`` national transform.""" + """Deprecated shim for one named ``Frame -> Frame`` national transform. + + UK national builds now execute shared :class:`microcosm.build.plan.Stage` + entries assembled into a :class:`microcosm.build.plan.StagePlan`. This + wrapper remains for one release so existing callers can pass a stage name + and transform; :func:`build_uk_national_dataset` converts it internally to + a shared stage with empty consumes/produces and no donor. + """ name: str transform: Callable[[Frame], Frame] @@ -141,6 +150,8 @@ class UKNationalBuildResult: #: alias; this field always carries the full battery payload). gate_report: Mapping[str, object] terminal_gate_path: Path + #: Shared stage evidence records, one per national build stage. + stage_records: tuple[StageRecord, ...] = () #: The #627 rung receipt; ``None`` on a full-scale (fraction 1.0) build. sampling_receipt: Mapping[str, object] | None = None @@ -333,7 +344,7 @@ def build_uk_national_dataset( staging_h5: str | Path, release_id: str, calibration_diagnostics_sha256: str, - stages: Sequence[UKNationalStage] = (), + stages: Sequence[UKNationalStage | PlanStage] | StagePlan = (), coverage_engine: Any | None = None, parity_evidence: UKReleaseParityEvidence | None = None, input_mass_reference: UKInputMassReference | None = None, @@ -413,7 +424,8 @@ def build_uk_national_dataset( "terminal_gate_path must differ from the input and staging H5 paths." ) - materialized_stages = tuple(stages) + stage_plan = _coerce_stage_plan(stages) + materialized_stages = stage_plan.stages _validate_stages(materialized_stages) # Invariant: no destructive step precedes argument validation. Every # configuration refusal sits above the sidecar unlinks and the battery, @@ -507,11 +519,9 @@ def build_uk_national_dataset( if callable(binder): binder(provenance, frame) if checkpoint_dir is None: - for stage in materialized_stages: - frame = stage.run(frame) - validate_uk_national_frame(frame) + frame, stage_records = _validating_stage_plan(stage_plan).run(frame) else: - frame = _run_stages_checkpointed( + frame, stage_records = _run_stages_checkpointed( materialized_stages, frame=frame, checkpoint_dir=Path(checkpoint_dir), @@ -594,17 +604,18 @@ def build_uk_national_dataset( ), gate_report=gate_report, terminal_gate_path=diagnostic_path, + stage_records=stage_records, sampling_receipt=sampling_receipt, ) def _run_stages_checkpointed( - stages: tuple[UKNationalStage, ...], + stages: tuple[PlanStage, ...], *, frame: Frame, checkpoint_dir: Path, run_config: Mapping[str, object], -) -> Frame: +) -> tuple[Frame, tuple[StageRecord, ...]]: """Run the national stages through the outer stage runtime. Each boundary persists a lossless Frame checkpoint (frame metadata rides @@ -636,6 +647,7 @@ def _run_stages_checkpointed( ) runtime = StageRuntime(checkpoint_dir, pipeline, run_config=dict(run_config)) completed = set(runtime.context.completed) + records: list[StageRecord] = [] for stage in stages: if stage.name in completed: loaded = load_uk_stage_checkpoint(runtime, stage.name) @@ -647,10 +659,12 @@ def _run_stages_checkpointed( if key != UK_FRAME_METADATA_KEY } resume(extra, loaded.frame) - frame = loaded.frame + frame, stage_records = _resume_stage_record(stage, frame, loaded.frame) + validate_uk_national_frame(frame) + records.extend(stage_records) continue - frame = stage.run(frame) - validate_uk_national_frame(frame) + frame, stage_records = _validating_stage_plan(StagePlan((stage,))).run(frame) + records.extend(stage_records) extra_metadata: dict[str, object] = {} hook = getattr(stage.transform, "checkpoint_metadata", None) if callable(hook): @@ -660,15 +674,84 @@ def _run_stages_checkpointed( frame, metadata=uk_stage_metadata(frame, extra=extra_metadata), ) - return frame + return frame, tuple(records) + + +def _coerce_stage_plan( + stages: Sequence[UKNationalStage | PlanStage] | StagePlan, +) -> StagePlan: + """Normalize legacy UK stages and shared stages to one StagePlan.""" + + if isinstance(stages, StagePlan): + return stages + materialized = tuple(_coerce_stage(stage) for stage in stages) + names: set[str] = set() + for stage in materialized: + if stage.name in names: + raise ValueError(f"Duplicate UK national stage {stage.name!r}.") + names.add(stage.name) + return StagePlan(materialized) + + +def _coerce_stage(stage: UKNationalStage | PlanStage) -> PlanStage: + if isinstance(stage, PlanStage): + return stage + if isinstance(stage, UKNationalStage): + return PlanStage(name=stage.name, transform=stage.transform) + raise TypeError( + "UK national stages must be shared Stage or UKNationalStage instances, " + f"got {type(stage).__name__}." + ) + + +def _validating_stage_plan(plan: StagePlan) -> StagePlan: + """Return a plan whose stages validate the UK national frame after each run.""" + + return StagePlan(_validating_stage(stage) for stage in plan.stages) + + +def _validating_stage(stage: PlanStage) -> PlanStage: + def transform(frame: Frame) -> Frame: + result = stage.transform(frame) + if not isinstance(result, Frame): + return result + validate_uk_national_frame(result) + return result + + return PlanStage( + name=stage.name, + transform=transform, + produces=stage.produces, + consumes=stage.consumes, + donor=stage.donor, + ) + + +def _resume_stage_record( + stage: PlanStage, + previous: Frame, + loaded: Frame, +) -> tuple[Frame, tuple[StageRecord, ...]]: + plan = StagePlan( + ( + PlanStage( + name=stage.name, + transform=lambda _frame: loaded, + produces=stage.produces, + consumes=stage.consumes, + donor=stage.donor, + ), + ) + ) + return plan.run(previous) -def _validate_stages(stages: tuple[UKNationalStage, ...]) -> None: +def _validate_stages(stages: tuple[PlanStage, ...]) -> None: names: set[str] = set() for stage in stages: - if not isinstance(stage, UKNationalStage): + if not isinstance(stage, PlanStage): raise TypeError( - "UK national stages must be UKNationalStage instances, " + "UK national stages must be shared Stage instances, " f"got {type(stage).__name__}." ) if stage.name in names: @@ -677,7 +760,7 @@ def _validate_stages(stages: tuple[UKNationalStage, ...]) -> None: def _stage_fit_weight_records( - stages: tuple[UKNationalStage, ...], + stages: tuple[PlanStage, ...], ) -> tuple[object, ...] | None: """The weights-audit evidence artifact: present iff the HMRC stage is. diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/source_runtime.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/source_runtime.py new file mode 100644 index 00000000..eb167f9c --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/source_runtime.py @@ -0,0 +1,113 @@ +"""UK source runtime seams for the shared source-stage manifest.""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +from typing import Any + +import pandas as pd + +from microcosm.build.source_manifest import SourceOperationSpec +from microcosm.build.source_runtime import ( + SourceOperationHandler, + SourceRuntimeContext, + SourceRuntimeError, +) +from microcosm.frame import Frame +from microcosm.frame.rules import materialize_rules_engine_predictors + +__all__ = [ + "materialize_uk_rules_engine_predictors_from_manifest", + "uk_source_operation_handlers", + "uk_stage_implementations", +] + + +def uk_stage_implementations( + *, + retained_leaves_transform: Callable[[Frame], Frame], + hmrc_income_transform: Callable[[Frame], Frame], +) -> dict[str, Callable[[Frame], Frame]]: + """Return the whole-stage implementation map for the UK source plan.""" + + return { + "frs_hmrc_retained_leaves": retained_leaves_transform, + "hmrc_spi_income": hmrc_income_transform, + } + + +def uk_source_operation_handlers() -> Mapping[str, SourceOperationHandler]: + """Return UK operation handlers available to declarative source runtimes.""" + + return { + "materialize_rules_engine_predictors": ( + materialize_uk_rules_engine_predictors_from_manifest + ) + } + + +def materialize_uk_rules_engine_predictors_from_manifest( + current: pd.DataFrame | None, + operation: SourceOperationSpec, + context: SourceRuntimeContext, +) -> Frame: + """Materialize declared rules-engine predictors onto a runtime Frame.""" + + if current is not None: + raise SourceRuntimeError( + "materialize_rules_engine_predictors operates on the runtime Frame, " + "not an intermediate source table." + ) + frame = _extra(context, "frame", Frame) + engine = _extra(context, "rules_engine", object) + country = _optional_country(context) + period = context.config.extra.get("period", context.config.target_year) + if period is None: + raise SourceRuntimeError( + "materialize_rules_engine_predictors requires a period in the " + "runtime context." + ) + predictors = operation.parameters.get("predictors") + if not isinstance(predictors, list) or not all( + isinstance(name, str) and name for name in predictors + ): + raise SourceRuntimeError( + "materialize_rules_engine_predictors requires a non-empty " + "'predictors' list." + ) + try: + return materialize_rules_engine_predictors( + frame, + variables=tuple(predictors), + period=period, + engine=engine, + country=country, + ) + except ValueError as error: + raise SourceRuntimeError(str(error)) from error + + +def _extra( + context: SourceRuntimeContext, + key: str, + expected_type: type, +) -> Any: + value = context.config.extra.get(key) + if not isinstance(value, expected_type): + raise SourceRuntimeError( + f"materialize_rules_engine_predictors requires context.config.extra" + f"[{key!r}] to be a {expected_type.__name__}." + ) + return value + + +def _optional_country(context: SourceRuntimeContext) -> str | None: + country = context.config.extra.get("country") + if country is None: + return None + if not isinstance(country, str) or not country: + raise SourceRuntimeError( + "materialize_rules_engine_predictors context country must be a " + "non-empty string." + ) + return country diff --git a/packages/microcosm-build/tests/test_country_spec.py b/packages/microcosm-build/tests/test_country_spec.py index 9ac420a1..22485574 100644 --- a/packages/microcosm-build/tests/test_country_spec.py +++ b/packages/microcosm-build/tests/test_country_spec.py @@ -245,6 +245,7 @@ def test_uk_package_loads(self) -> None: "hmrc_income_release_gate_report.json", "hmrc_income_replay_report.json", "hmrc_income_source_stages.json", + "source_stages.json", "input_mass_reviewed_exclusions.json", "national_staging_build_record.json", "qrf_tail_reviewed_exclusions.json", diff --git a/packages/microcosm-build/tests/test_uk_national_build.py b/packages/microcosm-build/tests/test_uk_national_build.py index 818b70d0..2016363f 100644 --- a/packages/microcosm-build/tests/test_uk_national_build.py +++ b/packages/microcosm-build/tests/test_uk_national_build.py @@ -7,11 +7,13 @@ import pandas as pd import pytest +from microcosm.build.country_spec import country_stage_plan, load_country_spec from microcosm.build.gate_battery import ( GateBatteryBlockedError, gate_signing_key_env, ) from microcosm.build.gates import FitWeightRecord, GateResult +from microcosm.build.plan import Stage, StagePlan from microcosm.build.uk_runtime.battery_bindings import UKGateBinding from microcosm.build.uk_runtime.national_build import ( UKNationalStage, @@ -113,6 +115,30 @@ def _replace_person(frame: Frame, person: pd.DataFrame) -> Frame: ) +def _assert_same_frame_payload(left: Frame, right: Frame) -> None: + assert left.schema == right.schema + assert left.entities == right.entities + for entity in left.entities: + pd.testing.assert_frame_equal( + left.table(entity), + right.table(entity), + check_exact=True, + check_dtype=True, + ) + assert left.weighted_entities == right.weighted_entities + for entity in left.weighted_entities: + assert left.weights_for(entity).kind is right.weights_for(entity).kind + pd.testing.assert_series_equal( + pd.Series(left.weights_for(entity).values), + pd.Series(right.weights_for(entity).values), + check_exact=True, + check_dtype=True, + ) + pd.testing.assert_series_equal(left.strata, right.strata, check_exact=True) + assert left.mass_log == right.mass_log + assert left.metadata == right.metadata + + @pytest.fixture(autouse=True) def _trusted_terminal_gate_signing_key(monkeypatch) -> None: monkeypatch.setenv( @@ -404,6 +430,92 @@ def recording_writer(frame, path): assert diagnostic["input_coverage"]["passed"] is True +def test_national_build_accepts_stage_plan_and_records_stage_evidence( + tmp_path, +) -> None: + pytest.importorskip("tables") + + input_h5 = tmp_path / "base.h5" + staging_h5 = tmp_path / "staging.h5" + _write_toy_h5(input_h5) + + def add_bonus(frame: Frame) -> Frame: + person = frame.table("person").copy() + person["bonus_income"] = [125.0] + return _replace_person(frame, person) + + result = _run_national_build( + input_h5=input_h5, + staging_h5=staging_h5, + stages=StagePlan( + ( + Stage( + name="income", + transform=add_bonus, + produces=("bonus_income",), + ), + ) + ), + coverage_engine=object(), + gate_registry=_registry_with_coverage(_passing_gate), + ) + + assert result.stage_names == ("income",) + assert [record.stage for record in result.stage_records] == ["income"] + assert result.stage_records[0].produced == ("bonus_income",) + assert result.stage_records[0].nonzero_share == {"bonus_income": 1.0} + + +def test_deprecated_shim_and_country_stage_plan_paths_are_payload_identical( + tmp_path, +) -> None: + pytest.importorskip("tables") + + input_h5 = tmp_path / "base.h5" + _write_toy_h5(input_h5, employment_income=40_000.0) + spec = load_country_spec("uk") + retained_outputs = spec.sources.stages[0].outputs + hmrc_outputs = spec.sources.stages[1].outputs + + def retained(frame: Frame) -> Frame: + person = frame.table("person").copy() + for index, column in enumerate(retained_outputs, start=1): + person[column] = float(index) + return _replace_person(frame, person) + + def hmrc(frame: Frame) -> Frame: + person = frame.table("person").copy() + for index, column in enumerate(hmrc_outputs, start=1): + person[column] = float(index * 10) + return _replace_person(frame, person) + + legacy = _run_national_build( + input_h5=input_h5, + staging_h5=tmp_path / "legacy.h5", + stages=( + UKNationalStage("frs_hmrc_retained_leaves", retained), + UKNationalStage("hmrc_spi_income", hmrc), + ), + coverage_engine=object(), + gate_registry=_registry_with_coverage(_passing_gate), + ) + shared = _run_national_build( + input_h5=input_h5, + staging_h5=tmp_path / "shared.h5", + stages=country_stage_plan( + spec, + { + "frs_hmrc_retained_leaves": retained, + "hmrc_spi_income": hmrc, + }, + ), + coverage_engine=object(), + gate_registry=_registry_with_coverage(_passing_gate), + ) + + _assert_same_frame_payload(legacy.frame, shared.frame) + + def _write_clone_family_h5(path: Path) -> None: """Four base clone families (canonical + one geography clone each). diff --git a/packages/microcosm-build/tests/test_uk_national_build_driver.py b/packages/microcosm-build/tests/test_uk_national_build_driver.py index ca6fc811..0383727d 100644 --- a/packages/microcosm-build/tests/test_uk_national_build_driver.py +++ b/packages/microcosm-build/tests/test_uk_national_build_driver.py @@ -160,10 +160,11 @@ def test_national_build_driver_uses_standalone_national_seam( def fake_build(**kwargs): calls.append(kwargs) - kwargs["stages"][0].transform.last_result = SimpleNamespace( + stages = kwargs["stages"] + stages[0].transform.last_result = SimpleNamespace( evidence=lambda: {"stage": "frs_hmrc_retained_leaves"} ) - kwargs["stages"][1].transform.last_result = SimpleNamespace( + stages[1].transform.last_result = SimpleNamespace( evidence=lambda: {"stage": "hmrc_spi_income"}, replay_report=SimpleNamespace(summary={"excluded_with_fence": 208}), ) @@ -239,13 +240,14 @@ def fake_build(**kwargs): assert calls[0]["staging_h5"] == staging_h5 assert calls[0]["release_id"] == "populace-uk-2023-frs-k535080" assert calls[0]["calibration_diagnostics_sha256"] == "c" * 64 - assert len(calls[0]["stages"]) == 3 - assert calls[0]["stages"][0].name == "frs_hmrc_retained_leaves" - retained_transform = calls[0]["stages"][0].transform + stages = calls[0]["stages"] + assert len(stages) == 3 + assert stages[0].name == "frs_hmrc_retained_leaves" + retained_transform = stages[0].transform assert retained_transform.adult_tab_path == adult_tab assert retained_transform.benefits_tab_path == benefits_tab - assert calls[0]["stages"][1].name == "hmrc_spi_income" - hmrc_transform = calls[0]["stages"][1].transform + assert stages[1].name == "hmrc_spi_income" + hmrc_transform = stages[1].transform assert hmrc_transform.spi_tab_path == spi_tab assert hmrc_transform.hmrc_ods_path == hmrc_ods assert hmrc_transform.certified_candidate.revision == "test-revision" @@ -465,10 +467,11 @@ def test_national_driver_writes_aggregate_reports_before_reraising_final_gate( replay_report = object() def fake_build(**kwargs): - kwargs["stages"][0].transform.last_result = SimpleNamespace( + stages = kwargs["stages"] + stages[0].transform.last_result = SimpleNamespace( evidence=lambda: {"stage": "frs_hmrc_retained_leaves"} ) - kwargs["stages"][1].transform.last_result = SimpleNamespace( + stages[1].transform.last_result = SimpleNamespace( evidence=lambda: {"stage": "hmrc_spi_income"}, replay_report=replay_report, ) diff --git a/packages/microcosm-build/tests/test_uk_source_runtime.py b/packages/microcosm-build/tests/test_uk_source_runtime.py new file mode 100644 index 00000000..e30c4fac --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_source_runtime.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.source_manifest import SourceOperationSpec +from microcosm.build.source_runtime import ( + SourceRuntimeConfig, + SourceRuntimeContext, + SourceRuntimeError, +) +from microcosm.build.uk_runtime.source_runtime import ( + uk_source_operation_handlers, + uk_stage_implementations, +) +from microcosm.frame import EntitySchema, ExportContract, Frame, WeightKind, Weights +from microcosm.frame.schema import VariableMetadata + + +class StubRulesEngine: + country = "uk" + + def __init__(self, values: Sequence[float] = (12.0, 34.0)) -> None: + self._values = np.asarray(values, dtype=float) + + def variable_metadata(self, name: str) -> VariableMetadata: + return VariableMetadata( + name=name, entity="person", dtype="float", period="year" + ) + + def variables(self) -> Sequence[str]: + return ("projected_income",) + + def entity_schema(self) -> EntitySchema: + return _schema() + + def materialize( + self, + bundle: Frame, + variables: Sequence[str], + period: int | str, + ) -> Mapping[str, np.ndarray]: + assert variables == ("projected_income",) + assert period == 2023 + assert bundle.n("person") == 2 + return {"projected_income": self._values} + + def export_contract(self) -> ExportContract: + return ExportContract.empty() + + def write_dataset( + self, + bundle: Frame, + path: str, + period: int | str, + ) -> None: + raise NotImplementedError + + +def _schema() -> EntitySchema: + return EntitySchema(group_entities=("household",)) + + +def _frame() -> Frame: + return Frame( + { + "person": pd.DataFrame( + { + "person_id": [1, 2], + "person_household_id": [10, 20], + } + ), + "household": pd.DataFrame({"household_id": [10, 20]}), + }, + _schema(), + { + "household": Weights( + values=np.asarray([1.0, 2.0]), + kind=WeightKind.DESIGN, + ) + }, + ) + + +def _operation() -> SourceOperationSpec: + return SourceOperationSpec.from_mapping( + { + "kind": "materialize_rules_engine_predictors", + "predictors": ["projected_income"], + } + ) + + +def _context(*, engine: object, country: str = "uk") -> SourceRuntimeContext: + return SourceRuntimeContext( + config=SourceRuntimeConfig( + target_year=2023, + extra={"frame": _frame(), "rules_engine": engine, "country": country}, + ), + tables={}, + ) + + +def test_uk_stage_implementations_names_whole_stage_transforms() -> None: + def retained(frame: Frame) -> Frame: + return frame + + def hmrc(frame: Frame) -> Frame: + return frame + + assert uk_stage_implementations( + retained_leaves_transform=retained, + hmrc_income_transform=hmrc, + ) == { + "frs_hmrc_retained_leaves": retained, + "hmrc_spi_income": hmrc, + } + + +def test_materialize_rules_engine_predictors_adds_declared_columns() -> None: + handler = uk_source_operation_handlers()["materialize_rules_engine_predictors"] + + result = handler(None, _operation(), _context(engine=StubRulesEngine())) + + assert isinstance(result, Frame) + assert result.table("person")["projected_income"].tolist() == [12.0, 34.0] + assert result.weights_for("household").values.tolist() == [1.0, 2.0] + + +def test_materialize_rules_engine_predictors_refuses_country_mismatch() -> None: + handler = uk_source_operation_handlers()["materialize_rules_engine_predictors"] + engine = StubRulesEngine() + engine.country = "us" + + with pytest.raises(SourceRuntimeError, match="does not match dataset country"): + handler(None, _operation(), _context(engine=engine)) diff --git a/packages/microcosm-build/tests/test_uk_source_stages.py b/packages/microcosm-build/tests/test_uk_source_stages.py new file mode 100644 index 00000000..e5c7ef2b --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_source_stages.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +import copy +import hashlib +import json +from pathlib import Path + +import pytest + +from microcosm.build.country_spec import country_stage_plan, load_country_spec +from microcosm.build.source_manifest import ( + FORBIDDEN_SOURCE_DEPENDENCIES, + SourceManifest, +) +from microcosm.frame import Frame + +ROOT = Path(__file__).resolve().parents[3] +UK_PACKAGE = ROOT / "packages/microcosm-build/src/microcosm/build/uk" +FROZEN_SOURCE_STAGES = UK_PACKAGE / "hmrc_income_source_stages.json" +CANONICAL_SOURCE_STAGES = UK_PACKAGE / "source_stages.json" +FROZEN_SOURCE_STAGES_SHA256 = ( + "c0341af7166ae3a85a3c1164e7d9e880c4b4aec122f1a8fa90c73b46c596e1ea" +) + + +def _load_json(path: Path) -> dict: + return json.loads(path.read_text(encoding="utf-8")) + + +def _identity(frame: Frame) -> Frame: + return frame + + +def _assert_no_forbidden_dependency(value: object) -> None: + text = json.dumps(value, sort_keys=True).lower() + for dependency in FORBIDDEN_SOURCE_DEPENDENCIES: + assert dependency not in text + + +def _expected_reviewed_source() -> str: + return ( + "PolicyEngine licensed UKDS mirror (private Hugging Face repository), " + "spi_2022_23.zip" + ) + + +def _rephrase_stage2_predictor_note(value: str) -> str: + return value.replace( + "policyengine-" + "uk-data frs_only.py", + "the incumbent UK data build's frs_only.py", + ) + + +class TestUKSourceStagesManifest: + def test_source_stages_json_loads_as_shared_manifest(self) -> None: + manifest = SourceManifest.from_mapping(_load_json(CANONICAL_SOURCE_STAGES)) + + assert manifest.country == "uk" + assert manifest.version == 1 + assert [stage.stage for stage in manifest.stages] == [ + "frs_hmrc_retained_leaves", + "hmrc_spi_income", + ] + + def test_country_spec_declares_two_uk_source_stages(self) -> None: + spec = load_country_spec("uk") + + assert spec.sources is not None + assert [stage.stage for stage in spec.sources.stages] == [ + "frs_hmrc_retained_leaves", + "hmrc_spi_income", + ] + + def test_copy_is_lockstep_with_frozen_original_except_citation_rewrites( + self, + ) -> None: + frozen = _load_json(FROZEN_SOURCE_STAGES) + canonical = _load_json(CANONICAL_SOURCE_STAGES) + frozen_stage = frozen["stages"][0] + stage1, stage2 = canonical["stages"] + + expected_operations = copy.deepcopy(frozen_stage["operations"]) + predictor_note = expected_operations[6]["reviewed_absent_predictors"][ + "other_investment_income" + ] + expected_operations[6]["reviewed_absent_predictors"][ + "other_investment_income" + ] = _rephrase_stage2_predictor_note(predictor_note) + + assert stage1["operations"] + stage2["operations"] == expected_operations + _assert_no_forbidden_dependency( + stage2["operations"][4]["reviewed_absent_predictors"][ + "other_investment_income" + ] + ) + + expected_artifacts = copy.deepcopy(frozen_stage["artifacts"]) + expected_artifacts[0]["reviewed_source"] = _expected_reviewed_source() + assert stage2["outputs"] == frozen_stage["outputs"] + assert stage2["grain"] == frozen_stage["grain"] + assert stage2["artifacts"] == expected_artifacts + _assert_no_forbidden_dependency(stage2["artifacts"]) + _assert_no_forbidden_dependency(stage2["notes"]) + + def test_frozen_original_bytes_are_pinned(self) -> None: + digest = hashlib.sha256(FROZEN_SOURCE_STAGES.read_bytes()).hexdigest() + + assert digest == FROZEN_SOURCE_STAGES_SHA256 + + def test_country_stage_plan_assembles_two_uk_stages(self) -> None: + spec = load_country_spec("uk") + plan = country_stage_plan( + spec, + { + "frs_hmrc_retained_leaves": _identity, + "hmrc_spi_income": _identity, + }, + ) + + assert [stage.name for stage in plan.stages] == [ + "frs_hmrc_retained_leaves", + "hmrc_spi_income", + ] + + @pytest.mark.parametrize( + "implementations, match", + [ + ({"frs_hmrc_retained_leaves": _identity}, "missing"), + ( + { + "frs_hmrc_retained_leaves": _identity, + "hmrc_spi_income": _identity, + "hmrc_spi_income_fallback": _identity, + }, + "Unknown stage implementation", + ), + ], + ) + def test_country_stage_plan_refuses_missing_or_unknown_uk_stage( + self, + implementations, + match: str, + ) -> None: + spec = load_country_spec("uk") + + with pytest.raises(ValueError, match=match): + country_stage_plan(spec, implementations) diff --git a/packages/microcosm-frame/pyproject.toml b/packages/microcosm-frame/pyproject.toml index 04fb27ae..94d93027 100644 --- a/packages/microcosm-frame/pyproject.toml +++ b/packages/microcosm-frame/pyproject.toml @@ -12,6 +12,7 @@ dependencies = [ [project.optional-dependencies] us = ["microunit>=0.1.0"] policyengine = ["policyengine-us>=1.745.0,<2", "microunit>=0.1.0"] +uk = ["policyengine-uk>=2.88"] # The Axiom adapter's PyPI-resolvable dependencies. The engine itself # (axiom-rules-engine + its dense native extension) is not on PyPI yet and # installs from a checkout; see microcosm/frame/adapters/axiom.py. diff --git a/packages/microcosm-frame/src/microcosm/frame/adapters/policyengine_uk.py b/packages/microcosm-frame/src/microcosm/frame/adapters/policyengine_uk.py new file mode 100644 index 00000000..b7763317 --- /dev/null +++ b/packages/microcosm-frame/src/microcosm/frame/adapters/policyengine_uk.py @@ -0,0 +1,160 @@ +"""PolicyEngine-UK adapter for the RulesEngine protocol. + +The PolicyEngine-UK import is deferred until a method needs engine metadata or +simulation. Importing this module therefore keeps ``microcosm-frame`` usable +without the UK extra. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from pathlib import Path +from typing import Any + +import numpy as np + +from microcosm.frame.bundle import Frame +from microcosm.frame.materialize import engine_tables +from microcosm.frame.rules import ExportContract +from microcosm.frame.schema import EntitySchema, VariableMetadata + +__all__ = ["PolicyEngineUKEngine", "UK_SCHEMA"] + +UK_SCHEMA = EntitySchema(group_entities=("benunit", "household")) +_PERSON_TABLE = "person" +_GROUP_TABLES = ("benunit", "household") +_DTYPE_KIND_BY_VALUE_TYPE: dict[type, str] = { + float: "float", + int: "int", + bool: "bool", + str: "str", +} +_PERIOD_BY_DEFINITION: dict[str, str] = {"year": "year", "month": "month"} + + +class PolicyEngineUKEngine: + """RulesEngine adapter backed by ``policyengine_uk``.""" + + country = "uk" + + def __init__(self) -> None: + self._system: Any = None + + def variable_metadata(self, name: str) -> VariableMetadata: + """Return entity, dtype kind, and period semantics for a UK variable.""" + + variable = self._variable(name) + return VariableMetadata( + name=name, + entity=variable.entity.key, + dtype=_DTYPE_KIND_BY_VALUE_TYPE.get(variable.value_type, "str"), + period=_PERIOD_BY_DEFINITION.get( + getattr(variable, "definition_period", "year"), "point" + ), + ) + + def variables(self) -> list[str]: + """Return the UK engine's input variable names.""" + + return sorted( + name + for name, variable in self._tax_benefit_system().variables.items() + if not _is_engine_computed(variable) + ) + + def entity_schema(self) -> EntitySchema: + """Return the UK national entity schema.""" + + return UK_SCHEMA + + def materialize( + self, + bundle: Frame, + variables: Sequence[str], + period: int | str, + ) -> dict[str, np.ndarray]: + """Compute ``variables`` for ``period`` with a UK Microsimulation.""" + + microsimulation_class = self._import_policyengine_uk().Microsimulation + dataset = self._build_dataset(bundle, period) + simulation = microsimulation_class(dataset=dataset) + results: dict[str, np.ndarray] = {} + for name in variables: + entity = self.variable_metadata(name).entity + values = np.asarray(simulation.calculate(name, period=period)) + expected = bundle.n(entity) + if values.shape != (expected,): + raise ValueError( + f"Materialized variable {name!r} has shape {values.shape} " + f"but entity {entity!r} has {expected} row(s)." + ) + results[name] = values + return results + + def export_contract(self) -> ExportContract: + """UK export contracts are owned by microcosm-build for this release.""" + + raise NotImplementedError( + "PolicyEngine-UK export contracts are not implemented in the frame " + "adapter yet; use the UK national build writer." + ) + + def write_dataset( + self, + bundle: Frame, + path: str | Path, + period: int | str, + ) -> None: + """UK dataset export remains on the national-build writer in E1.""" + + raise NotImplementedError( + "PolicyEngine-UK dataset export is not implemented in the frame " + "adapter yet; use microcosm.build.uk_runtime.national_build." + "write_uk_national_frame." + ) + + def _import_policyengine_uk(self) -> Any: + try: + import policyengine_uk + except ImportError as exc: + raise ImportError( + "The PolicyEngine-UK adapter requires the 'policyengine-uk' " + "package. Install it with 'microcosm-frame[uk]'." + ) from exc + return policyengine_uk + + def _tax_benefit_system(self) -> Any: + if self._system is None: + self._system = self._import_policyengine_uk().CountryTaxBenefitSystem() + return self._system + + def _variable(self, name: str) -> Any: + variables = self._tax_benefit_system().variables + if name not in variables: + raise ValueError(f"Unknown PolicyEngine-UK variable {name!r}.") + return variables[name] + + def _build_dataset(self, bundle: Frame, period: int | str) -> Any: + from policyengine_uk.data import UKSingleYearDataset + + expected = (_PERSON_TABLE, *_GROUP_TABLES) + if set(bundle.entities) != set(expected): + raise ValueError( + f"PolicyEngine-UK adapter requires the UK entities " + f"{list(expected)}; bundle has {list(bundle.entities)}." + ) + tables = engine_tables(bundle, weighted_entities=("household",)) + return UKSingleYearDataset( + person=tables["person"].copy(), + benunit=tables["benunit"].copy(), + household=tables["household"].copy(), + time_period=str(period), + ) + + +def _is_engine_computed(variable: Any) -> bool: + if getattr(variable, "adds", None) or getattr(variable, "subtracts", None): + return True + if getattr(variable, "formula", None) is not None: + return True + return bool(getattr(variable, "formulas", None)) diff --git a/packages/microcosm-frame/src/microcosm/frame/adapters/policyengine_us.py b/packages/microcosm-frame/src/microcosm/frame/adapters/policyengine_us.py index 236b6d7e..95406be0 100644 --- a/packages/microcosm-frame/src/microcosm/frame/adapters/policyengine_us.py +++ b/packages/microcosm-frame/src/microcosm/frame/adapters/policyengine_us.py @@ -505,6 +505,8 @@ class PolicyEngineUSEngine: ``policyengine_us``. """ + country = "us" + def __init__( self, contract: ExportContract | None = None, diff --git a/packages/microcosm-frame/src/microcosm/frame/rules.py b/packages/microcosm-frame/src/microcosm/frame/rules.py index bc75bf20..34a3e1ca 100644 --- a/packages/microcosm-frame/src/microcosm/frame/rules.py +++ b/packages/microcosm-frame/src/microcosm/frame/rules.py @@ -18,11 +18,17 @@ from typing import Any, Protocol, runtime_checkable import numpy as np +import pandas as pd from microcosm.frame.bundle import Frame from microcosm.frame.schema import EntitySchema, VariableMetadata -__all__ = ["RulesEngine", "ExportContract"] +__all__ = [ + "RulesEngine", + "ExportContract", + "assert_rules_engine_country", + "materialize_rules_engine_predictors", +] @runtime_checkable @@ -160,6 +166,87 @@ def from_path(cls, path: str | Path) -> "ExportContract": ) +def assert_rules_engine_country(engine: RulesEngine, country: str) -> None: + """Require a rules adapter to declare the dataset country it serves.""" + + engine_country = getattr(engine, "country", None) + if not isinstance(engine_country, str) or not engine_country: + raise ValueError("Rules engine adapter must declare a non-empty country.") + if engine_country != country: + raise ValueError( + f"Rules engine country {engine_country!r} does not match dataset " + f"country {country!r}." + ) + + +def materialize_rules_engine_predictors( + bundle: Frame, + *, + variables: Sequence[str], + period: int | str, + engine: RulesEngine, + country: str | None = None, +) -> Frame: + """Return ``bundle`` with rules-engine predictors materialized as columns.""" + + if country is not None: + assert_rules_engine_country(engine, country) + requested = tuple(variables) + if not requested: + raise ValueError("materialize_rules_engine_predictors requires variables.") + existing: list[str] = [] + for variable in requested: + try: + bundle.column_entity(variable) + except ValueError: + continue + existing.append(variable) + if existing: + raise ValueError( + "Cannot materialize rules-engine predictor(s) already present on " + f"the frame: {existing}." + ) + + materialized = engine.materialize(bundle, requested, period) + tables = {entity: bundle.table(entity).copy() for entity in bundle.entities} + for variable in requested: + if variable not in materialized: + raise ValueError( + f"Rules engine did not return materialized predictor {variable!r}." + ) + metadata = engine.variable_metadata(variable) + if metadata.entity not in tables: + raise ValueError( + f"Materialized predictor {variable!r} belongs to entity " + f"{metadata.entity!r}, which is not in the frame." + ) + values = np.asarray(materialized[variable]) + expected = bundle.n(metadata.entity) + if values.shape != (expected,): + raise ValueError( + f"Materialized predictor {variable!r} has shape {values.shape} " + f"but entity {metadata.entity!r} has {expected} row(s)." + ) + tables[metadata.entity][variable] = pd.Series( + values, + index=tables[metadata.entity].index, + ) + + for link in bundle.links: + tables[link] = bundle.link(link).copy() + weights = { + entity: bundle.weights_for(entity) for entity in bundle.weighted_entities + } + return Frame( + tables, + bundle.schema, + weights, + strata=bundle.strata.copy(), + mass_log=bundle.mass_log, + metadata=bundle.metadata, + ) + + def _as_str_tuple(values: Any) -> tuple[str, ...]: """Coerce a JSON list (or any iterable) into a tuple of strings.""" if values is None: diff --git a/packages/microcosm-frame/tests/test_policyengine_uk_adapter.py b/packages/microcosm-frame/tests/test_policyengine_uk_adapter.py new file mode 100644 index 00000000..ce376da4 --- /dev/null +++ b/packages/microcosm-frame/tests/test_policyengine_uk_adapter.py @@ -0,0 +1,28 @@ +"""PolicyEngine-UK adapter import/protocol behavior.""" + +from __future__ import annotations + +import pytest + +from microcosm.frame import RulesEngine +from microcosm.frame.adapters.policyengine_uk import ( + UK_SCHEMA, + PolicyEngineUKEngine, +) + + +def test_policyengine_uk_adapter_satisfies_rules_protocol_without_importing_engine() -> ( + None +): + adapter = PolicyEngineUKEngine() + + assert isinstance(adapter, RulesEngine) + assert adapter.country == "uk" + assert adapter.entity_schema() == UK_SCHEMA + + +def test_policyengine_uk_adapter_export_side_is_not_implemented() -> None: + adapter = PolicyEngineUKEngine() + + with pytest.raises(NotImplementedError, match="write_uk_national_frame"): + adapter.write_dataset(object(), "unused.h5", period=2023) # type: ignore[arg-type] diff --git a/tools/build_uk_national_dataset.py b/tools/build_uk_national_dataset.py index c878c79c..b723f99e 100644 --- a/tools/build_uk_national_dataset.py +++ b/tools/build_uk_national_dataset.py @@ -15,6 +15,7 @@ import pandas as pd +from microcosm.build.country_spec import country_stage_plan, load_country_spec from microcosm.build.gate_battery import GateBatteryBlockedError from microcosm.build.logbook import canonical_json_bytes from microcosm.build.logbook_adoption import ( @@ -42,10 +43,7 @@ UKHMRCIncomeStageTransform, verify_certified_uk_candidate, ) -from microcosm.build.uk_runtime.national_build import ( - UKNationalStage, - build_uk_national_dataset, -) +from microcosm.build.uk_runtime.national_build import build_uk_national_dataset from microcosm.build.uk_runtime.national_frame import ( uk_household_weight_kind, uk_time_period, @@ -55,6 +53,7 @@ UK_SAMPLE_SEED_DEFAULT, ) from microcosm.build.uk_runtime.release_identity import UK_RELEASE_TIERS +from microcosm.build.uk_runtime.source_runtime import uk_stage_implementations from microcosm.build.uk_runtime.terminal_gates import ( uk_default_degenerate_reviewed_exclusions, ) @@ -905,16 +904,17 @@ def _main_recording( calibration_diagnostics_sha256=args.calibration_diagnostics_sha256, reviewed_degenerate_exclusions=reviewed_degenerate_exclusions, stages=( - UKNationalStage( - name="frs_hmrc_retained_leaves", - transform=retained_leaves_transform, - ), - UKNationalStage( - name="hmrc_spi_income", - transform=hmrc_transform, - ), + *country_stage_plan( + load_country_spec("uk"), + uk_stage_implementations( + retained_leaves_transform=retained_leaves_transform, + hmrc_income_transform=hmrc_transform, + ), + ).stages, # Runs after the SPI restoration so the taxable-income proxy - # sees the restored income surface. + # sees the restored income surface. Declared today in the + # bespoke uk/cgt_source_stages.json; absorbing it into the + # canonical source_stages.json is WS-E follow-up work. uk_capital_gains_imputation_stage(args.cgt_ods), ), **gate_path_argument, From 0ef4259fe0c608515068d9b328484474e2f8c53d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Fri, 14 Aug 2026 15:15:11 +0200 Subject: [PATCH 2/4] Adversarial-review fix: UK predictor handler fails closed on country MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An absent country extra no longer skips the engine assertion — the UK handler map serves country "uk" by construction, so the handler now defaults the dataset country to "uk" and refuses a runtime context declaring any other country. Regression tests cover the wrong-country engine with no extra and the non-UK context. Co-Authored-By: Claude Fable 5 --- .../build/uk_runtime/source_runtime.py | 22 +++++++----- .../tests/test_uk_source_runtime.py | 34 ++++++++++++++++--- 2 files changed, 43 insertions(+), 13 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/source_runtime.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/source_runtime.py index eb167f9c..14b44ea5 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/source_runtime.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/source_runtime.py @@ -60,7 +60,7 @@ def materialize_uk_rules_engine_predictors_from_manifest( ) frame = _extra(context, "frame", Frame) engine = _extra(context, "rules_engine", object) - country = _optional_country(context) + country = _require_uk_country(context) period = context.config.extra.get("period", context.config.target_year) if period is None: raise SourceRuntimeError( @@ -101,13 +101,19 @@ def _extra( return value -def _optional_country(context: SourceRuntimeContext) -> str | None: - country = context.config.extra.get("country") - if country is None: - return None - if not isinstance(country, str) or not country: +def _require_uk_country(context: SourceRuntimeContext) -> str: + """Fail closed: this is the UK handler map, so the dataset country is UK. + + An absent ``country`` extra must not skip the engine assertion (the + shared materializer only asserts when a country is supplied), and a + context claiming another country must never reach a UK handler. + """ + + country = context.config.extra.get("country", "uk") + if country != "uk": raise SourceRuntimeError( - "materialize_rules_engine_predictors context country must be a " - "non-empty string." + "materialize_rules_engine_predictors ran through the UK handler " + f"map but the runtime context declares country {country!r}; UK " + "handlers only serve country 'uk'." ) return country diff --git a/packages/microcosm-build/tests/test_uk_source_runtime.py b/packages/microcosm-build/tests/test_uk_source_runtime.py index e30c4fac..25bb2eb4 100644 --- a/packages/microcosm-build/tests/test_uk_source_runtime.py +++ b/packages/microcosm-build/tests/test_uk_source_runtime.py @@ -94,12 +94,12 @@ def _operation() -> SourceOperationSpec: ) -def _context(*, engine: object, country: str = "uk") -> SourceRuntimeContext: +def _context(*, engine: object, country: str | None = "uk") -> SourceRuntimeContext: + extra: dict[str, object] = {"frame": _frame(), "rules_engine": engine} + if country is not None: + extra["country"] = country return SourceRuntimeContext( - config=SourceRuntimeConfig( - target_year=2023, - extra={"frame": _frame(), "rules_engine": engine, "country": country}, - ), + config=SourceRuntimeConfig(target_year=2023, extra=extra), tables={}, ) @@ -137,3 +137,27 @@ def test_materialize_rules_engine_predictors_refuses_country_mismatch() -> None: with pytest.raises(SourceRuntimeError, match="does not match dataset country"): handler(None, _operation(), _context(engine=engine)) + + +def test_materialize_rules_engine_predictors_asserts_country_without_extra() -> None: + # Regression for the adversarial-review bypass: an absent ``country`` + # extra must not skip the engine assertion — the UK handler map serves + # country "uk" by construction, so a wrong-country engine is refused + # even when the caller forgets the optional extra. + handler = uk_source_operation_handlers()["materialize_rules_engine_predictors"] + engine = StubRulesEngine() + engine.country = "us" + + with pytest.raises(SourceRuntimeError, match="does not match dataset country"): + handler(None, _operation(), _context(engine=engine, country=None)) + + +def test_materialize_rules_engine_predictors_refuses_non_uk_context() -> None: + handler = uk_source_operation_handlers()["materialize_rules_engine_predictors"] + + with pytest.raises(SourceRuntimeError, match="only serve country 'uk'"): + handler( + None, + _operation(), + _context(engine=StubRulesEngine(), country="us"), + ) From 440fff26b7a809ec72343306e8aa12e0a3477336 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Fri, 14 Aug 2026 15:38:11 +0200 Subject: [PATCH 3/4] Licensed-data acceptance fix: declare the persisted SPI state-pension column MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The frozen manifest's outputs listed the SPI concept "state_pension", but the stage writes the auxiliary column hmrc_spi_state_pension_income — the model input is formula-owned and never a frame column here. Outputs became load-bearing when country_stage_plan compiled them into StagePlan.produces, so the full-rung licensed-data build refused to stage. The copy now declares the persisted truth; operation payloads keep the concept name; the lockstep test records the correction as a declared exception. Co-Authored-By: Claude Fable 5 --- .../src/microcosm/build/uk/source_stages.json | 2 +- .../microcosm-build/tests/test_uk_source_stages.py | 14 +++++++++++++- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/uk/source_stages.json b/packages/microcosm-build/src/microcosm/build/uk/source_stages.json index 2a8e2654..d7736fba 100644 --- a/packages/microcosm-build/src/microcosm/build/uk/source_stages.json +++ b/packages/microcosm-build/src/microcosm/build/uk/source_stages.json @@ -725,7 +725,7 @@ "outputs": [ "employment_income", "self_employment_income", - "state_pension", + "hmrc_spi_state_pension_income", "private_pension_income", "property_income", "savings_interest_income", diff --git a/packages/microcosm-build/tests/test_uk_source_stages.py b/packages/microcosm-build/tests/test_uk_source_stages.py index e5c7ef2b..4cb9d27e 100644 --- a/packages/microcosm-build/tests/test_uk_source_stages.py +++ b/packages/microcosm-build/tests/test_uk_source_stages.py @@ -96,7 +96,19 @@ def test_copy_is_lockstep_with_frozen_original_except_citation_rewrites( expected_artifacts = copy.deepcopy(frozen_stage["artifacts"]) expected_artifacts[0]["reviewed_source"] = _expected_reviewed_source() - assert stage2["outputs"] == frozen_stage["outputs"] + # Declared output-name correction (licensed-data acceptance finding): + # the frozen original listed the SPI concept "state_pension", but the + # stage writes the auxiliary column SPI_HMRC_STATE_PENSION_INCOME_COLUMN + # ("hmrc_spi_state_pension_income") — the model input state_pension is + # formula-owned and never a frame column here. Outputs became + # load-bearing when country_stage_plan compiled them into + # StagePlan.produces, so the copy declares the persisted truth. The + # operation payloads keep the concept name unchanged. + expected_outputs = [ + "hmrc_spi_state_pension_income" if name == "state_pension" else name + for name in frozen_stage["outputs"] + ] + assert stage2["outputs"] == expected_outputs assert stage2["grain"] == frozen_stage["grain"] assert stage2["artifacts"] == expected_artifacts _assert_no_forbidden_dependency(stage2["artifacts"]) From 42128ba649e7107ef633c511fd0d5cc094c18fe2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Mon, 17 Aug 2026 10:17:16 +0200 Subject: [PATCH 4/4] Review response: pin declared outputs to runtime written-column constants Outputs are load-bearing through StagePlan.produces, and the licensed acceptance caught a declared concept that no stage persists. This pins stage-1 outputs exactly to the retained-leaf constants and requires every stage-2 output to be backed by a named written-column constant, so the class cannot recur without detection. Cross-manifest sweep: #700. Co-Authored-By: Claude Fable 5 --- .../tests/test_uk_source_stages.py | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/packages/microcosm-build/tests/test_uk_source_stages.py b/packages/microcosm-build/tests/test_uk_source_stages.py index 4cb9d27e..4ba6ee7f 100644 --- a/packages/microcosm-build/tests/test_uk_source_stages.py +++ b/packages/microcosm-build/tests/test_uk_source_stages.py @@ -157,3 +157,52 @@ def test_country_stage_plan_refuses_missing_or_unknown_uk_stage( with pytest.raises(ValueError, match=match): country_stage_plan(spec, implementations) + + +class TestDeclaredOutputsAreWrittenColumns: + """Declared outputs must name columns the stages actually write. + + Outputs are load-bearing (``country_stage_plan`` compiles them into + ``StagePlan.produces``), and the licensed-data acceptance for this + migration caught a declared output that was an SPI *concept* rather + than a persisted column — harmless while nothing read the field, + refused at full rung once it did. This pins every declared output to + a named runtime written-column constant so the class cannot recur + without a licensed build to find it (microcosm#690 review). + """ + + def test_stage1_outputs_are_exactly_the_retained_leaf_columns(self) -> None: + from microcosm.build.uk_runtime.frs_hmrc_leaves import ( + FRS_HMRC_RETAINED_LEAF_COLUMNS, + ) + + spec = load_country_spec("uk") + stages = {stage.stage: stage for stage in spec.sources.stages} + stage1 = stages["frs_hmrc_retained_leaves"] + assert stage1.outputs == tuple(FRS_HMRC_RETAINED_LEAF_COLUMNS) + + def test_stage2_outputs_are_backed_by_runtime_written_columns(self) -> None: + from microcosm.build.uk_runtime.spi_support import ( + SPI_HMRC_DERIVED_AUXILIARY_COLUMNS, + SPI_HMRC_QRF_AUXILIARY_COLUMNS, + SPI_INCOME_IMPUTATION_COLUMNS, + ) + + spec = load_country_spec("uk") + stages = {stage.stage: stage for stage in spec.sources.stages} + stage2 = stages["hmrc_spi_income"] + written = ( + set(SPI_INCOME_IMPUTATION_COLUMNS) + | set(SPI_HMRC_QRF_AUXILIARY_COLUMNS) + | set(SPI_HMRC_DERIVED_AUXILIARY_COLUMNS) + ) + # The narrow PAY+EPB+TAXTERM employment input is written on SPI rows + # by the stage even though the QRF output surface excludes it. + written.add("employment_income") + unbacked = [name for name in stage2.outputs if name not in written] + assert unbacked == [], ( + "Declared outputs with no named runtime written-column constant " + f"backing them: {unbacked}. Either the manifest declares a " + "concept instead of a persisted column, or the runtime constant " + "moved without the manifest following." + )