Create household reports through POST /reports and calculate them in one household function call

changelog.d/c1-15-household-reports.added.md

@@ -0,0 +1 @@
+The Simulation Entrypoint creates household reports through `POST /reports` with `report_type` `household`: one stored household, for its default year, under a baseline and an optional reform, each a new household simulation run. One household function call calculates the report run, writes each simulation run's `household_result` and finishes the report run.

docs/migration/stage-12-modal-worker-foundation.md

@@ -719,11 +719,12 @@ computed because no file existed.
 ## v2 household functions
 
 The versioned v2 application also defines a household function for each
-country, `calculate_household_us` and `calculate_household_uk`. Each call
-calculates one household, in the Public API's situation shape and with
+country, `calculate_household_us` and `calculate_household_uk`. A household
+calculation is one household, in the Public API's situation shape and with
 optional axes, for one year under one policy, as a household job or a
-household simulation run is one household, one year and one policy; a
-baseline and its reform are two calls. It takes a `HouseholdExecutionRequest`
+household simulation run is one household, one year and one policy. A call
+for a household job, or for deployment validation, makes one calculation:
+it takes a `HouseholdExecutionRequest`
 and returns a `HouseholdExecutionResult`, which holds the calculated
 household with its warnings and SPM receipt, or a typed
 `HouseholdExecutionFailure`, from
@@ -739,8 +740,8 @@ year. Otherwise the
 calculation follows the Public API's household calculation, including its
 input errors, warnings and SPM receipt. A function refuses a request for
 another country, and returns a `bundle_mismatch` failure for a request that
-names another bundle. Nothing calls these functions until the Simulation
-Entrypoint has household routes.
+names another bundle. The Simulation Entrypoint calls them for
+`/household-calculations` and for household reports on `/reports`.
 
 A call can name a household job, a `household_jobs` row the Simulation
 Entrypoint inserts pending, with the `household_job_id` keyword argument. As
@@ -759,9 +760,40 @@ only columns the role may update: `status`, `modal_call_id`,
 `error_message`, `result`, `started_at`, `completed_at` and `updated_at`. A
 call that names no job, as deployment validation makes, touches no database.
 
+A call can instead name a household report run, with the
+`household_report_run_id` keyword argument, and then takes a
+`HouseholdReportRunCalculation`
+(`policyengine_simulation_contract.household_report_execution`): the
+baseline's `HouseholdExecutionRequest` and, optionally, the reform's, each
+request's calculation ID naming its simulation run. A call names a job or a
+report run, never both. The function then:
+
+1. claims the report run under its own Modal call ID, as a report run's
+   coordinator does, and reads which baseline and reform simulation runs it
+   names;
+2. validates the payload and checks that its runs are those;
+3. for each side, claims the side's simulation run, calculates it, and
+   writes its `HouseholdExecutionResult`, or its `HouseholdExecutionFailure`
+   and one-line summary, to the run's `household_result` and
+   `error_message`;
+4. marks the report run succeeded only if every side's run did, or fails it
+   with the first such side's reason.
+
+Both sides are always calculated, so each run has its own outcome. A report
+run another call holds, or one already finished, returns `deduplicated`, the
+report run's ID and its status, without calculating. A restarted call keeps a
+side it already recorded. An error, a refused payload included, or Modal's
+cancellation fails the report run and each of its simulation runs left
+unfinished, with a summary naming only the error's type, then propagates; a
+pre-emption records nothing. An error in the report run's claim, or in the
+read of its simulation runs, fails the report run if the claim applied
+before the error, and leaves its simulation runs to the stuck-run sweep.
+
 The stuck-run sweep (`src/modal/run_sweep_app.py`) fails household jobs
 nothing will finish by the rule it applies to report runs and simulation
-runs, counting a job's age from its creation: a pending job with no call
+runs (a household report run and its simulation runs are judged as any
+report run and its simulation runs are), counting a job's age from its
+creation: a pending job with no call
 recorded ten minutes after its creation, a pending or running job once
 Modal reports its recorded call ended, and any job still unfinished 24 hours
 after its creation, whatever Modal reports.
@@ -771,7 +803,7 @@ after its creation, whatever Modal reports.
   cache. Building it needs no secret.
 - Secrets: only `policyengine-modal-worker-database`, whose
   `MODAL_WORKER_DATABASE_URL` connects as the Modal worker role, to record
-  the household job a call names. Like the coordinator and the simulation
+  the household job or household report run a call names. Like the coordinator and the simulation
   workers, the application cannot deploy to a Modal environment without it.
   A household calculation reads no private data or Cloud Storage. Telemetry
   authenticates with the identity token Modal gives every container

libs/policyengine-simulation-contract/src/policyengine_simulation_contract/household_report_execution.py

@@ -0,0 +1,72 @@
+"""The calculations of one household report run, as one household function takes them.
+
+A household report calculates one stored household under a baseline policy
+and, optionally, a reform. The simulation service creates a household
+simulation run for each side, and a report run naming them, then starts one
+household function call with the report run's ID and this document. Each
+side is a ``HouseholdExecutionRequest`` whose ``calculation_id`` is the
+simulation run it records; both sides calculate the same household, for the
+same year, the same way, on the same bundle, and differ only in their
+policy.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import model_validator
+
+from policyengine_simulation_contract.household_execution import (
+    HOUSEHOLD_CONTRACT_SCHEMA_VERSION,
+    HouseholdExecutionRequest,
+    StrictHouseholdModel,
+)
+
+# Which side of a household report a calculation is.
+HouseholdReportSide = Literal["baseline", "reform"]
+# The request fields both sides share: everything but the policy and the
+# simulation run each side records.
+_SHARED_FIELDS: set[str] = set(HouseholdExecutionRequest.model_fields) - {
+    "calculation_id",
+    "policy",
+}
+
+
+class HouseholdReportRunCalculation(StrictHouseholdModel):
+    """A household report run's baseline and, optionally, reform calculation.
+
+    ``reform`` is None for a report of the baseline alone.
+    """
+
+    schema_version: Literal[1] = HOUSEHOLD_CONTRACT_SCHEMA_VERSION
+    baseline: HouseholdExecutionRequest
+    reform: HouseholdExecutionRequest | None = None
+
+    @model_validator(mode="after")
+    def validate_one_household(self) -> HouseholdReportRunCalculation:
+        """Require two sides to record different runs of one household calculation."""
+        if self.reform is None:
+            return self
+        if self.reform.calculation_id == self.baseline.calculation_id:
+            msg = "the baseline and the reform must record different simulation runs"
+            raise ValueError(msg)
+        baseline = self.baseline.model_dump(include=_SHARED_FIELDS)
+        reform = self.reform.model_dump(include=_SHARED_FIELDS)
+        differing = sorted(
+            name for name in _SHARED_FIELDS if baseline.get(name) != reform.get(name)
+        )
+        if differing:
+            msg = (
+                "the baseline and the reform must calculate the same household "
+                f"the same way; they differ in {', '.join(differing)}"
+            )
+            raise ValueError(msg)
+        return self
+
+    def sides(
+        self,
+    ) -> tuple[tuple[HouseholdReportSide, HouseholdExecutionRequest], ...]:
+        """Return each side's request, the baseline's first."""
+        if self.reform is None:
+            return (("baseline", self.baseline),)
+        return (("baseline", self.baseline), ("reform", self.reform))

libs/policyengine-simulation-contract/tests/test_household_report_execution.py

@@ -0,0 +1,125 @@
+"""Tests for the calculations of one household report run, as one call takes them."""
+
+from __future__ import annotations
+
+from uuid import UUID
+
+import pytest
+from policyengine_simulation_contract.household_execution import (
+    HouseholdBundleProvenance,
+    HouseholdExecutionRequest,
+)
+from policyengine_simulation_contract.household_report_execution import (
+    HouseholdReportRunCalculation,
+)
+from pydantic import ValidationError
+
+BASELINE_RUN_ID = UUID("00000000-0000-4000-8000-0000000000b1")
+REFORM_RUN_ID = UUID("00000000-0000-4000-8000-0000000000b2")
+HOUSEHOLD = {
+    "people": {"you": {"age": {"2026": 40}, "employment_income": {"2026": 50_000}}},
+    "households": {
+        "your household": {
+            "members": ["you"],
+            "household_net_income": {"2026": None},
+        }
+    },
+}
+BUNDLE = HouseholdBundleProvenance(
+    policyengine_version="6.17.1",
+    core_package_version="3.33.0",
+    country_package_name="policyengine-us",
+    country_package_version="1.764.6",
+    bundle_manifest_sha256="a" * 64,
+)
+CTC_REFORM = {
+    "gov.irs.credits.ctc.amount.base[0].amount": {"2026-01-01.2200-12-31": 3_000}
+}
+
+
+def _request(calculation_id: UUID, **changes: object) -> dict[str, object]:
+    """Return one side's household function request as JSON."""
+    return {
+        "calculation_id": str(calculation_id),
+        "country": "us",
+        "year": "2026",
+        "household": HOUSEHOLD,
+        "policy": {"parameters": {}},
+        "add_missing_variables": True,
+        "bundle": BUNDLE.model_dump(mode="json"),
+        **changes,
+    }
+
+
+def _calculation(
+    *, reform: dict[str, object] | None = None
+) -> HouseholdReportRunCalculation:
+    return HouseholdReportRunCalculation.model_validate(
+        {"baseline": _request(BASELINE_RUN_ID), "reform": reform}
+    )
+
+
+def test_a_report_run_calculates_its_baseline_then_its_reform() -> None:
+    calculation = _calculation(
+        reform=_request(REFORM_RUN_ID, policy={"parameters": CTC_REFORM})
+    )
+
+    sides = calculation.sides()
+
+    assert [side for side, _ in sides] == ["baseline", "reform"]
+    assert [request.calculation_id for _, request in sides] == [
+        BASELINE_RUN_ID,
+        REFORM_RUN_ID,
+    ]
+    assert all(isinstance(request, HouseholdExecutionRequest) for _, request in sides)
+    assert sides[1][1].policy.parameters == CTC_REFORM
+
+
+def test_a_report_run_of_the_baseline_alone_calculates_one_side() -> None:
+    calculation = _calculation()
+
+    assert calculation.reform is None
+    assert [side for side, _ in calculation.sides()] == ["baseline"]
+
+
+def test_a_calculation_round_trips_through_json() -> None:
+    calculation = _calculation(reform=_request(REFORM_RUN_ID))
+
+    payload = calculation.model_dump(mode="json")
+
+    assert payload["schema_version"] == 1
+    assert HouseholdReportRunCalculation.model_validate(payload) == calculation
+
+
+def test_each_side_is_its_own_calculation() -> None:
+    with pytest.raises(ValidationError, match="different simulation runs"):
+        _calculation(reform=_request(BASELINE_RUN_ID))
+
+
+@pytest.mark.parametrize(
+    "reform_change",
+    [
+        {"year": "2027"},
+        {"household": {"people": {"you": {"age": {"2026": 41}}}}},
+        {"add_missing_variables": False},
+        {
+            "bundle": {
+                **BUNDLE.model_dump(mode="json"),
+                "bundle_manifest_sha256": "b" * 64,
+            }
+        },
+    ],
+    ids=["year", "household", "add_missing_variables", "bundle"],
+)
+def test_both_sides_calculate_one_household_the_same_way(
+    reform_change: dict[str, object],
+) -> None:
+    with pytest.raises(ValidationError, match="the same household"):
+        _calculation(reform=_request(REFORM_RUN_ID, **reform_change))
+
+
+def test_unknown_fields_are_refused() -> None:
+    with pytest.raises(ValidationError, match="extra"):
+        HouseholdReportRunCalculation.model_validate(
+            {"baseline": _request(BASELINE_RUN_ID), "report_run_id": "x"}
+        )

libs/policyengine-simulation-persistence/README.md

@@ -125,6 +125,29 @@ commit fails after its rows were inserted raises
 `EconomyRecordOutcomeUnknownError`: PostgreSQL may have applied the commit,
 so the rows may exist.
 
+## Household reports
+
+`household_report_records.py` holds the simulation service's writes for a
+household report: one stored household, for one year, under a baseline and,
+optionally, a reform. `insert_household_report` creates, in the caller's
+transaction, a household simulation (no dataset or region) and its first
+run for each side, never reusing one; the report, of type `household`,
+naming the household, both simulations and the reform's policy, with its
+options in `inputs`; and its first report run, trigger `initial`, naming no
+reform run for a report of the baseline alone. Every run is `pending` with
+no Modal call ID, and a row naming one that does not exist is refused as
+`EconomyRecordReferenceError`.
+
+`HouseholdReportStore` runs the steps in transactions of their own:
+`read_inputs` reads the catalog version for the .py version, the country's
+model, the stored household and each side's policy values together,
+raising `UnknownCatalogVersionError`, `UnknownHouseholdError` or
+`ReportSideError` for the side whose policy is missing or sets parameters
+the catalog version lacks; `create_report` raises
+`EconomyRecordOutcomeUnknownError` when the commit fails after the rows were
+inserted; and `record_report_call` and `fail_unstarted_report` use the
+economy report run writes.
+
 ## Reading a policy
 
 `read_policy_reform` (`policy_reform.py`) reads a stored policy as the reform

libs/policyengine-simulation-persistence/src/policyengine_simulation_persistence/__init__.py

@@ -81,6 +81,16 @@
     UnknownCatalogVersionError,
     UnknownHouseholdError,
 )
+from policyengine_simulation_persistence.household_report_records import (
+    HOUSEHOLD_REPORT_TYPE,
+    HouseholdReportSimulation,
+    NewHouseholdReport,
+    insert_household_report,
+)
+from policyengine_simulation_persistence.household_report_store import (
+    HouseholdReportInputs,
+    HouseholdReportStore,
+)
 from policyengine_simulation_persistence.input_tables import input_metadata
 from policyengine_simulation_persistence.policy_reform import (
     MissingPolicyParametersError,
@@ -111,6 +121,7 @@
 )
 
 __all__ = [
+    "HOUSEHOLD_REPORT_TYPE",
     "LATEST_YEAR",
     "MAX_BUDGET_WINDOW_YEARS",
     "SERVICE_TABLE_PRIVILEGES",
@@ -134,11 +145,15 @@
     "HouseholdJobReferenceError",
     "HouseholdJobStatus",
     "HouseholdJobStore",
+    "HouseholdReportInputs",
+    "HouseholdReportSimulation",
+    "HouseholdReportStore",
     "MissingPolicyParametersError",
     "NewBudgetWindowReport",
     "NewEconomyReport",
     "NewEconomySimulation",
     "NewHouseholdJob",
+    "NewHouseholdReport",
     "OverlappingReformError",
     "PolicyReform",
     "ReportRunState",
@@ -177,6 +192,7 @@
     "insert_economy_report",
     "insert_economy_simulation",
     "insert_household_job",
+    "insert_household_report",
     "missing_table_privileges",
     "normalized_database_url",
     "read_catalog_version_id",