diff --git a/changelog.d/3862.breaking.md b/changelog.d/3862.breaking.md new file mode 100644 index 000000000..4a719cf50 --- /dev/null +++ b/changelog.d/3862.breaking.md @@ -0,0 +1 @@ +Replace large SPM diagnostic provenance payloads and duplicate completed-result SPM configuration with strict compact v2 receipts for household, society-wide, report, and Stage 12 artifact results. Older cached or stored SPM output is recalculated instead of translated, and runtime version values are required. diff --git a/docs/canonical-spm.md b/docs/canonical-spm.md index 042e3be71..590f9fe07 100644 --- a/docs/canonical-spm.md +++ b/docs/canonical-spm.md @@ -155,16 +155,44 @@ storage; no additional schema migration is required. selected policy. No independent simulation-level override is supported. Top-level `spm` on simulation POST/PATCH is rejected instead of silently ignored. -Successful canonical calculations add `spm_config` and `spm_provenance` beside -`result`. A receipt describes the measurement the simulation was constructed -with, not a guarantee that every SPM-dependent variable produced a value: a +Successful canonical calculations add only `spm_provenance` beside `result`. +Its forecast, scenario, geography, county-vintage and as-of fields are the +complete resolved calculation configuration; completed results do not repeat +that configuration in a sibling `spm_config` field. A receipt describes the +measurement the simulation was constructed with, not a guarantee that every +SPM-dependent variable produced a value: a calculation that never chose a measurement still carries the inherited one's receipt beside its null cells. Read the values to learn which of them a -measurement produced. Provenance comes from the actual simulation and includes artifact, -scenario, years, geography, composition/storage methods and runtime versions. -It remains in JSON form through stored replay and cache hits. A tax-only receipt -may have empty `years` and `geographies` because no SPM measurement was requested. -Clients saving simulation outputs must retain this full response envelope. +measurement produced. Provenance comes from the actual simulation and uses the +required compact `canonical-spm-provenance-v2` schema: + +```json +{ + "schema_version": "canonical-spm-provenance-v2", + "forecast_id": "canonical-spm-forecast", + "forecast_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "scenario": "baseline", + "geography_kind": "national", + "geography_id": null, + "county_vintage": "2020", + "as_of": null, + "years": ["2026"], + "runtime_versions": { + "policyengine": "6.2.1", + "policyengine-core": "3.32.10", + "policyengine-us": "2.2.1", + "spm-calculator": "1.0.0" + } +} +``` + +All fields are required. The four runtime-version values are non-null strings. +`years` is sorted and unique, and it is empty only when no SPM dependency ran. +Large diagnostic, mapping, request, effective-setting and source +objects from the country package are intentionally excluded from HTTP and +cache payloads. The compact receipt remains in JSON form through stored replay +and cache hits. Clients saving simulation outputs must retain this response +envelope. `POST /us/simulation` takes `population_id`, `population_type` (`household` or `geography`) and integer `policy_id`. It creates a pending record (HTTP 201) or @@ -181,12 +209,13 @@ country, certification, or whether the household has saved `spm`. `PATCH /us/simulation` identifies the record with body `id` and accepts `status` (`pending`, `complete` or `error`), `output` and `error_message`. At least one update field must be non-null, and `complete` requires non-null `output`. Store -the full calculation envelope inside `output`, including its `spm_config` and +the full calculation envelope inside `output`, including its `spm_provenance`. A JSON-encoded output string is also accepted. This endpoint -stores the supplied output; it does not validate SPM receipt integrity. Null -update fields are ignored. The legacy `api_version` input is ignored; writes -record the installed country model version. Top-level `spm`, even null, returns -HTTP 400 `SPM_SETTINGS_UNSUPPORTED` on POST and PATCH. +validates the one compact SPM receipt strictly before storage. Completed output +that includes the former sibling `spm_config`, an old receipt or a rich receipt +is rejected rather than translated. Null update fields are ignored. The legacy `api_version` input +is ignored; writes record the installed country model version. Top-level `spm`, +even null, returns HTTP 400 `SPM_SETTINGS_UNSUPPORTED` on POST and PATCH. POST, PATCH and `GET /us/simulation/{id}` return the simulation record inside `result`. Non-string JSON `output` and `simulation_spec_json` are returned as @@ -201,9 +230,10 @@ Flask and the native specification route. HTTP response cache identity includes the normalized selection and model/bundle versions, and validates certification before reading the cache. The stored -calculated-household cache uses schema version 2, includes the selection in -identity, and stores settings/provenance atomically with the result and its -warnings. Missing or mismatched canonical receipts are cache misses. +calculated-household cache uses schema version 4, includes the selection in +identity, and stores provenance atomically with the result and its +warnings. Missing, old, rich or mismatched canonical receipts are cache misses +and are recalculated rather than translated. ## Errors @@ -261,16 +291,28 @@ from `/versions`, including agreement with the bundle's independently pinned artifact hash. An old or uncertified worker returns `SPM_CONFIGURATION_UNAVAILABLE`. Normalized settings are sent to both annual and budget-window jobs and participate -in economy cache identity. Results carry `spm_config` and `spm_provenance` with -`baseline` and `reform` receipt lists (one per executed regional segment). Each -budget-window annual row carries these fields; annual results expose them inside -`result`, and budget-window results inside `result.annualImpacts`. Stored results require matching -settings and valid receipts. Typed SPM input errors remain structured 400s +in economy cache identity. Results carry only `spm_provenance` using +`canonical-spm-comparison-v2`. `baseline` and `reform` each contain one compact +`receipt` plus an `execution_count`; the receipt collapses identical child +execution receipts without transporting a repeated list. Baseline and reform +receipts must be identical, and their receipt fields are the sole resolved SPM +configuration. Each budget-window annual row carries this field; +annual results expose it inside `result`, and budget-window results inside +`result.annualImpacts`. Stored results require matching settings and valid +receipts. Economy cache identity includes the comparison schema version, so old +entries are not selected and are recomputed rather than translated. Typed SPM input errors remain structured 400s through asynchronous submission and polling. Poll-time typed failures retain their code and message for the existing cache lifetime. Later reads replay that failure after API service restarts without polling or resubmitting the failed job. Canonical cache identity and runtime-bundle refresh rules still apply. +Completed US economy report outputs pass the same compact comparison validation +before `report_outputs` or `report_output_runs` persistence. Their report cache +identity uses schema version 3. Reads of earlier report records resolve +to a new pending record with no copied output, preventing another client from +receiving a previously stored rich receipt. Household and non-US report output +shapes retain their existing behavior. + See the [canonical SPM worker PR](https://github.com/PolicyEngine/policyengine-sim-api/pull/677) for the implemented worker paths and remaining coordinated release gates. @@ -278,5 +320,5 @@ for the implemented worker paths and remaining coordinated release gates. Partial selections preserve omitted fields in JSON; omissions inherit the certified bundle defaults. The selection schema deliberately supplies no wire defaults, so generated clients preserve that distinction. Resolved settings freeze all six -fields. Responses may omit null geography_id/as_of values, but must include every -non-null resolved field; receipt replay never substitutes current defaults. +fields. Responses include all six fields, including explicit null values for +`geography_id` or `as_of`; receipt replay never substitutes current defaults. diff --git a/docs/migration/stage-12-worker-contract.md b/docs/migration/stage-12-worker-contract.md index dbb7a8aa2..1e830a360 100644 --- a/docs/migration/stage-12-worker-contract.md +++ b/docs/migration/stage-12-worker-contract.md @@ -171,6 +171,14 @@ SHA-256 content digests, schema versions, row-identity metadata, and bounded provenance. When an option such as SPM produces a detached calculation receipt, the simulation descriptor and Parquet metadata retain it so the coordinator can validate the receipt before reproducing the existing aggregate response. +`SimulationArtifactDescriptor.calculation_provenance` is either null or the +exact typed `canonical-spm-provenance-v2` receipt. The receipt's forecast, +scenario, geography, county-vintage and as-of fields are the complete resolved +configuration; the descriptor does not transport a duplicate `spm_config`. +It rejects partial, rich, legacy and dual-field receipt shapes rather than +preserving or normalizing them. +`AggregateReportArtifactPayload.result` applies the same rule to comparison +provenance and rejects a sibling `spm_config` before artifact storage. Only the Simulation Entrypoint, the versioned worker application, the bucket's object-lifecycle process, and explicitly authorized operators may access the diff --git a/policyengine_api/constants.py b/policyengine_api/constants.py index 683fb4aa9..7056574cb 100644 --- a/policyengine_api/constants.py +++ b/policyengine_api/constants.py @@ -176,8 +176,8 @@ def get_bundle_default_dataset_option(country_id: str) -> dict: ) or _resolve_distribution_version(_dist_versions, "policyengine-core", "policyengine") RUNTIME_CACHE_SCHEMA_VERSIONS = { - "economy_impact": 2, - "report_output": 1, + "economy_impact": 3, + "report_output": 3, } diff --git a/policyengine_api/openapi_spec.yaml b/policyengine_api/openapi_spec.yaml index 4e0d0f8b2..3155fe341 100644 --- a/policyengine_api/openapi_spec.yaml +++ b/policyengine_api/openapi_spec.yaml @@ -439,8 +439,6 @@ paths: nullable: true result: type: object - spm_config: - $ref: '#/components/schemas/SPMSelection' spm_provenance: $ref: '#/components/schemas/SPMProvenance' 404: @@ -531,8 +529,6 @@ paths: nullable: true result: type: object - spm_config: - $ref: '#/components/schemas/SPMSelection' spm_provenance: $ref: '#/components/schemas/SPMProvenance' 500: @@ -623,9 +619,9 @@ paths: non-null output. Null update fields are ignored, and omitted fields retain their values. The installed country model version and current run state are updated with the record. Preserve the full household calculation - envelope, including spm_config and spm_provenance, inside output. Its SPM - selection belongs to the linked household; this operation does not select - a measurement or validate stored receipt integrity. Top-level spm is + envelope, including spm_provenance, inside output. Its SPM selection is + derived from that receipt and belongs to the linked household; this + operation does not select a measurement. Top-level spm is rejected, including null, with HTTP 400 SPM_SETTINGS_UNSUPPORTED. parameters: - name: country_id @@ -658,7 +654,7 @@ paths: Object and array outputs and simulation_spec_json use the legacy JSON-encoded string representation, or null when absent. Stored scalar strings are returned unchanged. Decode a saved household output envelope - to read its spm_config and spm_provenance. Run identifiers and + to read its spm_provenance. Run identifiers and specification metadata may be null on historical records. parameters: - name: country_id @@ -760,7 +756,10 @@ paths: patch: summary: Update a report output operationId: update_report_output - description: Update a report output status, result, or error message. + description: >- + Update a report output status, result, or error message. Completed US + economy outputs require compact v2 comparison provenance. Each receipt + contains the complete resolved configuration; a sibling spm_config is rejected. parameters: - name: country_id in: path @@ -880,7 +879,7 @@ paths: # Query parameters are generated from policyengine_api.query_parameters. responses: 200: - description: Annual economic impact, progress, or persisted error state. Canonical completed results include the selected SPM configuration and actual baseline/reform receipts. + description: Annual economic impact, progress, or persisted error state. Canonical completed results include actual baseline/reform receipts containing the selected SPM configuration. content: application/json: schema: @@ -896,10 +895,8 @@ paths: type: object nullable: true properties: - spm_config: - $ref: '#/components/schemas/SPMSelection' spm_provenance: - $ref: '#/components/schemas/SPMWorkerProvenance' + $ref: '#/components/schemas/SPMComparisonProvenance' 400: $ref: '#/components/responses/SPMValidationError' 404: @@ -969,10 +966,8 @@ paths: properties: year: type: string - spm_config: - $ref: '#/components/schemas/SPMSelection' spm_provenance: - $ref: '#/components/schemas/SPMWorkerProvenance' + $ref: '#/components/schemas/SPMComparisonProvenance' progress: type: integer nullable: true @@ -1188,7 +1183,7 @@ components: description: >- Non-string JSON output is JSON-encoded, or null when absent. Stored scalar strings are returned unchanged. Decode a saved household - calculation envelope once to recover spm_config and spm_provenance + calculation envelope once to recover spm_provenance beside its result, not beside this record. x-household-envelope-schema: $ref: '#/components/schemas/StoredHouseholdOutputEnvelope' @@ -1217,7 +1212,7 @@ components: type: object description: >- Household calculation envelope saved as simulation output. Keep result - and both SPM fields together. Legacy calculations omit SPM fields; this + and SPM provenance together. Legacy calculations omit SPM provenance; this descriptive schema does not add validation to the persistence endpoint. properties: status: @@ -1228,8 +1223,6 @@ components: result: type: object nullable: true - spm_config: - $ref: '#/components/schemas/SPMSelection' spm_provenance: $ref: '#/components/schemas/SPMProvenance' SimulationError: @@ -1270,25 +1263,16 @@ components: description: JSON Schema for SPMSelection, preserving omitted fields. defaults: $ref: '#/components/schemas/SPMSelection' - SPMWorkerProvenance: - type: object - additionalProperties: false - required: [baseline, reform] - properties: - baseline: - type: array - minItems: 1 - items: - $ref: '#/components/schemas/SPMProvenance' - reform: - type: array - minItems: 1 - items: - $ref: '#/components/schemas/SPMProvenance' SPMSelection: description: Generated from policyengine_api.spm.SPMSelection. + SPMRuntimeVersions: + description: Generated from policyengine_api.spm.SPMRuntimeVersions. SPMProvenance: description: Generated from policyengine_api.spm.SPMProvenance. + SPMExecutionProvenance: + description: Generated from policyengine_api.spm.SPMExecutionProvenance. + SPMComparisonProvenance: + description: Generated from policyengine_api.spm.SPMComparisonProvenance. responses: SPMValidationError: description: >- diff --git a/policyengine_api/routes/household_routes.py b/policyengine_api/routes/household_routes.py index 5f2c65dd9..300f2d9d3 100644 --- a/policyengine_api/routes/household_routes.py +++ b/policyengine_api/routes/household_routes.py @@ -137,10 +137,8 @@ def _calculation_response(calculation) -> dict: result = dict(status="ok", message=None, result=calculation.household) if calculation.warnings: result["warnings"] = list(calculation.warnings) - for field in ("spm_config", "spm_provenance"): - value = getattr(calculation, field, None) - if value is not None: - result[field] = value + if calculation.spm_provenance is not None: + result["spm_provenance"] = calculation.spm_provenance return result @@ -231,7 +229,7 @@ def _calculation_cache_key(*args, **kwargs): country_id = request.view_args["country_id"] return hash_object( { - "schema": 2, + "schema": 3, "path": request.full_path, "payload": request.get_json(), "spm": g.spm, diff --git a/policyengine_api/runtime_cache/household_calculations.py b/policyengine_api/runtime_cache/household_calculations.py index 363f9d49e..4ab7ceaf7 100644 --- a/policyengine_api/runtime_cache/household_calculations.py +++ b/policyengine_api/runtime_cache/household_calculations.py @@ -10,10 +10,13 @@ CacheNamespace, RecoverableJSONCache, ) -from policyengine_api.spm import SPMProvenance, resolved_spm_settings +from policyengine_api.spm import ( + resolved_spm_configuration, + validate_spm_calculation_provenance, +) -HOUSEHOLD_CALCULATION_SCHEMA_VERSION = 2 +HOUSEHOLD_CALCULATION_SCHEMA_VERSION = 4 HOUSEHOLD_CALCULATION_TTL_SECONDS = 86_400 @@ -33,7 +36,6 @@ class HouseholdCalculationIdentity: class CachedHouseholdCalculation: household: dict[str, Any] warnings: tuple[str, ...] = () - spm_config: dict[str, Any] | None = None spm_provenance: dict[str, Any] | None = None @@ -65,33 +67,28 @@ def get( return None if not all(isinstance(warning, str) for warning in warnings): return None - spm_config = payload.get("spm_config") + if "spm_config" in payload: + return None spm_provenance = payload.get("spm_provenance") - if any( - value is not None and not isinstance(value, dict) - for value in (spm_config, spm_provenance) - ): + if spm_provenance is not None and not isinstance(spm_provenance, dict): return None - if identity.spm is not None: - # A stored receipt may omit its null values, so compare resolved - # settings rather than raw JSON. Requiring exact equality would miss - # the cache on every replay of a canonical calculation. - if resolved_spm_settings(spm_config) != identity.spm: + if identity.spm is None: + if spm_provenance is not None: return None + else: try: - receipt = SPMProvenance.model_validate(spm_provenance) - except ValidationError: + receipt = validate_spm_calculation_provenance(spm_provenance) + except (ValidationError, ValueError): return None if ( - receipt.forecast_sha256 != identity.spm.get("forecast_content_sha256") - or receipt.scenario != identity.spm.get("scenario") - or receipt.geography_kind != identity.spm.get("geography_kind") + resolved_spm_configuration(receipt).model_dump(mode="json") + != identity.spm ): return None + spm_provenance = receipt.model_dump(mode="json", by_alias=True) return CachedHouseholdCalculation( household=household, warnings=tuple(warnings), - spm_config=spm_config, spm_provenance=spm_provenance, ) @@ -100,4 +97,16 @@ def set( identity: HouseholdCalculationIdentity, value: CachedHouseholdCalculation, ) -> bool: - return self._cache.set(asdict(identity), asdict(value)) + if identity.spm is None: + if value.spm_provenance is not None: + return False + return self._cache.set(asdict(identity), asdict(value)) + try: + receipt = validate_spm_calculation_provenance(value.spm_provenance) + except (ValidationError, ValueError): + return False + if resolved_spm_configuration(receipt).model_dump(mode="json") != identity.spm: + return False + payload = asdict(value) + payload["spm_provenance"] = receipt.model_dump(mode="json", by_alias=True) + return self._cache.set(asdict(identity), payload) diff --git a/policyengine_api/runtime_cache/reform_impacts.py b/policyengine_api/runtime_cache/reform_impacts.py index 571e5b24c..8a7ed5dc8 100644 --- a/policyengine_api/runtime_cache/reform_impacts.py +++ b/policyengine_api/runtime_cache/reform_impacts.py @@ -19,8 +19,7 @@ record_cache_event, ) - -REFORM_IMPACT_SCHEMA_VERSION = 1 +REFORM_IMPACT_SCHEMA_VERSION = 3 REFORM_IMPACT_TTL_SECONDS = 2_592_000 REFORM_IMPACT_INDEX_LIMIT = 1_000 REFORM_IMPACT_START_CLAIM_TTL_SECONDS = 300 diff --git a/policyengine_api/services/budget_window_cache.py b/policyengine_api/services/budget_window_cache.py index 9402bc952..b6190341a 100644 --- a/policyengine_api/services/budget_window_cache.py +++ b/policyengine_api/services/budget_window_cache.py @@ -19,9 +19,8 @@ ) from policyengine_api.runtime_cache.dependencies import get_runtime_cache_context - BUDGET_WINDOW_CACHE_FAMILY = "budget-window" -BUDGET_WINDOW_CACHE_SCHEMA_VERSION = 2 +BUDGET_WINDOW_CACHE_SCHEMA_VERSION = 4 BUDGET_WINDOW_STARTING_TTL_SECONDS = 300 BUDGET_WINDOW_BATCH_TTL_SECONDS = 86_400 BUDGET_WINDOW_RESULT_TTL_SECONDS = 2_592_000 diff --git a/policyengine_api/services/economy_service.py b/policyengine_api/services/economy_service.py index 9cec6e90c..ce7112851 100644 --- a/policyengine_api/services/economy_service.py +++ b/policyengine_api/services/economy_service.py @@ -1530,6 +1530,8 @@ def _build_options_hash( bundle_parts.append(f"data_version={data_version}") if policyengine_version: bundle_parts.append(f"policyengine_version={policyengine_version}") + if options.get("spm") is not None: + bundle_parts.append("spm_provenance_schema=canonical-spm-comparison-v2") if runtime_app_name: bundle_parts.append(f"runtime_app_name={runtime_app_name}") return "[" + "&".join([option_pairs, *bundle_parts]).strip("&") + "]" diff --git a/policyengine_api/services/household_calculation_service.py b/policyengine_api/services/household_calculation_service.py index 3d6284567..bce5743c7 100644 --- a/policyengine_api/services/household_calculation_service.py +++ b/policyengine_api/services/household_calculation_service.py @@ -26,14 +26,17 @@ ) from policyengine_api.utils.deprecated_inputs import drop_deprecated_inputs from policyengine_api.utils.input_validation import find_unrecognized_inputs -from policyengine_api.spm import normalize_spm_selection +from policyengine_api.spm import ( + normalize_spm_selection, + resolved_spm_configuration, + validate_spm_calculation_provenance, +) @dataclass(frozen=True) class CalculationResult: household: dict warnings: tuple[str, ...] = () - spm_config: dict | None = None spm_provenance: dict | None = None @@ -42,7 +45,6 @@ class HouseholdCalculationResult: household: dict warnings: tuple[str, ...] = () cached: bool = False - spm_config: dict | None = None spm_provenance: dict | None = None @@ -73,6 +75,21 @@ def __init__(self, invalid_inputs: list[Any]) -> None: super().__init__("Household or policy contains unrecognized inputs") +def _validated_spm_provenance( + spm_provenance: object, + expected_selection: dict | None, +) -> dict | None: + if expected_selection is None: + if spm_provenance is None: + return None + raise ValueError("Unexpected SPM provenance for a calculation without SPM") + receipt = validate_spm_calculation_provenance(spm_provenance) + canonical_config = resolved_spm_configuration(receipt).model_dump(mode="json") + if canonical_config != expected_selection: + raise ValueError("SPM calculation configuration differs from the request") + return receipt.model_dump(mode="json", by_alias=True) + + def get_household_year(household: dict) -> int | str: household_year: int | str = date.today().year household_age_list = list( @@ -192,7 +209,6 @@ def _store_result( CachedHouseholdCalculation( household=calculation.household, warnings=warnings, - spm_config=calculation.spm_config, spm_provenance=calculation.spm_provenance, ), ) @@ -244,7 +260,6 @@ def calculate_stored_household( household=cached.household, warnings=cached.warnings, cached=True, - spm_config=cached.spm_config, spm_provenance=cached.spm_provenance, ) @@ -296,7 +311,6 @@ def calculate_stored_household( calculation = CalculationResult( household=raw_calculation.household, warnings=tuple(getattr(raw_calculation, "warnings", ())), - spm_config=getattr(raw_calculation, "spm_config", None), spm_provenance=getattr(raw_calculation, "spm_provenance", None), ) else: @@ -305,6 +319,14 @@ def calculate_stored_household( calculation = CalculationResult( household=raw_calculation, ) + spm_provenance = _validated_spm_provenance( + calculation.spm_provenance, + spm, + ) + calculation = replace( + calculation, + spm_provenance=spm_provenance, + ) record_cache_event( family="household-calculation", event="recompute", @@ -322,7 +344,6 @@ def calculate_stored_household( return HouseholdCalculationResult( household=calculation.household, warnings=response_warnings, - spm_config=calculation.spm_config, spm_provenance=calculation.spm_provenance, ) @@ -398,11 +419,14 @@ def calculate_prepared_household( else: household = raw_calculation.household calculation_warnings = tuple(getattr(raw_calculation, "warnings", ())) + spm_provenance = _validated_spm_provenance( + getattr(raw_calculation, "spm_provenance", None), + prepared.spm, + ) return HouseholdCalculationResult( household=household, warnings=(prepared.warnings + calculation_warnings), - spm_config=getattr(raw_calculation, "spm_config", None), - spm_provenance=getattr(raw_calculation, "spm_provenance", None), + spm_provenance=spm_provenance, ) @observability_runtime.span( diff --git a/policyengine_api/services/report_output_service.py b/policyengine_api/services/report_output_service.py index 28bdd3e9e..69246e827 100644 --- a/policyengine_api/services/report_output_service.py +++ b/policyengine_api/services/report_output_service.py @@ -1,6 +1,7 @@ import json from dataclasses import dataclass from datetime import datetime, timezone +from typing import cast from sqlalchemy import select from sqlalchemy.orm import Session, sessionmaker @@ -8,6 +9,7 @@ from policyengine_api.constants import get_report_output_cache_version from policyengine_api.data.orm import get_v1_session_factory from policyengine_api.data.v1_models import ReportOutput, ReportOutputRun, Simulation +from policyengine_api.json_types import JSONObject, JSONValue from policyengine_api.services.report_run_service import ReportRunService from policyengine_api.services.report_spec_service import ( ECONOMY_REPORT_KINDS, @@ -15,6 +17,72 @@ ReportSpecService, ) from policyengine_api.services.simulation_service import SimulationService +from policyengine_api.spm import ( + SPMComparisonProvenance, + error_message, +) + + +def _canonicalize_annual_spm_output( + output: JSONObject, + *, + expected_year: str | None, +) -> JSONObject: + """Validate and serialize compact SPM provenance on one economy result.""" + + if "spm_config" in output: + raise ValueError( + "Completed US economy output must not include legacy spm_config" + ) + if "spm_provenance" not in output: + raise ValueError("Completed US economy output requires SPM provenance") + + comparison = SPMComparisonProvenance.model_validate(output["spm_provenance"]) + if expected_year is not None and any( + expected_year not in execution.receipt.years + for execution in (comparison.baseline, comparison.reform) + ): + raise ValueError("SPM receipt does not cover the report year") + return { + **output, + "spm_provenance": cast( + JSONValue, + comparison.model_dump(mode="json", by_alias=True), + ), + } + + +def _canonicalize_society_wide_spm_output( + output: object, + *, + expected_year: str | None, +) -> object: + """Validate compact annual or budget-window society-wide SPM output.""" + + if not isinstance(output, dict): + return output + typed_output = cast(JSONObject, output) + if output.get("kind") != "budgetWindow": + return _canonicalize_annual_spm_output( + typed_output, + expected_year=expected_year, + ) + + annual_impacts = output.get("annualImpacts") + if not isinstance(annual_impacts, list): + raise ValueError("SPM budget-window output requires annual impact objects") + canonical_rows: list[JSONValue] = [] + for row in annual_impacts: + if not isinstance(row, dict): + raise ValueError("SPM budget-window output requires annual impact objects") + row_year = row.get("year") + canonical_rows.append( + _canonicalize_annual_spm_output( + cast(JSONObject, row), + expected_year=str(row_year) if row_year is not None else None, + ) + ) + return {**typed_output, "annualImpacts": canonical_rows} @dataclass(frozen=True) @@ -648,6 +716,23 @@ def _update_report_output( "Cannot mark report output running without an active pending " "or running report run" ) + effective_status = values.get("status", report_output.status) + effective_output = values.get("output", report_output.output) + if ( + effective_status == "complete" + and report_output.country_id == "us" + and report_output.report_kind in ECONOMY_REPORT_KINDS + ): + try: + values["output"] = _canonicalize_society_wide_spm_output( + effective_output, + expected_year=report_output.year, + ) + except (ValueError, TypeError, KeyError) as error: + raise ValueError( + "Completed US economy report has invalid SPM provenance: " + f"{error_message(error)}" + ) from error for field, value in values.items(): setattr(report_output, field, value) self._ensure_report_output_dual_write_state( diff --git a/policyengine_api/services/simulation_service.py b/policyengine_api/services/simulation_service.py index dfbe828ce..943b0fb00 100644 --- a/policyengine_api/services/simulation_service.py +++ b/policyengine_api/services/simulation_service.py @@ -8,6 +8,7 @@ from policyengine_api.constants import COUNTRY_PACKAGE_VERSIONS from policyengine_api.data.orm import get_v1_session_factory from policyengine_api.data.v1_models import Simulation, SimulationRun +from policyengine_api.spm import validate_spm_calculation_provenance from policyengine_api.utils.population_identity import ( canonical_numeric_household_id, population_id_matches, @@ -52,6 +53,20 @@ def _select_simulation( def _latest_successful_run_id(runs: list[SimulationRun]) -> str | None: return next((run.id for run in runs if run.status == "complete"), None) + @staticmethod + def _has_stale_spm_output(output: object) -> bool: + if not isinstance(output, dict): + return False + if "spm_config" in output: + return True + if "spm_provenance" not in output: + return False + try: + validate_spm_calculation_provenance(output["spm_provenance"]) + except ValueError: + return True + return False + def _ensure_simulation_dual_write_state( self, session: Session, @@ -67,6 +82,11 @@ def _ensure_simulation_dual_write_state( if simulation is None: raise ValueError(f"Simulation #{simulation_id} not found") + if self._has_stale_spm_output(simulation.output): + simulation.status = "pending" + simulation.output = None + simulation.error_message = None + spec = { "country_id": simulation.country_id, "population_id": simulation.population_id, @@ -252,6 +272,19 @@ def update_simulation( return None if isinstance(values.get("output"), str): values["output"] = json.loads(values["output"]) + output_value = values.get("output") + if isinstance(output_value, dict) and "spm_config" in output_value: + raise ValueError( + "Completed simulation output must not include legacy spm_config" + ) + if isinstance(output_value, dict) and "spm_provenance" in output_value: + receipt = validate_spm_calculation_provenance( + output_value["spm_provenance"] + ) + values["output"] = { + **output_value, + "spm_provenance": receipt.model_dump(mode="json", by_alias=True), + } with self._sessions.begin() as session: simulation = self._select_simulation( session, diff --git a/policyengine_api/services/v2/reports/types.py b/policyengine_api/services/v2/reports/types.py index 9d0298f43..ef4d997f4 100644 --- a/policyengine_api/services/v2/reports/types.py +++ b/policyengine_api/services/v2/reports/types.py @@ -16,6 +16,7 @@ SimulationRole, StrictContractModel, ) +from policyengine_api.spm import SPMComparisonProvenance ContractVersion = Literal[1] @@ -102,3 +103,13 @@ def require_unique_aggregates( if len(value) != len(set(value)): raise ValueError("requested aggregates must be unique") return value + + @model_validator(mode="after") + def validate_spm_provenance(self) -> AggregateReportArtifactPayload: + """Reject duplicate or noncanonical SPM data in aggregate artifacts.""" + + if "spm_config" in self.result: + raise ValueError("aggregate result must not include legacy spm_config") + if "spm_provenance" in self.result: + SPMComparisonProvenance.model_validate(self.result["spm_provenance"]) + return self diff --git a/policyengine_api/services/v2/simulations/types.py b/policyengine_api/services/v2/simulations/types.py index cc9b0a723..d1e466b13 100644 --- a/policyengine_api/services/v2/simulations/types.py +++ b/policyengine_api/services/v2/simulations/types.py @@ -16,6 +16,7 @@ ) from policyengine_api.query_parameters import CountryId +from policyengine_api.spm import SPMProvenance ContractText = Annotated[str, Field(min_length=1, max_length=255)] ContractVersion = Literal[1] @@ -205,4 +206,4 @@ class SimulationArtifactDescriptor(StrictContractModel): output_schema_version: ContractVersion = 1 row_identity: RowIdentity bundle: BundleProvenance - calculation_provenance: dict[str, JsonValue] | None = None + calculation_provenance: SPMProvenance | None = None diff --git a/policyengine_api/specification.py b/policyengine_api/specification.py index 11b4b3c8f..e2c9296b6 100644 --- a/policyengine_api/specification.py +++ b/policyengine_api/specification.py @@ -13,7 +13,13 @@ AnnualEconomyQuery, BudgetWindowEconomyQuery, ) -from policyengine_api.spm import SPMProvenance, SPMSelection +from policyengine_api.spm import ( + SPMComparisonProvenance, + SPMExecutionProvenance, + SPMProvenance, + SPMRuntimeVersions, + SPMSelection, +) DEFAULT_SPECIFICATION_PATH = Path(__file__).with_name("openapi_spec.yaml") @@ -62,6 +68,15 @@ def _economy_query_parameters(model): return result +def _component_schema(model): + """Generate one OpenAPI component and publish nested models separately.""" + schema = model.model_json_schema( + ref_template="#/components/schemas/{model}", + ) + schema.pop("$defs", None) + return _openapi_30_schema(schema) + + def load_specification( path: Path = DEFAULT_SPECIFICATION_PATH, version: str = VERSION, @@ -74,10 +89,13 @@ def load_specification( schemas = document.get("components", {}).get("schemas", {}) for name, model in { "SPMSelection": SPMSelection, + "SPMRuntimeVersions": SPMRuntimeVersions, "SPMProvenance": SPMProvenance, + "SPMExecutionProvenance": SPMExecutionProvenance, + "SPMComparisonProvenance": SPMComparisonProvenance, }.items(): if name in schemas: - schemas[name] = _openapi_30_schema(model.model_json_schema()) + schemas[name] = _component_schema(model) economy_path = "/{country_id}/economy/{policy_id}/over/{baseline_policy_id}" for suffix, query_model in ( ("", AnnualEconomyQuery), diff --git a/policyengine_api/spm.py b/policyengine_api/spm.py index c3c627ea9..4f92a3b96 100644 --- a/policyengine_api/spm.py +++ b/policyengine_api/spm.py @@ -8,9 +8,10 @@ import importlib import inspect -from functools import lru_cache +from collections.abc import Mapping, Sequence from datetime import date -from typing import Any, Literal +from functools import lru_cache +from typing import Literal from pydantic import ( BaseModel, @@ -96,19 +97,218 @@ def validate_location(self): return self +class SPMResolvedConfiguration(BaseModel): + """An internal complete SPM selection derived from a compact receipt.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + forecast_content_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + scenario: str = Field(min_length=1, pattern=r"^\S+$") + geography_kind: Literal["county", "national", "metro"] + geography_id: str | None = Field(min_length=1) + county_vintage: str = Field(pattern=r"^[0-9]{4}$") + as_of: str | None + + @field_validator("as_of") + @classmethod + def validate_as_of(cls, value: str | None) -> str | None: + if value is not None and date.fromisoformat(value).isoformat() != value: + raise ValueError("as_of must be an ISO calendar date (YYYY-MM-DD)") + return value + + @model_validator(mode="after") + def validate_location(self) -> "SPMResolvedConfiguration": + if self.geography_kind == "metro": + if self.geography_id is None or not self.geography_id.strip(): + raise ValueError("An SPM area selection requires geography_id") + elif self.geography_id is not None: + raise ValueError("Only an SPM area selection accepts geography_id") + return self + + class SPMProvenance(BaseModel): - """JSON receipt matching the country's public provenance contract.""" + """Compact public receipt for one canonical SPM execution.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + schema_version: Literal["canonical-spm-provenance-v2"] + forecast_id: str = Field(min_length=1, pattern=r"^\S+$") + forecast_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + scenario: str = Field(min_length=1, pattern=r"^\S+$") + geography_kind: Literal["county", "national", "metro"] + geography_id: str | None = Field(min_length=1) + county_vintage: str = Field(pattern=r"^[0-9]{4}$") + as_of: date | None + years: tuple[str, ...] + runtime_versions: "SPMRuntimeVersions" + + @field_validator("years") + @classmethod + def validate_years(cls, years: tuple[str, ...]) -> tuple[str, ...]: + if any( + len(year) != 4 or not year.isascii() or not year.isdigit() for year in years + ): + raise ValueError("years must contain four-digit calendar years") + if tuple(sorted(set(years))) != years: + raise ValueError("years must be sorted and unique") + return years + + @model_validator(mode="after") + def validate_location(self) -> "SPMProvenance": + if self.geography_kind == "metro": + if self.geography_id is None or not self.geography_id.strip(): + raise ValueError("A metro SPM receipt requires geography_id") + elif self.geography_id is not None: + raise ValueError("Only a metro SPM receipt accepts geography_id") + return self + + +class SPMRuntimeVersions(BaseModel): + """Package versions required to reproduce one certified SPM execution.""" + + model_config = ConfigDict(frozen=True, extra="forbid", populate_by_name=True) + policyengine: str = Field(min_length=1) + policyengine_core: str = Field(min_length=1, alias="policyengine-core") + policyengine_us: str = Field(min_length=1, alias="policyengine-us") + spm_calculator: str = Field(min_length=1, alias="spm-calculator") + + +SPMProvenance.model_rebuild() + + +class SPMExecutionProvenance(BaseModel): + """One shared compact receipt and the executions that produced it.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + receipt: SPMProvenance + execution_count: int = Field(ge=1) + - model_config = ConfigDict(extra="forbid") - forecast_id: str - forecast_sha256: str - scenario: str - geography_kind: str - runtime_versions: dict[str, str | None] - years: dict[str, dict[str, Any]] - geographies: list[dict[str, Any]] - composition_method: str - storage_method: str +class SPMComparisonProvenance(BaseModel): + """Compact baseline and reform receipts for a society-wide result.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + schema_version: Literal["canonical-spm-comparison-v2"] + baseline: SPMExecutionProvenance + reform: SPMExecutionProvenance + + @model_validator(mode="after") + def validate_compatible_receipts(self) -> "SPMComparisonProvenance": + if self.baseline.receipt != self.reform.receipt: + raise ValueError("baseline and reform SPM receipts differ") + return self + + +def build_spm_provenance( + *, + forecast_id: str, + forecast_sha256: str, + selection: SPMSelection, + years: Sequence[str], + runtime_versions: SPMRuntimeVersions, +) -> SPMProvenance: + """Build one compact receipt from typed calculation inputs.""" + + if selection.forecast_content_sha256 != forecast_sha256: + raise ValueError("SPM receipt differs from the SPM configuration") + if selection.scenario is None: + raise ValueError("Resolved SPM configuration must include scenario") + return SPMProvenance( + schema_version="canonical-spm-provenance-v2", + forecast_id=forecast_id, + forecast_sha256=forecast_sha256, + scenario=selection.scenario, + geography_kind=selection.geography_kind, + geography_id=selection.geography_id, + county_vintage=selection.county_vintage, + as_of=selection.as_of, + years=tuple(years), + runtime_versions=runtime_versions, + ) + + +def _required_string(source: Mapping[str, object], field: str) -> str: + value = source.get(field) + if not isinstance(value, str) or not value: + raise ValueError(f"Country SPM receipt requires {field}") + return value + + +def build_spm_calculation_provenance( + config: SPMSelection, + receipt: Mapping[str, object], +) -> SPMProvenance: + """Reduce the country-owned rich receipt to the public compact contract.""" + + forecast_sha256 = _required_string(receipt, "forecast_sha256") + scenario = _required_string(receipt, "scenario") + geography_kind = _required_string(receipt, "geography_kind") + if ( + forecast_sha256 != config.forecast_content_sha256 + or scenario != config.scenario + or geography_kind != config.geography_kind + ): + raise ValueError("Country SPM receipt differs from the SPM configuration") + + source_years = receipt.get("years") + if not isinstance(source_years, Mapping): + raise ValueError("Country SPM receipt requires a years mapping") + years: list[str] = [] + for year in source_years: + if not isinstance(year, str): + raise ValueError("Country SPM receipt years must be strings") + years.append(year) + + source_versions = receipt.get("runtime_versions") + if not isinstance(source_versions, Mapping): + raise ValueError("Country SPM receipt requires runtime_versions") + runtime_versions = SPMRuntimeVersions.model_validate(source_versions) + return build_spm_provenance( + forecast_id=_required_string(receipt, "forecast_id"), + forecast_sha256=forecast_sha256, + selection=config, + years=tuple(sorted(years)), + runtime_versions=runtime_versions, + ) + + +def resolved_spm_configuration( + provenance: SPMProvenance, +) -> SPMResolvedConfiguration: + """Derive the complete resolved selection from its compact receipt.""" + + return SPMResolvedConfiguration( + forecast_content_sha256=provenance.forecast_sha256, + scenario=provenance.scenario, + geography_kind=provenance.geography_kind, + geography_id=provenance.geography_id, + county_vintage=provenance.county_vintage, + as_of=provenance.as_of.isoformat() if provenance.as_of is not None else None, + ) + + +def _collapse_execution_receipts( + side: Literal["baseline", "reform"], + receipts: Sequence[SPMProvenance], +) -> SPMExecutionProvenance: + if not receipts: + raise ValueError(f"{side} SPM receipts are empty") + first = receipts[0] + if any(receipt != first for receipt in receipts[1:]): + raise ValueError(f"{side} SPM receipts differ") + return SPMExecutionProvenance(receipt=first, execution_count=len(receipts)) + + +def build_spm_comparison_provenance( + *, + baseline_receipts: Sequence[SPMProvenance], + reform_receipts: Sequence[SPMProvenance], +) -> SPMComparisonProvenance: + """Collapse matching child receipts into one comparison receipt.""" + + return SPMComparisonProvenance( + schema_version="canonical-spm-comparison-v2", + baseline=_collapse_execution_receipts("baseline", baseline_receipts), + reform=_collapse_execution_receipts("reform", reform_receipts), + ) class SPMValidationError(ValueError): @@ -190,12 +390,9 @@ def simulation_supports_spm(simulation_type) -> bool: ) -# A resolved selection freezes all six fields, but output transports may omit -# null values. Only those omissions are safe: a receipt must never inherit a -# non-null setting from today's bundle defaults. -REQUIRED_RESOLVED_SPM_FIELDS = frozenset( - {"forecast_content_sha256", "scenario", "geography_kind", "county_vintage"} -) +# A transported resolved selection freezes all six fields, including explicit +# null values. Receipt validation never fills output from current defaults. +REQUIRED_RESOLVED_SPM_FIELDS = frozenset(SPMSelection.model_fields) def resolved_spm_settings(settings: object) -> dict | None: @@ -214,6 +411,14 @@ def resolved_spm_settings(settings: object) -> dict | None: return {name: getattr(config, name) for name in SPMSelection.model_fields} +def validate_spm_calculation_provenance( + provenance: object, +) -> SPMProvenance: + """Validate the sole transported compact receipt for a completed result.""" + + return SPMProvenance.model_validate(provenance) + + def _current_bundle() -> dict: from policyengine_api.constants import _policyengine_bundle @@ -413,25 +618,24 @@ def spm_metadata(country_id: str) -> dict: def calculation_spm_receipt(simulation) -> dict: - """Read existing calculation receipts; never request SPM calculations here.""" + """Reduce a country's existing receipt; never request calculations here.""" if not hasattr(simulation, "spm_config"): return {} - # A receipt this API cannot read in full is an uncertified country contract, - # not a caller error and not an internal failure. Report it as a typed - # configuration failure rather than letting pydantic surface a 500, and - # never publish a receipt with unrecognized fields dropped. try: - return { - "spm_config": SPMSelection.model_validate(simulation.spm_config).model_dump( - mode="json" - ), - "spm_provenance": SPMProvenance.model_validate( - simulation.spm_provenance() - ).model_dump(mode="json"), - } - except ValidationError as error: + resolved = resolved_spm_settings(simulation.spm_config) + if resolved is None: + raise ValueError("Country SPM configuration is incomplete") + source = simulation.spm_provenance() + if not isinstance(source, Mapping): + raise ValueError("Country SPM receipt must be an object") + calculation = build_spm_calculation_provenance( + SPMSelection.model_validate(resolved), + source, + ) + return {"spm_provenance": calculation.model_dump(mode="json", by_alias=True)} + except (ValidationError, ValueError) as error: raise SPMValidationError( "SPM_CONFIGURATION_UNAVAILABLE", "The installed country model reported an SPM receipt this API cannot " - f"certify: {readable_validation_error(error)}", + f"certify: {error_message(error)}", ) from error diff --git a/policyengine_api/worker_spm.py b/policyengine_api/worker_spm.py index 28d8566c8..2d9977d90 100644 --- a/policyengine_api/worker_spm.py +++ b/policyengine_api/worker_spm.py @@ -1,12 +1,13 @@ """Canonical economy execution requires the selected worker's certified capability.""" from policyengine_api.spm import ( + SPMComparisonProvenance, SPMSelection, - SPMProvenance, SPMValidationError, error_message, normalize_spm_selection, - resolved_spm_settings, + resolved_spm_configuration, + validate_spm_calculation_provenance, ) @@ -95,32 +96,19 @@ def validate_worker_result( return if expected_years is not None: raise ValueError("Expected a budget window SPM result") - resolved = resolved_spm_settings(result.get("spm_config")) - if resolved is None: - raise ValueError("Worker result has incomplete resolved SPM settings") - if resolved != selection: - raise ValueError("Worker result SPM settings differ from the request") - receipts = result.get("spm_provenance") - if not isinstance(receipts, dict) or set(receipts) != {"baseline", "reform"}: - raise ValueError("Worker result has no baseline/reform SPM receipts") + if "spm_config" in result: + raise ValueError("Worker result must not include legacy spm_config") + comparison = SPMComparisonProvenance.model_validate( + result.get("spm_provenance") + ) for side in ("baseline", "reform"): - if not isinstance(receipts[side], list) or not receipts[side]: - raise ValueError("Worker result has an empty SPM receipt") - for item in receipts[side]: - receipt = SPMProvenance.model_validate(item) - if ( - expected_year is not None - and str(expected_year) not in receipt.years - ): - raise ValueError("Worker SPM receipt does not cover requested year") - if ( - receipt.forecast_sha256 != selection["forecast_content_sha256"] - or receipt.scenario != selection["scenario"] - or receipt.geography_kind != selection["geography_kind"] - ): - raise ValueError( - "Worker result SPM provenance differs from the request" - ) + execution = getattr(comparison, side) + receipt = validate_spm_calculation_provenance(execution.receipt) + resolved = resolved_spm_configuration(receipt).model_dump(mode="json") + if resolved != selection: + raise ValueError("Worker result SPM settings differ from the request") + if expected_year is not None and str(expected_year) not in receipt.years: + raise ValueError("Worker SPM receipt does not cover requested year") except (ValueError, TypeError, KeyError) as exc: raise SPMValidationError( "SPM_CONFIGURATION_UNAVAILABLE", error_message(exc) diff --git a/tests/integration/test_runtime_cache_redis.py b/tests/integration/test_runtime_cache_redis.py index 913e9193d..e12b04575 100644 --- a/tests/integration/test_runtime_cache_redis.py +++ b/tests/integration/test_runtime_cache_redis.py @@ -120,25 +120,29 @@ def test_cached_calculation_receipts_survive_a_real_round_trip(redis_pair) -> No value = CachedHouseholdCalculation( household={"people": {"you": {}}}, warnings=(), - # A canonical receipt may omit its null settings; that is the same choice. - spm_config={ - name: setting for name, setting in selection.items() if setting is not None - }, spm_provenance={ + "schema_version": "canonical-spm-provenance-v2", "forecast_id": "test-artifact", "forecast_sha256": selection["forecast_content_sha256"], "scenario": "baseline", "geography_kind": "national", - "runtime_versions": {"policyengine-us": "test-only"}, - "years": {"2026": {"status": "forecast"}}, - "geographies": [], - "composition_method": "classified-inputs", - "storage_method": "formula", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, + "runtime_versions": { + "policyengine": "test-only", + "policyengine-core": "test-only", + "policyengine-us": "test-only", + "spm-calculator": "test-only", + }, + "years": ["2026"], }, ) assert HouseholdCalculationCache(first, namespace).set(identity, value) - assert HouseholdCalculationCache(second, namespace).get(identity) == value + cached = HouseholdCalculationCache(second, namespace).get(identity) + assert cached is not None + assert cached.spm_provenance == value.spm_provenance other = HouseholdCalculationIdentity( **{**vars(identity), "spm": {**selection, "geography_kind": "county"}} ) diff --git a/tests/unit/routes/test_canonical_spm.py b/tests/unit/routes/test_canonical_spm.py index 2dd51f9d3..767de092a 100644 --- a/tests/unit/routes/test_canonical_spm.py +++ b/tests/unit/routes/test_canonical_spm.py @@ -69,18 +69,24 @@ def calculate(self, household, policy, **kwargs): None if config is None else { + "schema_version": "canonical-spm-provenance-v2", "forecast_id": "test-artifact", "forecast_sha256": config["forecast_content_sha256"], "scenario": config["scenario"], "geography_kind": config["geography_kind"], - "runtime_versions": {"policyengine-us": "test-only"}, - "years": {"2026": {"status": "forecast"}}, - "geographies": [], - "composition_method": "classified-inputs", - "storage_method": "formula", + "geography_id": config["geography_id"], + "county_vintage": config["county_vintage"], + "as_of": config["as_of"], + "runtime_versions": { + "policyengine": "test-only", + "policyengine-core": "test-only", + "policyengine-us": "test-only", + "spm-calculator": "test-only", + }, + "years": ["2026"], } ) - return CalculationResult(household, (), config, receipt) + return CalculationResult(household, (), receipt) @pytest.fixture @@ -215,7 +221,7 @@ def test_settings_and_provenance_survive_storage_reform_replay_and_replacement( cached = client.get(url + "/policy/2") assert first.status_code == cached.status_code == 200 assert first.json == cached.json - assert first.json["spm_config"] == stored["spm"] + assert "spm_config" not in first.json assert first.json["spm_provenance"]["forecast_sha256"] == FORECAST_HASH assert len(country.calls) == 1 assert "spm" not in country.calls[0][0] @@ -251,7 +257,7 @@ def test_settings_and_provenance_survive_storage_reform_replay_and_replacement( latest = client.get(latest_url).json["result"] assert latest["household_hash"] != edited["household_hash"] assert ( - client.get(latest_url + "/policy/2").json["spm_config"]["geography_kind"] + client.get(latest_url + "/policy/2").json["spm_provenance"]["geography_kind"] == "county" ) assert len(country.calls) == 4 @@ -280,7 +286,7 @@ def test_http_cache_varies_with_measurement_settings(certified, harness, selecti second = client.post("/us/calculate", json=payload) assert second.status_code == 200 assert len(country.calls) == 2 - assert first.json["spm_config"] != second.json["spm_config"] + assert first.json["spm_provenance"] != second.json["spm_provenance"] def test_successful_http_cache_hit_still_starts_observability(certified, harness): @@ -330,7 +336,7 @@ def test_http_cache_default_artifact_hash_is_part_of_identity(certified, harness second = client.post("/us/calculate", json=payload) assert first.status_code == second.status_code == 200 assert len(country.calls) == 2 - assert first.json["spm_config"] != second.json["spm_config"] + assert first.json["spm_provenance"] != second.json["spm_provenance"] def test_linked_household_measurement_is_immutable_for_saved_reports( @@ -538,14 +544,8 @@ def test_valid_spm_database_timeout_keeps_stage11_safe_persistence_response( copy_event.assert_not_called() -def test_stored_replay_hits_the_calculation_cache_when_a_receipt_omits_nulls( - certified, harness -): - """A canonical country may omit null receipt settings; replay must still hit. - - Requiring exact JSON equality between the receipt and the resolved identity - made every stored replay recompute. - """ +def test_stored_replay_rejects_a_receipt_that_omits_null_settings(certified, harness): + """The sole receipt must state every field in its resolved configuration.""" client, country = harness calculate = country.calculate @@ -554,12 +554,11 @@ def omitting_null_settings(household, policy, **kwargs): return CalculationResult( household=result.household, warnings=result.warnings, - spm_config={ + spm_provenance={ key: value - for key, value in result.spm_config.items() + for key, value in result.spm_provenance.items() if value is not None }, - spm_provenance=result.spm_provenance, ) country.calculate = omitting_null_settings @@ -568,18 +567,9 @@ def omitting_null_settings(household, policy, **kwargs): ) assert created.status_code == 201, created.json url = f"/us/household/{created.json['result']['household_id']}/policy/2" - first = client.get(url) - assert first.status_code == 200, first.json - assert set(first.json["spm_config"]) == { - "forecast_content_sha256", - "scenario", - "geography_kind", - "county_vintage", - } - cached = client.get(url) - assert cached.status_code == 200 - assert cached.json == first.json - assert len(country.calls) == 1 + assert client.get(url).status_code == 500 + assert client.get(url).status_code == 500 + assert len(country.calls) == 2 @pytest.mark.parametrize("version", ["5.2.1", "5.4.0", "6.0.0"]) diff --git a/tests/unit/routes/test_household_numeric_identity.py b/tests/unit/routes/test_household_numeric_identity.py index 524cf4e5b..e65e4b13b 100644 --- a/tests/unit/routes/test_household_numeric_identity.py +++ b/tests/unit/routes/test_household_numeric_identity.py @@ -219,9 +219,12 @@ def test_numeric_link_preserves_immutable_household_and_saved_simulation( assert saved["population_id"] == saved_spelling assert json.loads(saved["simulation_spec_json"])["population_id"] == saved_spelling if reference == "historical-padded": - assert saved["status"] == "complete" - assert json.loads(saved["output"]) == output + assert saved["status"] == "pending" + assert saved["output"] is None + assert saved["latest_successful_run_id"] is None + assert saved["active_run_id"] is not None with orm_session_factory() as session: - run = session.get(SimulationRun, saved["latest_successful_run_id"]) + run = session.get(SimulationRun, saved["active_run_id"]) assert run.simulation_spec_snapshot_json["population_id"] == saved_spelling - assert run.output == output + assert run.status == "pending" + assert run.output is None diff --git a/tests/unit/routes/test_simulation_spm.py b/tests/unit/routes/test_simulation_spm.py index 59b591a5a..22a151696 100644 --- a/tests/unit/routes/test_simulation_spm.py +++ b/tests/unit/routes/test_simulation_spm.py @@ -46,7 +46,7 @@ def test_simulation_records_reject_spm_overrides_before_read_or_write( @pytest.mark.parametrize("population_id", ["00001", "1", "0000001", 1]) -def test_repost_historical_numeric_household_alias_replays_completed_output( +def test_repost_historical_numeric_household_alias_recalculates_old_spm_output( orm_session_factory, population_id ): output = {"household": {"people": {}}, "spm_config": {"geography_kind": "national"}} @@ -81,8 +81,8 @@ def test_repost_historical_numeric_household_alias_replays_completed_output( assert response.status_code == 200 result = response.get_json()["result"] assert result["id"] == historical_id - assert result["status"] == "complete" - assert json.loads(result["output"]) == output + assert result["status"] == "pending" + assert result["output"] is None assert result["population_id"] == "00001" assert ( json.loads(result["simulation_spec_json"])["population_id"] == "00001" @@ -91,6 +91,6 @@ def test_repost_historical_numeric_household_alias_replays_completed_output( with orm_session_factory() as session: assert session.scalar(select(func.count()).select_from(Simulation)) == 1 run = session.scalar(select(SimulationRun)) - assert run.status == "complete" - assert run.output == output + assert run.status == "pending" + assert run.output is None assert run.simulation_spec_snapshot_json["population_id"] == "00001" diff --git a/tests/unit/routes/test_spm_year_worker_polling.py b/tests/unit/routes/test_spm_year_worker_polling.py index c79f023f4..3faa0ab02 100644 --- a/tests/unit/routes/test_spm_year_worker_polling.py +++ b/tests/unit/routes/test_spm_year_worker_polling.py @@ -15,7 +15,6 @@ from policyengine_api.services.economy_service import EconomyService from policyengine_api.services.reform_impacts_service import ReformImpactsService - SELECTION = { "forecast_content_sha256": "a" * 64, "scenario": "baseline", @@ -32,26 +31,32 @@ def segmented_result(year, invalid_side=None): receipt = { + "schema_version": "canonical-spm-provenance-v2", "forecast_id": "test-only", "forecast_sha256": SELECTION["forecast_content_sha256"], "scenario": SELECTION["scenario"], "geography_kind": SELECTION["geography_kind"], - "runtime_versions": {}, - "years": {year: {}}, - "geographies": [], - "composition_method": "classified", - "storage_method": "formula", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, + "runtime_versions": { + "policyengine": "test", + "policyengine-core": "test", + "policyengine-us": "test", + "spm-calculator": "test", + }, + "years": [year], } result = { "year": year, - "spm_config": SELECTION, "spm_provenance": { - "baseline": [dict(receipt), dict(receipt)], - "reform": [dict(receipt), dict(receipt)], + "schema_version": "canonical-spm-comparison-v2", + "baseline": {"receipt": dict(receipt), "execution_count": 2}, + "reform": {"receipt": dict(receipt), "execution_count": 2}, }, } if invalid_side: - result["spm_provenance"][invalid_side][1]["years"] = {"2000": {}} + result["spm_provenance"][invalid_side]["receipt"]["years"] = ["2000"] return result @@ -75,7 +80,7 @@ def test_typed_worker_error_is_terminal_and_replays_after_service_recreation( if isinstance(failure, str): typed_error = { "code": "SPM_CONFIGURATION_UNAVAILABLE", - "message": "Worker SPM receipt does not cover requested year", + "message": "baseline and reform SPM receipts differ", } side = failure.removeprefix("segmented-") worker_result = ( diff --git a/tests/unit/runtime_cache/test_household_calculations.py b/tests/unit/runtime_cache/test_household_calculations.py index 27b23a044..cfe422efe 100644 --- a/tests/unit/runtime_cache/test_household_calculations.py +++ b/tests/unit/runtime_cache/test_household_calculations.py @@ -10,6 +10,7 @@ HouseholdCalculationCache, HouseholdCalculationIdentity, ) +from policyengine_api.spm import SPMSelection, build_spm_calculation_provenance def _namespace() -> CacheNamespace: @@ -93,15 +94,21 @@ def test_receipt_bearing_payload_supersedes_the_receiptless_schema() -> None: "as_of": None, } SPM_RECEIPT = { + "schema_version": "canonical-spm-provenance-v2", "forecast_id": "test-artifact", "forecast_sha256": "a" * 64, "scenario": "baseline", "geography_kind": "national", - "runtime_versions": {"policyengine-us": "test"}, - "years": {}, - "geographies": [], - "composition_method": "test composition", - "storage_method": "test storage", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, + "runtime_versions": { + "policyengine": "test", + "policyengine-core": "test", + "policyengine-us": "test", + "spm-calculator": "test", + }, + "years": [], } @@ -112,7 +119,7 @@ def test_receipt_bearing_payload_supersedes_the_receiptless_schema() -> None: {**SPM_RECEIPT, "forecast_sha256": "b" * 64}, {**SPM_RECEIPT, "scenario": "other"}, {**SPM_RECEIPT, "geography_kind": "county"}, - {**SPM_RECEIPT, "years": []}, + {**SPM_RECEIPT, "years": ["2027", "2026"]}, {**SPM_RECEIPT, "unexpected": True}, ], ids=["empty", "hash", "scenario", "geography", "years-shape", "unknown-field"], @@ -122,24 +129,53 @@ def test_household_cache_rejects_invalid_or_mismatched_spm_receipts(receipt): identity = _identity(spm=SPM_CONFIG) value = CachedHouseholdCalculation( household={"people": {}}, - spm_config=SPM_CONFIG, spm_provenance=receipt, ) - assert cache.set(identity, value) is True + assert cache.set(identity, value) is False assert cache.get(identity) is None -@pytest.mark.parametrize("years", [{}, {"2024": {"source": "test"}}]) +def test_household_cache_rejects_previous_diagnostic_receipt_before_write(): + backend = InMemoryCacheBackend() + cache = HouseholdCalculationCache(backend, _namespace()) + identity = _identity(spm=SPM_CONFIG) + previous_receipt = { + "forecast_id": "test-artifact", + "forecast_sha256": "a" * 64, + "scenario": "baseline", + "geography_kind": "national", + "runtime_versions": {}, + "years": {}, + "geographies": [], + "composition_method": "test", + "storage_method": "test", + } + + assert ( + cache.set( + identity, + CachedHouseholdCalculation( + household={"people": {}}, + spm_provenance=previous_receipt, + ), + ) + is False + ) + assert backend._values == {} + + +@pytest.mark.parametrize("years", [[], ["2024"]]) def test_household_cache_keeps_valid_receipts_including_lazy_tax_only(years): cache = HouseholdCalculationCache(InMemoryCacheBackend(), _namespace()) identity = _identity(spm=SPM_CONFIG) value = CachedHouseholdCalculation( household={"people": {}}, - spm_config=SPM_CONFIG, spm_provenance={**SPM_RECEIPT, "years": years}, ) assert cache.set(identity, value) is True - assert cache.get(identity) == value + cached = cache.get(identity) + assert cached is not None + assert cached.spm_provenance == {**SPM_RECEIPT, "years": years} @pytest.mark.parametrize("calculate_spm", [False, True], ids=["tax-only", "measured"]) @@ -154,51 +190,45 @@ def test_real_country_provider_receipt_round_trips_household_cache(calculate_spm tenure="renter", ) config = country_spm.spm_config(provider) - receipt = provider.provenance() + receipt = build_spm_calculation_provenance( + SPMSelection.model_validate(config), provider.provenance() + ).model_dump(mode="json", by_alias=True) assert bool(receipt["years"]) is calculate_spm cache = HouseholdCalculationCache(InMemoryCacheBackend(), _namespace()) identity = _identity(spm=config) value = CachedHouseholdCalculation( household={"people": {}}, - spm_config=config, spm_provenance=receipt, ) assert cache.set(identity, value) is True - assert cache.get(identity) == value - - -OMITTED_NULL_CONFIG = { - key: value for key, value in SPM_CONFIG.items() if value is not None -} + cached = cache.get(identity) + assert cached is not None + assert cached.spm_provenance == receipt -def test_household_cache_hits_when_receipt_omits_null_settings(): - """A canonical receipt may omit its null values; that is still the same selection.""" - cache = HouseholdCalculationCache(InMemoryCacheBackend(), _namespace()) - identity = _identity(spm=SPM_CONFIG) - value = CachedHouseholdCalculation( - household={"people": {}}, - spm_config=OMITTED_NULL_CONFIG, - spm_provenance=SPM_RECEIPT, - ) - assert cache.set(identity, value) is True - assert cache.get(identity) == value - - -@pytest.mark.parametrize("omitted", sorted(OMITTED_NULL_CONFIG)) -def test_household_cache_rejects_receipt_omitting_a_nonnull_setting(omitted): - """An omitted non-null setting must never inherit today's resolved default.""" +@pytest.mark.parametrize( + "omitted", + [ + "forecast_sha256", + "scenario", + "geography_kind", + "geography_id", + "county_vintage", + "as_of", + ], +) +def test_household_cache_rejects_receipt_omitting_resolved_setting(omitted): + """Every resolved setting is required on the sole completed-result receipt.""" cache = HouseholdCalculationCache(InMemoryCacheBackend(), _namespace()) identity = _identity(spm=SPM_CONFIG) value = CachedHouseholdCalculation( household={"people": {}}, - spm_config={ - key: item for key, item in OMITTED_NULL_CONFIG.items() if key != omitted + spm_provenance={ + key: item for key, item in SPM_RECEIPT.items() if key != omitted }, - spm_provenance=SPM_RECEIPT, ) - assert cache.set(identity, value) is True + assert cache.set(identity, value) is False assert cache.get(identity) is None @@ -218,8 +248,28 @@ def test_household_cache_rejects_receipt_with_different_settings(changed): identity = _identity(spm=SPM_CONFIG) value = CachedHouseholdCalculation( household={"people": {}}, - spm_config={**OMITTED_NULL_CONFIG, **changed}, - spm_provenance=SPM_RECEIPT, + spm_provenance={**SPM_RECEIPT, **changed}, ) - assert cache.set(identity, value) is True + assert cache.set(identity, value) is False + assert cache.get(identity) is None + + +def test_household_cache_rejects_legacy_dual_field_payload() -> None: + backend = InMemoryCacheBackend() + cache = HouseholdCalculationCache(backend, _namespace()) + identity = _identity(spm=SPM_CONFIG) + backend.set( + cache.cache_key(identity), + encode_envelope( + "household-calculation", + HOUSEHOLD_CALCULATION_SCHEMA_VERSION, + { + "household": {"people": {}}, + "warnings": [], + "spm_config": SPM_CONFIG, + "spm_provenance": SPM_RECEIPT, + }, + ), + ) + assert cache.get(identity) is None diff --git a/tests/unit/runtime_cache/test_reform_impacts.py b/tests/unit/runtime_cache/test_reform_impacts.py index 43445b86a..26a02cca7 100644 --- a/tests/unit/runtime_cache/test_reform_impacts.py +++ b/tests/unit/runtime_cache/test_reform_impacts.py @@ -4,9 +4,14 @@ import pytest -from policyengine_api.runtime_cache.core import CacheCoordinationError, CacheNamespace +from policyengine_api.runtime_cache.core import ( + CacheCoordinationError, + CacheNamespace, + encode_envelope, +) from policyengine_api.runtime_cache.fake import InMemoryCacheBackend from policyengine_api.runtime_cache.reform_impacts import ( + REFORM_IMPACT_SCHEMA_VERSION, REFORM_IMPACT_START_CLAIM_TTL_SECONDS, CachedReformImpact, ReformImpactCache, @@ -189,6 +194,31 @@ def test_reform_impact_record_and_indexes_share_one_jittered_ttl( assert set(backend._expires.values()) == {123} +def test_previous_schema_reform_impact_is_not_read() -> None: + backend = InMemoryCacheBackend() + namespace = _namespace() + cache = ReformImpactCache(backend, namespace) + legacy = _impact("legacy", "[spm=canonical]", 1) + previous_version = REFORM_IMPACT_SCHEMA_VERSION - 1 + legacy_key = namespace.key( + "reform-impact", previous_version, {"execution_id": "legacy"} + ) + backend.set( + legacy_key, + encode_envelope( + "reform-impact", + previous_version, + { + **legacy.__dict__, + "start_time": legacy.start_time.isoformat(), + "end_time": None, + }, + ), + ) + + assert cache.get_by_execution_id("legacy") is None + + def test_reform_impact_updates_and_deletes_only_matching_computing_values() -> None: backend = InMemoryCacheBackend() cache = ReformImpactCache(backend, _namespace()) diff --git a/tests/unit/services/test_budget_window_cache.py b/tests/unit/services/test_budget_window_cache.py index 1c9009c5e..ff344e924 100644 --- a/tests/unit/services/test_budget_window_cache.py +++ b/tests/unit/services/test_budget_window_cache.py @@ -3,9 +3,11 @@ import pytest from policyengine_api.runtime_cache.core import CacheCoordinationError +from policyengine_api.runtime_cache.core import CacheNamespace, encode_envelope from policyengine_api.runtime_cache.fake import InMemoryCacheBackend from policyengine_api.services.budget_window_cache import ( BUDGET_WINDOW_BATCH_TTL_SECONDS, + BUDGET_WINDOW_CACHE_SCHEMA_VERSION, BUDGET_WINDOW_STARTING_TTL_SECONDS, BudgetWindowCache, BudgetWindowCacheState, @@ -80,7 +82,37 @@ def test_build_key_is_stable_for_request_identity(): ) assert first == second - assert first.startswith("policyengine:test:api:budget-window:v2:") + assert first.startswith("policyengine:test:api:budget-window:v4:") + + +def test_previous_schema_cache_identity_is_not_read(): + backend = FakeRedis() + namespace = CacheNamespace("test", "api") + cache = BudgetWindowCache(client=backend, namespace=namespace) + identity = { + "api_version": "v1", + "baseline_policy_id": 1, + "country_id": "us", + "dataset": "default", + "options_hash": "[spm=canonical]", + "reform_policy_id": 2, + "region": "us", + "time_period": "budget_window:2026:2", + } + previous_version = BUDGET_WINDOW_CACHE_SCHEMA_VERSION - 1 + old_key = namespace.key("budget-window", previous_version, identity) + backend.set( + f"{old_key}:state", + encode_envelope( + "budget-window", + previous_version, + {"status": "completed", "result": {"legacy": True}}, + ), + ) + current_key = cache.build_key(**identity) + + assert current_key != old_key + assert cache.get_state(current_key) is None def test_claim_batch_start_allows_one_starter_and_preserves_identity(): diff --git a/tests/unit/services/test_report_output_service.py b/tests/unit/services/test_report_output_service.py index f4e6d67ea..92344b68e 100644 --- a/tests/unit/services/test_report_output_service.py +++ b/tests/unit/services/test_report_output_service.py @@ -1,6 +1,7 @@ import pytest from sqlalchemy import func, select +from policyengine_api import constants from policyengine_api.constants import get_report_output_cache_version from policyengine_api.data.v1_models import ReportOutput, ReportOutputRun from policyengine_api.services.report_output_service import ReportOutputService @@ -9,6 +10,42 @@ run_service = ReportRunService() +SPM_CONFIG = { + "forecast_content_sha256": "a" * 64, + "scenario": "ce_trend", + "geography_kind": "national", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, +} +SPM_RECEIPT = { + "schema_version": "canonical-spm-provenance-v2", + "forecast_id": "canonical-forecast", + "forecast_sha256": "a" * 64, + "scenario": "ce_trend", + "geography_kind": "national", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, + "years": ["2025"], + "runtime_versions": { + "policyengine": "6.2.1", + "policyengine-core": "3.32.10", + "policyengine-us": "2.2.1", + "spm-calculator": "1.0.0", + }, +} + + +def compact_economy_output() -> dict: + return { + "spm_provenance": { + "schema_version": "canonical-spm-comparison-v2", + "baseline": {"receipt": SPM_RECEIPT, "execution_count": 2}, + "reform": {"receipt": SPM_RECEIPT, "execution_count": 2}, + }, + "budgetary_impact": 42, + } @pytest.fixture @@ -19,15 +56,17 @@ def service(orm_session_factory): def create_simulation( orm_session_factory, *, + country_id="us", policy_id=1, population_id="household-1", + population_type="household", ): return ( SimulationService(orm_session_factory) .get_or_create_simulation( - country_id="us", + country_id=country_id, population_id=population_id, - population_type="household", + population_type=population_type, policy_id=policy_id, ) .simulation @@ -182,6 +221,201 @@ def test_update_accepts_existing_v1_json_string_boundary( assert view.report_output.output == {"ok": True} +def test_update_complete_persists_only_compact_society_wide_spm_output( + service, + orm_session_factory, +): + baseline = create_simulation( + orm_session_factory, + policy_id=1, + population_id="us", + population_type="geography", + ) + reform = create_simulation( + orm_session_factory, + policy_id=2, + population_id="us", + population_type="geography", + ) + report = service.create_or_reuse_report_output( + "us", baseline.id, reform.id, "2025" + ).view.report_output + + view = service.update_report_output( + "us", + report.id, + status="complete", + output=compact_economy_output(), + ) + + assert view.report_output.output == compact_economy_output() + assert view.display_run.output == compact_economy_output() + + +@pytest.mark.parametrize( + ("output", "error"), + [ + ({"budgetary_impact": 42}, "requires SPM provenance"), + ( + {"budgetary_impact": 42, "spm_config": SPM_CONFIG}, + "legacy spm_config", + ), + ( + {**compact_economy_output(), "spm_config": SPM_CONFIG}, + "legacy spm_config", + ), + ], + ids=["absent", "config-only", "dual-field"], +) +def test_update_complete_requires_only_compact_society_wide_spm_provenance( + service, + orm_session_factory, + output, + error, +): + baseline = create_simulation( + orm_session_factory, + policy_id=1, + population_id="us", + population_type="geography", + ) + reform = create_simulation( + orm_session_factory, + policy_id=2, + population_id="us", + population_type="geography", + ) + report = service.create_or_reuse_report_output( + "us", baseline.id, reform.id, "2025" + ).view.report_output + + with pytest.raises(ValueError, match=error): + service.update_report_output("us", report.id, status="complete", output=output) + + with orm_session_factory() as session: + stored = session.get(ReportOutput, report.id) + assert stored.status == "pending" + assert stored.output is None + + +@pytest.mark.parametrize( + ("country_id", "population_type"), + [("us", "household"), ("uk", "geography")], + ids=["us-household", "non-us-economy"], +) +def test_update_complete_preserves_optional_spm_for_other_report_kinds( + service, + orm_session_factory, + country_id, + population_type, +): + simulation = create_simulation( + orm_session_factory, + country_id=country_id, + population_id="household-1" if population_type == "household" else country_id, + population_type=population_type, + ) + report = service.create_or_reuse_report_output( + country_id, simulation.id, year="2025" + ).view.report_output + + view = service.update_report_output( + country_id, + report.id, + status="complete", + output={"ok": True}, + ) + + assert view.report_output.output == {"ok": True} + + +def test_update_complete_rejects_legacy_society_wide_spm_output( + service, + orm_session_factory, +): + baseline = create_simulation( + orm_session_factory, + policy_id=1, + population_id="us", + population_type="geography", + ) + reform = create_simulation( + orm_session_factory, + policy_id=2, + population_id="us", + population_type="geography", + ) + report = service.create_or_reuse_report_output( + "us", baseline.id, reform.id, "2025" + ).view.report_output + legacy = compact_economy_output() + legacy["spm_provenance"] = { + "baseline": [SPM_RECEIPT], + "reform": [SPM_RECEIPT], + } + + with pytest.raises(ValueError, match="SPM"): + service.update_report_output("us", report.id, status="complete", output=legacy) + + with orm_session_factory() as session: + stored = session.get(ReportOutput, report.id) + assert stored.status == "pending" + assert stored.output is None + run = session.get(ReportOutputRun, stored.active_run_id) + assert run.status == "pending" + assert run.output is None + + +def test_resolve_legacy_spm_report_returns_new_empty_cache_record( + service, + orm_session_factory, + monkeypatch, +): + baseline = create_simulation( + orm_session_factory, + policy_id=1, + population_id="us", + population_type="geography", + ) + reform = create_simulation( + orm_session_factory, + policy_id=2, + population_id="us", + population_type="geography", + ) + monkeypatch.setitem(constants.RUNTIME_CACHE_SCHEMA_VERSIONS, "report_output", 2) + legacy_cache_version = get_report_output_cache_version("us") + monkeypatch.setitem(constants.RUNTIME_CACHE_SCHEMA_VERSIONS, "report_output", 3) + assert legacy_cache_version != get_report_output_cache_version("us") + with orm_session_factory.begin() as session: + legacy = ReportOutput( + country_id="us", + simulation_1_id=baseline.id, + simulation_2_id=reform.id, + api_version=legacy_cache_version, + status="complete", + output={ + "spm_config": SPM_CONFIG, + "spm_provenance": { + "baseline": [SPM_RECEIPT], + "reform": [SPM_RECEIPT], + }, + }, + year="2025", + ) + session.add(legacy) + session.flush() + legacy_id = legacy.id + + view = service.resolve_report_output("us", legacy_id) + + assert view.response_id == legacy_id + assert view.report_output.id != legacy_id + assert view.report_output.api_version == get_report_output_cache_version("us") + assert view.report_output.status == "pending" + assert view.report_output.output is None + + def test_update_running_requires_mutable_run(service, orm_session_factory): simulation = create_simulation(orm_session_factory) report = create_report(service, simulation.id) diff --git a/tests/unit/services/test_simulation_service.py b/tests/unit/services/test_simulation_service.py index bc5d4ef2e..464ae50b9 100644 --- a/tests/unit/services/test_simulation_service.py +++ b/tests/unit/services/test_simulation_service.py @@ -162,6 +162,95 @@ def test_update_simulation_accepts_legacy_json_text_at_wire_boundary(service): assert updated.output == {"result": 42} +def test_update_simulation_rejects_legacy_spm_config(service): + simulation = service.get_or_create_simulation( + "us", "household-1", "household", 1 + ).simulation + + with pytest.raises(ValueError, match="legacy spm_config"): + service.update_simulation( + "us", + simulation.id, + status="complete", + output={"spm_config": {"geography_kind": "national"}}, + ) + + +def test_update_simulation_rejects_incomplete_spm_provenance(service): + simulation = service.get_or_create_simulation( + "us", "household-1", "household", 1 + ).simulation + + with pytest.raises(ValueError, match="Field required"): + service.update_simulation( + "us", + simulation.id, + status="complete", + output={"spm_provenance": {"geography_kind": "national"}}, + ) + + +def test_update_simulation_persists_only_compact_spm_provenance(service): + simulation = service.get_or_create_simulation( + "us", "household-1", "household", 1 + ).simulation + provenance = { + "schema_version": "canonical-spm-provenance-v2", + "forecast_id": "spm-rolling-2026-09-09", + "forecast_sha256": "a" * 64, + "scenario": "ce_trend", + "geography_kind": "national", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, + "years": ["2026"], + "runtime_versions": { + "policyengine": "6.2.1", + "policyengine-core": "3.32.10", + "policyengine-us": "2.2.1", + "spm-calculator": "1.0.0", + }, + } + + updated = service.update_simulation( + "us", + simulation.id, + status="complete", + output={"result": 42, "spm_provenance": provenance}, + ) + + assert updated.output == { + "result": 42, + "spm_provenance": provenance, + } + + +def test_get_or_create_marks_previous_spm_output_for_recalculation( + service, + orm_session_factory, +): + simulation = service.get_or_create_simulation( + "us", "household-1", "household", 1 + ).simulation + with orm_session_factory.begin() as session: + stored = session.get(Simulation, simulation.id) + stored.status = "complete" + stored.output = { + "result": 42, + "spm_config": {"geography_kind": "national"}, + "spm_provenance": { + "years": {"2026": {"large": "diagnostics"}}, + }, + } + + refreshed = service.get_or_create_simulation( + "us", "household-1", "household", 1 + ).simulation + + assert refreshed.status == "pending" + assert refreshed.output is None + + def test_update_simulation_without_values_is_a_noop(service): simulation = service.get_or_create_simulation( "us", "household-1", "household", 1 diff --git a/tests/unit/services/test_spm_household_identity.py b/tests/unit/services/test_spm_household_identity.py index 2ed2572f1..661d3ec03 100644 --- a/tests/unit/services/test_spm_household_identity.py +++ b/tests/unit/services/test_spm_household_identity.py @@ -158,7 +158,8 @@ def test_replacement_household_preserves_historical_zero_padded_reference( ) assert replay.created is False assert replay.simulation.population_id == str(household.id).zfill(5) - assert replay.simulation.output == {"spm_config": NATIONAL} + assert replay.simulation.status == "pending" + assert replay.simulation.output is None saved = service.get_household("us", household.id) assert saved.household_json == original diff --git a/tests/unit/services/test_worker_spm.py b/tests/unit/services/test_worker_spm.py index 1bbcc17c7..e3b35f8d3 100644 --- a/tests/unit/services/test_worker_spm.py +++ b/tests/unit/services/test_worker_spm.py @@ -9,7 +9,7 @@ from policyengine_api.routes.economy_routes import economy_bp from policyengine_api.services.economy_service import EconomyService from policyengine_api.spm import SPMValidationError -from policyengine_api.worker_spm import validate_worker_spm +from policyengine_api.worker_spm import validate_worker_result, validate_worker_spm @pytest.mark.parametrize("selection", [None, {"geography_kind": "national"}]) @@ -174,6 +174,37 @@ def test_canonical_economy_errors_use_existing_validation_envelope(path): "as_of": None, } CAPABILITY = {"contract_version": "canonical-spm-v1", "defaults": SELECTION} +RUNTIME_VERSIONS = { + "policyengine": "6.2.1", + "policyengine-core": "3.32.10", + "policyengine-us": "2.2.1", + "spm-calculator": "1.0.0", +} + + +def worker_receipt(*, years=("2026",)): + return { + "schema_version": "canonical-spm-provenance-v2", + "forecast_id": "test-only", + "forecast_sha256": SELECTION["forecast_content_sha256"], + "scenario": SELECTION["scenario"], + "geography_kind": SELECTION["geography_kind"], + "geography_id": SELECTION["geography_id"], + "county_vintage": SELECTION["county_vintage"], + "as_of": SELECTION["as_of"], + "years": list(years), + "runtime_versions": RUNTIME_VERSIONS, + } + + +def worker_comparison(*, baseline=None, reform=None, execution_count=1): + baseline = baseline or worker_receipt() + reform = reform or worker_receipt() + return { + "schema_version": "canonical-spm-comparison-v2", + "baseline": {"receipt": baseline, "execution_count": execution_count}, + "reform": {"receipt": reform, "execution_count": execution_count}, + } def test_certified_worker_selection_enters_identity_and_submission(): @@ -200,6 +231,15 @@ def test_certified_worker_selection_enters_identity_and_submission(): ) != service._build_options_hash( options={"spm": changed}, model_version="test", dataset="default" ) + assert ( + "spm_provenance_schema=canonical-spm-comparison-v2" + in service._build_options_hash( + options={"spm": SELECTION}, model_version="test", dataset="default" + ) + ) + assert "spm_provenance_schema" not in service._build_options_hash( + options={}, model_version="test", dataset="default" + ) options = service._setup_sim_options("us", {}, {}, "us", "2026", spm=SELECTION) assert options.model_dump(mode="json")["spm"] == SELECTION @@ -403,31 +443,10 @@ def test_worker_cache_receipts_are_required_and_json_roundtrip(): import json from policyengine_api.worker_spm import validate_worker_result - receipt = dict( - forecast_id="test-only", - forecast_sha256=SELECTION["forecast_content_sha256"], - scenario="ce_trend", - geography_kind="national", - runtime_versions={}, - years={"2026": {}}, - geographies=[], - composition_method="classified-inputs", - storage_method="formula", - ) output = { - "spm_config": SELECTION, - "spm_provenance": {"baseline": [receipt], "reform": [receipt]}, + "spm_provenance": worker_comparison(), } validate_worker_result(json.loads(json.dumps(output)), SELECTION) - validate_worker_result( - { - **output, - "spm_config": { - key: value for key, value in SELECTION.items() if value is not None - }, - }, - SELECTION, - ) validate_worker_result( { "kind": "budgetWindow", @@ -439,7 +458,19 @@ def test_worker_cache_receipts_are_required_and_json_roundtrip(): for incomplete in ( {}, {"spm_config": SELECTION}, - {**output, "spm_config": {**SELECTION, "scenario": "zero_real"}}, + { + **output, + "spm_config": SELECTION, + }, + { + "spm_provenance": worker_comparison( + baseline={ + key: value + for key, value in worker_receipt().items() + if key != "scenario" + } + ) + }, ): with pytest.raises(SPMValidationError): validate_worker_result(incomplete, SELECTION) @@ -452,37 +483,55 @@ def test_worker_cache_receipts_are_required_and_json_roundtrip(): def test_worker_receipt_cannot_inherit_nonnull_settings(missing): from policyengine_api.worker_spm import validate_worker_result - config = {key: value for key, value in SELECTION.items() if key != missing} - with pytest.raises(SPMValidationError, match="incomplete resolved"): - validate_worker_result({"spm_config": config}, SELECTION) + receipt_field = ( + "forecast_sha256" if missing == "forecast_content_sha256" else missing + ) + receipt = { + key: value for key, value in worker_receipt().items() if key != receipt_field + } + with pytest.raises(SPMValidationError): + validate_worker_result( + {"spm_provenance": worker_comparison(baseline=receipt)}, SELECTION + ) -@pytest.mark.parametrize("side", ["baseline", "reform"]) -def test_worker_receipt_must_cover_requested_year(side): +def test_worker_receipt_must_cover_requested_year(): from policyengine_api.worker_spm import validate_worker_result - receipt = { - "forecast_id": "test-only", - "forecast_sha256": SELECTION["forecast_content_sha256"], - "scenario": "ce_trend", - "geography_kind": "national", - "runtime_versions": {}, - "years": {"2026": {}}, - "geographies": [], - "composition_method": "classified", - "storage_method": "formula", + result = { + "spm_provenance": worker_comparison(), + } + validate_worker_result(result, SELECTION, expected_year="2026") + result["spm_provenance"] = worker_comparison( + baseline=worker_receipt(years=("2025",)), + reform=worker_receipt(years=("2025",)), + ) + with pytest.raises(SPMValidationError, match="requested year"): + validate_worker_result(result, SELECTION, expected_year="2026") + + +def test_worker_receipt_rejects_baseline_reform_mismatch(): + result = { + "spm_provenance": worker_comparison( + reform=worker_receipt(years=("2025",)), + ), } + + with pytest.raises(SPMValidationError, match="baseline and reform"): + validate_worker_result(result, SELECTION, expected_year="2026") + + +def test_worker_receipt_rejects_removed_segment_list_shape(): result = { - "spm_config": SELECTION, "spm_provenance": { - "baseline": [dict(receipt)], - "reform": [dict(receipt)], + "baseline": [worker_receipt()], + "reform": [worker_receipt()], }, } - validate_worker_result(result, SELECTION, expected_year="2026") - result["spm_provenance"][side][0]["years"] = {"2025": {}} - with pytest.raises(SPMValidationError, match="requested year"): + + with pytest.raises(SPMValidationError) as caught: validate_worker_result(result, SELECTION, expected_year="2026") + assert caught.value.code == "SPM_CONFIGURATION_UNAVAILABLE" @pytest.mark.parametrize( @@ -565,21 +614,9 @@ def test_budget_result_cannot_nest_another_budget_window(): def annual_shape_result(year="2026"): - receipt = { - "forecast_id": "test-only", - "forecast_sha256": SELECTION["forecast_content_sha256"], - "scenario": SELECTION["scenario"], - "geography_kind": SELECTION["geography_kind"], - "runtime_versions": {}, - "years": {"2026": {}}, - "geographies": [], - "composition_method": "classified", - "storage_method": "formula", - } return { "year": year, - "spm_config": SELECTION, - "spm_provenance": {"baseline": [receipt], "reform": [receipt]}, + "spm_provenance": worker_comparison(), } @@ -613,17 +650,7 @@ def test_worker_valid_annual_and_window_shapes_cover_requested_year(year): PYDANTIC_INTERNALS = ("errors.pydantic.dev", "input_value", "validation error for") -WORKER_RECEIPT = { - "forecast_id": "test-artifact", - "forecast_sha256": SELECTION["forecast_content_sha256"], - "scenario": SELECTION["scenario"], - "geography_kind": SELECTION["geography_kind"], - "runtime_versions": {"policyengine-us": "test"}, - "years": {"2026": {}}, - "geographies": [], - "composition_method": "test composition", - "storage_method": "test storage", -} +WORKER_RECEIPT = worker_receipt() @pytest.mark.parametrize( @@ -667,11 +694,7 @@ def test_a_worker_receipt_failure_quotes_no_validator_internals(receipt): with pytest.raises(SPMValidationError) as caught: validate_worker_result( { - "spm_config": SELECTION, - "spm_provenance": { - "baseline": [WORKER_RECEIPT], - "reform": [receipt], - }, + "spm_provenance": worker_comparison(reform=receipt), }, SELECTION, expected_year="2026", diff --git a/tests/unit/test_country_spm.py b/tests/unit/test_country_spm.py index a36d9f8a0..4b525bb35 100644 --- a/tests/unit/test_country_spm.py +++ b/tests/unit/test_country_spm.py @@ -191,13 +191,25 @@ def __init__(self, *, tax_benefit_system, situation, spm): def test_tax_only_result_reads_provenance_without_calculating_spm(monkeypatch): calculations = [] - selection = spm.SPMSelection(geography_kind="national").model_dump() + selection = spm.SPMSelection( + forecast_content_sha256="a" * 64, + scenario="baseline", + geography_kind="national", + geography_id=None, + county_vintage="2020", + as_of=None, + ).model_dump() receipt = { "forecast_id": "test-artifact", "forecast_sha256": "a" * 64, "scenario": "baseline", "geography_kind": "national", - "runtime_versions": {"policyengine-us": "test"}, + "runtime_versions": { + "policyengine": "test", + "policyengine-core": "test", + "policyengine-us": "test", + "spm-calculator": "test", + }, "years": {}, "geographies": [], "composition_method": "test adult classification", @@ -224,12 +236,22 @@ def calculate(variable, period): ) result = country.calculate(requested_household("spm_unit_federal_tax"), None) assert calculations == ["spm_unit_federal_tax"] - assert result.spm_provenance["years"] == {} + assert result.spm_provenance["years"] == [] assert ( result.household["spm_units"]["spm_unit"]["spm_unit_federal_tax"]["2024"] == 123 ) - assert json.loads(json.dumps(result.spm_config)) == selection - assert json.loads(json.dumps(result.spm_provenance)) == receipt + assert result.spm_provenance == { + "schema_version": "canonical-spm-provenance-v2", + "forecast_id": "test-artifact", + "forecast_sha256": "a" * 64, + "scenario": "baseline", + "geography_kind": "national", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, + "years": [], + "runtime_versions": receipt["runtime_versions"], + } @pytest.fixture @@ -263,7 +285,7 @@ def test_real_country_state_only_tax_succeeds_and_receipt_stays_empty( result.household["spm_units"]["spm_unit"]["spm_unit_federal_tax"]["2024"] is not None ) - assert result.spm_provenance["years"] == {} + assert result.spm_provenance["years"] == [] json.dumps(result.spm_provenance) @@ -307,7 +329,7 @@ def test_real_country_state_only_dependency_is_null_when_nothing_was_chosen( """ result = real_canonical_country.calculate(requested_household(variable), None) assert result.household["spm_units"]["spm_unit"][variable]["2024"] is None - assert result.spm_config["geography_kind"] == "county" + assert result.spm_provenance["geography_kind"] == "county" @pytest.mark.parametrize("county", ["99999", "malformed"]) @@ -352,6 +374,7 @@ def test_real_country_explicit_geography_yields_json_receipt( resolved = spm.normalize_spm_selection("us", None) forecast = spm._selected_forecast(resolved["forecast_content_sha256"]) selection["geography_id"] = forecast.resolve_county(2024, "06037")["area_id"] + resolved_selection = spm.normalize_spm_selection("us", selection) original = deepcopy(household) result = real_canonical_country.calculate( household, None, spm=selection, spm_requested=True @@ -360,10 +383,10 @@ def test_real_country_explicit_geography_yields_json_receipt( assert ( result.household["spm_units"]["spm_unit"]["spm_unit_spm_threshold"]["2024"] > 0 ) - assert result.spm_config["geography_kind"] == geography + assert result.spm_provenance["geography_kind"] == geography assert ( result.spm_provenance["forecast_sha256"] - == result.spm_config["forecast_content_sha256"] + == resolved_selection["forecast_content_sha256"] ) assert "2024" in result.spm_provenance["years"] json.dumps(result.spm_provenance) @@ -422,7 +445,7 @@ def test_real_http_tax_only_succeeds_without_geography(real_http_client): json={"household": requested_household("spm_unit_federal_tax")}, ) assert response.status_code == 200, response.json - assert response.json["spm_provenance"]["years"] == {} + assert response.json["spm_provenance"]["years"] == [] @pytest.mark.parametrize( @@ -472,7 +495,7 @@ def test_real_http_missing_primitives_are_null_when_nothing_was_chosen( response = real_http_client.post("/us/calculate", json={"household": household}) assert response.status_code == 200, response.json assert response.json["status"] == "ok" - assert response.json["spm_config"]["geography_kind"] == "county" + assert response.json["spm_provenance"]["geography_kind"] == "county" if kind != "axes": threshold = response.json["result"]["spm_units"]["spm_unit"][ "spm_unit_spm_threshold" @@ -490,7 +513,7 @@ def test_real_http_national_result_and_receipt_survive_response_cache(real_http_ assert response.status_code == 200, response.json assert cached.status_code == 200 assert cached.json == response.json - assert response.json["spm_config"]["geography_kind"] == "national" + assert response.json["spm_provenance"]["geography_kind"] == "national" assert "2024" in response.json["spm_provenance"]["years"] @@ -511,7 +534,7 @@ def test_real_http_full_national_calculation_keeps_threshold_and_provenance( ] > 0 ) - assert response.json["spm_config"]["geography_kind"] == "national" + assert response.json["spm_provenance"]["geography_kind"] == "national" assert "2024" in response.json["spm_provenance"]["years"] @@ -592,7 +615,7 @@ def test_real_http_tax_only_outside_artifact_years_is_lazy( ] is not None ) - assert response.json["spm_provenance"]["years"] == {} + assert response.json["spm_provenance"]["years"] == [] @pytest.mark.parametrize("geography", ["national", "metro"]) diff --git a/tests/unit/test_openapi_simulation.py b/tests/unit/test_openapi_simulation.py index ff4fd6c38..3d9ca1f76 100644 --- a/tests/unit/test_openapi_simulation.py +++ b/tests/unit/test_openapi_simulation.py @@ -147,17 +147,22 @@ def test_simulation_persistence_responses_preserve_spm_output_and_run_fields( "status": "ok", "message": None, "result": {"people": {}}, - "spm_config": {"geography_kind": "national"}, "spm_provenance": { + "schema_version": "canonical-spm-provenance-v2", "forecast_id": "test-forecast", "forecast_sha256": "a" * 64, "scenario": "baseline", "geography_kind": "national", - "runtime_versions": {"policyengine-us": "test"}, - "years": {}, - "geographies": [], - "composition_method": "test", - "storage_method": "test", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, + "runtime_versions": { + "policyengine": "test", + "policyengine-core": "test", + "policyengine-us": "test", + "spm-calculator": "test", + }, + "years": ["2026"], }, } update_payload = { diff --git a/tests/unit/test_openapi_spm.py b/tests/unit/test_openapi_spm.py index 75a6e6d46..a7d709a60 100644 --- a/tests/unit/test_openapi_spm.py +++ b/tests/unit/test_openapi_spm.py @@ -5,7 +5,13 @@ from flask import Flask from policyengine_api.routes.system_routes import system_bp -from policyengine_api.spm import SPMProvenance, SPMSelection +from policyengine_api.spm import ( + SPMComparisonProvenance, + SPMExecutionProvenance, + SPMProvenance, + SPMRuntimeVersions, + SPMSelection, +) from policyengine_api.query_parameters import ( AnnualEconomyQuery, BudgetWindowEconomyQuery, @@ -46,19 +52,25 @@ def test_economy_query_and_receipt_schemas_match_public_http_contract(): ]["result"] if suffix: result = result["properties"]["annualImpacts"]["items"] - assert result["properties"]["spm_config"] == { - "$ref": "#/components/schemas/SPMSelection" - } + assert "spm_config" not in result["properties"] assert result["properties"]["spm_provenance"] == { - "$ref": "#/components/schemas/SPMWorkerProvenance" + "$ref": "#/components/schemas/SPMComparisonProvenance" } assert operation["responses"]["400"]["$ref"].endswith("SPMValidationError") - receipt = spec["components"]["schemas"]["SPMWorkerProvenance"] - assert receipt["required"] == ["baseline", "reform"] - for side in receipt["required"]: - assert receipt["properties"][side]["items"] == { - "$ref": "#/components/schemas/SPMProvenance" + comparison = spec["components"]["schemas"]["SPMComparisonProvenance"] + assert comparison["required"] == ["schema_version", "baseline", "reform"] + assert comparison["properties"]["schema_version"]["enum"] == [ + "canonical-spm-comparison-v2" + ] + for side in ("baseline", "reform"): + assert comparison["properties"][side] == { + "$ref": "#/components/schemas/SPMExecutionProvenance" } + execution = spec["components"]["schemas"]["SPMExecutionProvenance"] + assert execution["required"] == ["receipt", "execution_count"] + assert execution["properties"]["receipt"] == { + "$ref": "#/components/schemas/SPMProvenance" + } def test_metadata_discovery_and_selection_constraints_are_published(): @@ -106,10 +118,17 @@ def test_spm_documentation_uses_public_models_and_actual_route_envelopes(): for name, model in { "SPMSelection": SPMSelection, + "SPMRuntimeVersions": SPMRuntimeVersions, "SPMProvenance": SPMProvenance, + "SPMExecutionProvenance": SPMExecutionProvenance, + "SPMComparisonProvenance": SPMComparisonProvenance, }.items(): schema = schemas[name] - assert set(schema["properties"]) == set(model.model_fields) + expected_fields = { + field.alias or field_name + for field_name, field in model.model_fields.items() + } + assert set(schema["properties"]) == expected_fields assert schema["additionalProperties"] is False assert '"type": "null"' not in json.dumps(schema) @@ -119,12 +138,49 @@ def test_spm_documentation_uses_public_models_and_actual_route_envelopes(): for field in ("forecast_content_sha256", "scenario", "geography_id", "as_of"): assert selection[field]["type"] == "string" assert selection[field]["nullable"] is True - assert schemas["SPMProvenance"]["properties"]["runtime_versions"][ - "additionalProperties" - ] == { - "type": "string", - "nullable": True, + provenance = schemas["SPMProvenance"] + assert provenance["required"] == [ + "schema_version", + "forecast_id", + "forecast_sha256", + "scenario", + "geography_kind", + "geography_id", + "county_vintage", + "as_of", + "years", + "runtime_versions", + ] + assert provenance["properties"]["runtime_versions"] == { + "$ref": "#/components/schemas/SPMRuntimeVersions" + } + assert set(schemas["SPMRuntimeVersions"]["properties"]) == { + "policyengine", + "policyengine-core", + "policyengine-us", + "spm-calculator", + } + assert schemas["SPMRuntimeVersions"]["required"] == [ + "policyengine", + "policyengine-core", + "policyengine-us", + "spm-calculator", + ] + assert all( + "nullable" not in field + for field in schemas["SPMRuntimeVersions"]["properties"].values() + ) + assert "SPMResolvedConfiguration" not in schemas + removed_rich_fields = { + "geographies", + "requested", + "effective", + "execution_count", + "sources", + "diagnostics", + "mappings", } + assert not removed_rich_fields.intersection(provenance["properties"]) for route in ("calculate", "calculate-full"): operation = paths[f"/{{country_id}}/{route}"]["post"] @@ -146,7 +202,7 @@ def test_spm_documentation_uses_public_models_and_actual_route_envelopes(): "properties" ] assert {"status", "message", "result"} <= set(result) - assert result["spm_config"] == selection_ref + assert "spm_config" not in result assert result["spm_provenance"] == provenance_ref assert ( operation["responses"]["400"]["$ref"] diff --git a/tests/unit/test_simulation_and_report_routes.py b/tests/unit/test_simulation_and_report_routes.py index 2d2502f7c..23aa8c76f 100644 --- a/tests/unit/test_simulation_and_report_routes.py +++ b/tests/unit/test_simulation_and_report_routes.py @@ -27,15 +27,48 @@ def create_test_client() -> Flask: return app.test_client() -def create_simulation(factory, *, population_id="household-1", policy_id=1): +def create_simulation( + factory, + *, + population_id="household-1", + policy_id=1, + population_type="household", +): simulation = ( SimulationService(factory) - .get_or_create_simulation("us", population_id, "household", policy_id) + .get_or_create_simulation("us", population_id, population_type, policy_id) .simulation ) return simulation.id +def compact_spm_report_output() -> dict: + receipt = { + "schema_version": "canonical-spm-provenance-v2", + "forecast_id": "canonical-forecast", + "forecast_sha256": "a" * 64, + "scenario": "ce_trend", + "geography_kind": "national", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, + "years": ["2025"], + "runtime_versions": { + "policyengine": "6.2.1", + "policyengine-core": "3.32.10", + "policyengine-us": "2.2.1", + "spm-calculator": "1.0.0", + }, + } + return { + "spm_provenance": { + "schema_version": "canonical-spm-comparison-v2", + "baseline": {"receipt": receipt, "execution_count": 1}, + "reform": {"receipt": receipt, "execution_count": 1}, + }, + } + + def create_report(factory, simulation_id): report = ( ReportOutputService(factory) @@ -261,6 +294,45 @@ def test_report_patch_complete_promotes_active_rerun(orm_session_factory): assert report.latest_successful_run_id == rerun_id +def test_report_patch_rejects_legacy_spm_output_before_persistence( + orm_session_factory, +): + baseline_id = create_simulation( + orm_session_factory, + population_id="us", + policy_id=50, + population_type="geography", + ) + reform_id = create_simulation( + orm_session_factory, + population_id="us", + policy_id=51, + population_type="geography", + ) + report_id = ( + ReportOutputService(orm_session_factory) + .create_or_reuse_report_output("us", baseline_id, reform_id, "2025") + .view.report_output.id + ) + legacy = compact_spm_report_output() + receipt = legacy["spm_provenance"]["baseline"]["receipt"] + legacy["spm_provenance"] = { + "baseline": [receipt], + "reform": [receipt], + } + + response = create_test_client().patch( + "/us/report", + json={"id": report_id, "status": "complete", "output": legacy}, + ) + + assert response.status_code == 400 + with orm_session_factory() as session: + report = session.get(ReportOutput, report_id) + assert report.status == "pending" + assert report.output is None + + def test_simulation_v1_routes_keep_json_fields_as_strings(orm_session_factory): simulation_id = create_simulation(orm_session_factory, policy_id=49) output = {"result": "ok", "values": [1, 2, 3]} diff --git a/tests/unit/test_spm_provenance.py b/tests/unit/test_spm_provenance.py new file mode 100644 index 000000000..08521a1cf --- /dev/null +++ b/tests/unit/test_spm_provenance.py @@ -0,0 +1,232 @@ +"""Compact canonical SPM provenance contracts.""" + +import json + +import pytest +from pydantic import ValidationError + +from policyengine_api.spm import ( + SPMComparisonProvenance, + SPMProvenance, + SPMRuntimeVersions, + SPMSelection, + build_spm_calculation_provenance, + build_spm_comparison_provenance, + build_spm_provenance, + resolved_spm_configuration, + validate_spm_calculation_provenance, +) + + +SHA256 = "a" * 64 +SELECTION = SPMSelection.model_validate( + { + "forecast_content_sha256": SHA256, + "scenario": "ce_trend", + "geography_kind": "national", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, + } +) +# Synthetic fixture values used only to exercise the public receipt contract. +# Deployed workers obtain real package versions from the PolicyEngine.py bundle. +TEST_RUNTIME_VERSIONS = SPMRuntimeVersions.model_validate( + { + "policyengine": "0.0.0-test-policyengine", + "policyengine-core": "0.0.0-test-policyengine-core", + "policyengine-us": "0.0.0-test-policyengine-us", + "spm-calculator": "0.0.0-test-spm-calculator", + } +) + + +def _receipt(**changes: object) -> SPMProvenance: + values: dict[str, object] = { + "forecast_id": "spm-rolling-2026-09-09", + "forecast_sha256": SHA256, + "selection": SELECTION, + "years": ("2026",), + "runtime_versions": TEST_RUNTIME_VERSIONS, + } + values.update(changes) + return build_spm_provenance(**values) + + +def test_compact_receipt_has_only_the_public_v2_fields() -> None: + receipt = _receipt() + + assert receipt.model_dump(mode="json", by_alias=True) == { + "schema_version": "canonical-spm-provenance-v2", + "forecast_id": "spm-rolling-2026-09-09", + "forecast_sha256": SHA256, + "scenario": "ce_trend", + "geography_kind": "national", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, + "years": ["2026"], + "runtime_versions": { + "policyengine": "0.0.0-test-policyengine", + "policyengine-core": "0.0.0-test-policyengine-core", + "policyengine-us": "0.0.0-test-policyengine-us", + "spm-calculator": "0.0.0-test-spm-calculator", + }, + } + assert len(receipt.model_dump_json(by_alias=True).encode()) < 1_024 + + +@pytest.mark.parametrize( + "changes", + [ + {"forecast_sha256": "A" * 64}, + {"years": ("26",)}, + {"years": ("2027", "2026")}, + {"years": ("2026", "2026")}, + ], +) +def test_compact_receipt_rejects_noncanonical_values( + changes: dict[str, object], +) -> None: + with pytest.raises((ValidationError, ValueError)): + _receipt(**changes) + + +@pytest.mark.parametrize( + "invalid_version", + ["policyengine", "policyengine-core", "policyengine-us", "spm-calculator"], +) +@pytest.mark.parametrize("value", [None, ""]) +def test_runtime_versions_require_four_nonempty_strings( + invalid_version: str, + value: object, +) -> None: + versions = TEST_RUNTIME_VERSIONS.model_dump(mode="json", by_alias=True) + versions[invalid_version] = value + + with pytest.raises(ValidationError): + SPMRuntimeVersions.model_validate(versions) + + +def test_completed_calculation_contract_rejects_legacy_dual_field_shape() -> None: + receipt = _receipt().model_dump(mode="json", by_alias=True) + + with pytest.raises(ValidationError): + validate_spm_calculation_provenance( + { + "spm_config": SELECTION.model_dump(mode="json"), + "spm_provenance": receipt, + } + ) + + +def test_compact_receipt_rejects_the_previous_diagnostic_shape() -> None: + previous = { + "forecast_id": "spm-rolling-2026-09-09", + "forecast_sha256": SHA256, + "scenario": "ce_trend", + "geography_kind": "national", + "runtime_versions": {}, + "years": {"2026": {}}, + "geographies": [], + "composition_method": "classified inputs", + "storage_method": "formula", + } + + with pytest.raises(ValidationError): + SPMProvenance.model_validate(previous) + + +def test_country_receipt_adapter_extracts_only_compact_scalars() -> None: + country_receipt: dict[str, object] = { + "forecast_id": "spm-rolling-2026-09-09", + "forecast_sha256": SHA256, + "scenario": "ce_trend", + "geography_kind": "national", + "runtime_versions": TEST_RUNTIME_VERSIONS.model_dump( + mode="json", by_alias=True + ), + "years": {"2027": {"large": "diagnostic"}, "2026": {}}, + "geographies": [{"large": "mapping"}], + "composition_method": "classified inputs", + "storage_method": "formula", + } + + calculation = build_spm_calculation_provenance(SELECTION, country_receipt) + + assert resolved_spm_configuration(calculation).model_dump( + mode="json" + ) == SELECTION.model_dump(mode="json") + assert calculation.years == ("2026", "2027") + encoded = calculation.model_dump_json(by_alias=True) + assert "diagnostic" not in encoded + assert "mapping" not in encoded + assert "composition_method" not in encoded + + +def test_country_receipt_must_agree_with_adjacent_configuration() -> None: + country_receipt: dict[str, object] = { + "forecast_id": "spm-rolling-2026-09-09", + "forecast_sha256": SHA256, + "scenario": "different", + "geography_kind": "national", + "runtime_versions": TEST_RUNTIME_VERSIONS.model_dump( + mode="json", by_alias=True + ), + "years": {}, + } + + with pytest.raises(ValueError, match="differs from the SPM configuration"): + build_spm_calculation_provenance(SELECTION, country_receipt) + + +def test_comparison_collapses_identical_child_receipts() -> None: + receipt = _receipt() + + comparison = build_spm_comparison_provenance( + baseline_receipts=(receipt,) * 20, + reform_receipts=(receipt,) * 20, + ) + + assert comparison.baseline.receipt == receipt + assert comparison.baseline.execution_count == 20 + assert comparison.reform.receipt == receipt + assert comparison.reform.execution_count == 20 + assert len(comparison.model_dump_json(by_alias=True).encode()) < 2_048 + + +@pytest.mark.parametrize("side", ["baseline", "reform"]) +def test_comparison_rejects_inconsistent_child_receipts(side: str) -> None: + receipt = _receipt() + different = _receipt(years=("2027",)) + inputs = { + "baseline_receipts": (receipt,), + "reform_receipts": (receipt,), + } + inputs[f"{side}_receipts"] = (receipt, different) + + with pytest.raises(ValueError, match=f"{side} SPM receipts differ"): + build_spm_comparison_provenance(**inputs) + + +def test_comparison_rejects_previous_list_shape() -> None: + receipt = _receipt().model_dump(mode="json", by_alias=True) + + with pytest.raises(ValidationError): + SPMComparisonProvenance.model_validate( + {"baseline": [receipt], "reform": [receipt]} + ) + + +def test_compact_models_survive_json_round_trip() -> None: + comparison = build_spm_comparison_provenance( + baseline_receipts=(_receipt(),), + reform_receipts=(_receipt(),), + ) + + assert ( + SPMComparisonProvenance.model_validate_json( + json.dumps(comparison.model_dump(mode="json", by_alias=True)) + ) + == comparison + ) diff --git a/tests/unit/test_spm_settings.py b/tests/unit/test_spm_settings.py index 615fba9d2..c4aa91a04 100644 --- a/tests/unit/test_spm_settings.py +++ b/tests/unit/test_spm_settings.py @@ -403,13 +403,13 @@ def test_canonical_country_cannot_inherit_the_legacy_exception(monkeypatch, vers } -def test_resolved_settings_recover_receipt_omitted_null_values(certified_bundle): - """A receipt that omits its null values still names the same selection.""" +def test_resolved_settings_require_explicit_null_values(certified_bundle): + """A transported resolved config includes all six selection fields.""" assert spm.resolved_spm_settings(RESOLVED_SELECTION) == RESOLVED_SELECTION omitted = { key: value for key, value in RESOLVED_SELECTION.items() if value is not None } - assert spm.resolved_spm_settings(omitted) == RESOLVED_SELECTION + assert spm.resolved_spm_settings(omitted) is None assert spm.resolved_spm_settings(spm.normalize_spm_selection("us", None)) == ( RESOLVED_SELECTION ) @@ -432,17 +432,28 @@ def test_resolved_settings_reject_unreadable_receipt_settings(settings): assert spm.resolved_spm_settings(settings) is None -def test_receipt_keeps_the_countrys_own_omissions(certified_bundle): - """A published receipt repeats the country's wire shape, nulls and all.""" - omitted = { - key: value for key, value in RESOLVED_SELECTION.items() if value is not None - } +def test_receipt_reduces_country_diagnostics_to_one_compact_provenance( + certified_bundle, +): + """The public receipt itself contains every resolved configuration field.""" simulation = SimpleNamespace( - spm_config=omitted, spm_provenance=lambda: deepcopy(COUNTRY_RECEIPT) + spm_config=RESOLVED_SELECTION, + spm_provenance=lambda: deepcopy(COUNTRY_RECEIPT), ) receipt = spm.calculation_spm_receipt(simulation) - assert receipt["spm_config"] == omitted - assert receipt["spm_provenance"] == COUNTRY_RECEIPT + assert set(receipt) == {"spm_provenance"} + assert receipt["spm_provenance"] == { + "schema_version": "canonical-spm-provenance-v2", + "forecast_id": "test-artifact", + "forecast_sha256": ARTIFACT_HASH, + "scenario": "baseline", + "geography_kind": "county", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, + "years": [], + "runtime_versions": COUNTRY_RECEIPT["runtime_versions"], + } COUNTRY_RECEIPT = { @@ -450,7 +461,12 @@ def test_receipt_keeps_the_countrys_own_omissions(certified_bundle): "forecast_sha256": ARTIFACT_HASH, "scenario": "baseline", "geography_kind": "county", - "runtime_versions": {"policyengine-us": "test"}, + "runtime_versions": { + "policyengine": "test", + "policyengine-core": "test", + "policyengine-us": "test", + "spm-calculator": "test", + }, "years": {}, "geographies": [], "composition_method": "test composition", @@ -465,13 +481,9 @@ def test_receipt_keeps_the_countrys_own_omissions(certified_bundle): spm_config={**RESOLVED_SELECTION, "threshold_method": "canonical"}, spm_provenance=lambda: COUNTRY_RECEIPT, ), - SimpleNamespace( - spm_config=RESOLVED_SELECTION, - spm_provenance=lambda: {**COUNTRY_RECEIPT, "unreviewed_field": True}, - ), SimpleNamespace(spm_config=RESOLVED_SELECTION, spm_provenance=lambda: {}), ], - ids=["extra-setting", "extra-receipt-field", "incomplete-receipt"], + ids=["extra-setting", "incomplete-receipt"], ) def test_uncertifiable_country_receipt_is_typed_not_an_internal_failure(simulation): """An unreadable receipt is an uncertified country contract, never a 500.""" diff --git a/tests/unit/v2/test_stage12_worker_contracts.py b/tests/unit/v2/test_stage12_worker_contracts.py index df177f792..330af0483 100644 --- a/tests/unit/v2/test_stage12_worker_contracts.py +++ b/tests/unit/v2/test_stage12_worker_contracts.py @@ -27,6 +27,35 @@ DIGEST_C = "c" * 64 +def _calculation_provenance() -> dict[str, object]: + return { + "schema_version": "canonical-spm-provenance-v2", + "forecast_id": "canonical-forecast", + "forecast_sha256": DIGEST_A, + "scenario": "official", + "geography_kind": "national", + "geography_id": None, + "county_vintage": "2020", + "as_of": None, + "years": ["2026"], + "runtime_versions": { + "policyengine": "6.2.1", + "policyengine-core": "3.32.10", + "policyengine-us": "2.2.1", + "spm-calculator": "1.0.0", + }, + } + + +def _comparison_provenance() -> dict[str, object]: + receipt = _calculation_provenance() + return { + "schema_version": "canonical-spm-comparison-v2", + "baseline": {"receipt": receipt, "execution_count": 1}, + "reform": {"receipt": receipt, "execution_count": 1}, + } + + def _bundle() -> dict[str, object]: return { "policyengine_version": "5.2.0", @@ -173,10 +202,7 @@ def test_artifact_descriptors_retain_identity_and_bundle_provenance() -> None: "identity_sha256": DIGEST_B, }, "bundle": _bundle(), - "calculation_provenance": { - "spm_config": {"scenario": "official"}, - "spm_provenance": {"forecast_sha256": DIGEST_A}, - }, + "calculation_provenance": _calculation_provenance(), } ) report_artifact = AggregateReportArtifactDescriptor.model_validate( @@ -197,16 +223,58 @@ def test_artifact_descriptors_retain_identity_and_bundle_provenance() -> None: ) assert simulation_artifact.row_identity.identifier_columns == ("household_id",) - assert simulation_artifact.calculation_provenance == { - "spm_config": {"scenario": "official"}, - "spm_provenance": {"forecast_sha256": DIGEST_A}, - } + assert simulation_artifact.calculation_provenance is not None + assert ( + simulation_artifact.calculation_provenance.model_dump( + mode="json", by_alias=True + ) + == _calculation_provenance() + ) assert ( report_artifact.baseline_artifact_sha256 == simulation_artifact.artifact.content_sha256 ) +@pytest.mark.parametrize( + "provenance", + [ + {"scenario": "official"}, + { + "spm_config": {"scenario": "official"}, + "spm_provenance": _calculation_provenance(), + }, + { + **_calculation_provenance(), + "geographies": [], + }, + ], + ids=["partial", "legacy-dual-field", "legacy-rich"], +) +def test_simulation_artifact_rejects_noncompact_calculation_provenance( + provenance: dict[str, object], +) -> None: + descriptor = { + "contract_version": 1, + "evaluation_id": str(EVALUATION_ID), + "simulation_execution_id": str(BASELINE_ID), + "role": "baseline", + "artifact": {**_population_artifact(), "content_sha256": DIGEST_C}, + "output_schema_version": 1, + "row_identity": { + "schema_version": 1, + "identifier_columns": ["household_id"], + "row_count": 100, + "identity_sha256": DIGEST_B, + }, + "bundle": _bundle(), + "calculation_provenance": provenance, + } + + with pytest.raises(ValidationError): + SimulationArtifactDescriptor.model_validate(descriptor) + + def test_artifact_payload_contracts_cover_parquet_and_aggregate_json() -> None: assert SIMULATION_PARQUET_PAYLOAD_CONTRACT.model_dump(mode="json") == { "payload_schema_version": 1, @@ -230,7 +298,10 @@ def test_artifact_payload_contracts_cover_parquet_and_aggregate_json() -> None: "evaluation_id": str(EVALUATION_ID), "requested_aggregates": ["budget"], "bundle": _bundle(), - "result": {"budget": {"change": 100.0}}, + "result": { + "budget": {"change": 100.0}, + "spm_provenance": _comparison_provenance(), + }, } ) @@ -243,6 +314,19 @@ def test_artifact_payload_contracts_cover_parquet_and_aggregate_json() -> None: } ) + with pytest.raises(ValidationError, match="legacy spm_config"): + AggregateReportArtifactPayload.model_validate( + { + **payload.model_dump(mode="json"), + "result": { + **payload.result, + "spm_config": { + "forecast_content_sha256": DIGEST_A, + }, + }, + } + ) + def test_contract_models_are_immutable_and_reject_duplicate_lists() -> None: report = ReportExecutionInput.model_validate(_report())