Add the simulation report tables

docs/engineering/skills/typing-only-imports.md

@@ -81,7 +81,8 @@ misses breaks the model at runtime.
   with it (`policyengine_fastapi`, `policyengine_simulation_contract`,
   `policyengine_simulation_observability` and
   `policyengine_stage12_persistence`);
-- the entrypoint's, over `policyengine_simulation_entry`.
+- the entrypoint's, over `policyengine_simulation_entry` and
+  `policyengine_simulation_persistence`, which only the entrypoint installs.
 
 Each imports every module of its packages and checks every Pydantic model a
 module-level class statement defines:

libs/policyengine-simulation-persistence/README.md

@@ -62,6 +62,31 @@ Scheduler sweep re-enqueues running jobs nobody has checked for five minutes
 and fails jobs whose dispatch stalled for ten. Status reads use only these
 tables (`sim.persistence.status-reads-never-advance-work`).
 
+## Report tables
+
+`simulation.reports` records economy and household reports (`kind`). A report
+migrated from the Public API keeps its v1 report ID in `legacy_report_id`
+(unique) and starts `not_started` with no job. Once a job is attached the
+report is `submitted`, and its progress is that job's status, so the report
+and job never disagree; a check constraint ties the status to the presence of
+`job_id`.
+
+`simulation.report_results` holds one aggregate per finished report job,
+keyed by the job, so reports with identical inputs (one shared job) share one
+stored result. `JobStore.complete_report` validates the aggregate with the
+Pydantic model registered for its report kind and schema version in
+`REPORT_AGGREGATE_MODELS` (economy, version 1: the contract's
+`SingleYearMacroOutput`) before writing anything, then stores it as JSONB with
+`aggregate_schema_version`, `aggregate_sha256`, summary columns (budget
+impacts, households, baseline net income, model and data versions) and
+`artifacts`, a JSON array of validated `ReportArtifact` references. It
+finishes the job in the same transaction and, like every job write, only
+while the job is running.
+
+A job kind with `report_kind` set stores its result here; other kinds use
+`job_results`. No report job kind is registered and nothing creates report
+rows yet; household aggregates get a model when their contract exists.
+
 ## Database roles
 
 | Role | Used by | Privileges |

libs/policyengine-simulation-persistence/pyproject.toml

@@ -8,7 +8,10 @@ authors = [
 license = {file = "../../LICENSE"}
 requires-python = ">=3.13,<3.14"
 dependencies = [
+    # Report aggregates are validated by the contract's Pydantic models.
+    "policyengine-simulation-contract",
     "psycopg[binary]>=3.2,<4",
+    "pydantic>=2,<3",
     "sqlalchemy>=2,<3",
 ]
 
@@ -33,6 +36,10 @@ sources = ["src"]
 only-include = ["src/policyengine_simulation_persistence"]
 packages = ["src/policyengine_simulation_persistence"]
 
+[tool.uv.sources]
+policyengine-simulation-contract = { path = "../policyengine-simulation-contract", editable = true }
+policyengine-simulation-observability = { path = "../policyengine-simulation-observability", editable = true }
+
 [tool.pytest.ini_options]
 pythonpath = ["src"]
 testpaths = ["tests"]

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

@@ -0,0 +1,28 @@
+"""Canonical JSON hashing and the clock shared by the job and report stores."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import UTC, datetime
+
+
+def canonical_json(value: object) -> bytes:
+    """Encode ``value`` as JSON with sorted keys and no whitespace."""
+    return json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+        allow_nan=False,
+    ).encode("utf-8")
+
+
+def result_sha256(result: object) -> str:
+    """Return the SHA-256 of a result as canonical JSON."""
+    return hashlib.sha256(canonical_json(result)).hexdigest()
+
+
+def utc_now() -> datetime:
+    """Return the current time in UTC."""
+    return datetime.now(UTC)

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

@@ -18,9 +18,7 @@
 from __future__ import annotations
 
 import hashlib
-import json
 from dataclasses import dataclass, fields
-from datetime import UTC, datetime
 from enum import StrEnum
 from typing import TYPE_CHECKING, Any
 from uuid import UUID, uuid4
@@ -46,10 +44,23 @@
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.dialects.postgresql import insert as postgres_insert
 
+from policyengine_simulation_persistence.canonical import (
+    canonical_json,
+    result_sha256,
+    utc_now,
+)
+from policyengine_simulation_persistence.reports import (
+    ReportArtifact,
+    ReportKind,
+    ReportResultRecord,
+    report_result_values,
+    report_results,
+)
 from policyengine_simulation_persistence.schema import simulation_metadata
 
 if TYPE_CHECKING:
     from collections.abc import Callable
+    from datetime import datetime
 
     from sqlalchemy import Connection, Engine, RowMapping
 
@@ -131,24 +142,14 @@ class JobStatus(StrEnum):
 )
 
 
-def _canonical_json(value: object) -> bytes:
-    return json.dumps(
-        value,
-        sort_keys=True,
-        separators=(",", ":"),
-        ensure_ascii=False,
-        allow_nan=False,
-    ).encode("utf-8")
-
-
 def input_sha256(*, kind: str, policyengine_version: str, request: object) -> str:
     """Return the idempotency hash of one calculation's inputs.
 
     It covers the job kind, the policyengine.py bundle version and the request
     as canonical JSON, so the same request under another bundle is another job.
     """
     return hashlib.sha256(
-        _canonical_json(
+        canonical_json(
             {
                 "kind": kind,
                 "policyengine_version": policyengine_version,
@@ -158,16 +159,6 @@ def input_sha256(*, kind: str, policyengine_version: str, request: object) -> st
     ).hexdigest()
 
 
-def result_sha256(result: object) -> str:
-    """Return the SHA-256 of a result as canonical JSON."""
-    return hashlib.sha256(_canonical_json(result)).hexdigest()
-
-
-def utc_now() -> datetime:
-    """Return the current time in UTC."""
-    return datetime.now(UTC)
-
-
 @dataclass(frozen=True)
 class NewJob:
     """What a submission records before anything is dispatched."""
@@ -383,6 +374,18 @@ def get_result(self, job_id: UUID) -> JobResultRecord | None:
             )
         return None if row is None else JobResultRecord(**row)
 
+    def get_report_result(self, job_id: UUID) -> ReportResultRecord | None:
+        """Return the stored report result of a succeeded report job, or None."""
+        with self.engine.connect() as connection:
+            row = (
+                connection.execute(
+                    select(report_results).where(report_results.c.job_id == job_id)
+                )
+                .mappings()
+                .first()
+            )
+        return None if row is None else ReportResultRecord.from_row(row)
+
     # --- Collection ------------------------------------------------------------
 
     def record_check(self, job_id: UUID) -> int | None:
@@ -400,6 +403,27 @@ def record_check(self, job_id: UUID) -> int | None:
                 .returning(jobs.c.collection_attempts)
             ).scalar()
 
+    def _finish_succeeded(
+        self,
+        connection: Connection,
+        job_id: UUID,
+        now: datetime,
+    ) -> bool:
+        return (
+            connection.execute(
+                update(jobs)
+                .where(jobs.c.id == job_id, jobs.c.status == JobStatus.RUNNING.value)
+                .values(
+                    status=JobStatus.SUCCEEDED.value,
+                    completed_at=now,
+                    last_checked_at=now,
+                    updated_at=now,
+                )
+                .returning(jobs.c.id)
+            ).first()
+            is not None
+        )
+
     def _complete(
         self,
         connection: Connection,
@@ -409,18 +433,7 @@ def _complete(
         result_schema_version: int,
     ) -> bool:
         now = self._clock()
-        finished = connection.execute(
-            update(jobs)
-            .where(jobs.c.id == job_id, jobs.c.status == JobStatus.RUNNING.value)
-            .values(
-                status=JobStatus.SUCCEEDED.value,
-                completed_at=now,
-                last_checked_at=now,
-                updated_at=now,
-            )
-            .returning(jobs.c.id)
-        ).first()
-        if finished is None:
+        if not self._finish_succeeded(connection, job_id, now):
             return False
         connection.execute(
             insert(job_results).values(
@@ -453,6 +466,37 @@ def complete(
                 result_schema_version=result_schema_version,
             )
 
+    def complete_report(
+        self,
+        job_id: UUID,
+        *,
+        report_kind: ReportKind,
+        aggregate: object,
+        aggregate_schema_version: int,
+        artifacts: tuple[ReportArtifact, ...] = (),
+    ) -> bool:
+        """Store a running report job's aggregate and mark the job succeeded.
+
+        The aggregate is validated by the model registered for its report kind
+        and schema version before anything is written; an invalid aggregate
+        raises and leaves the job running. Return False, writing nothing, when
+        another collector already finished the job.
+        """
+        values = report_result_values(
+            report_kind=report_kind,
+            aggregate=aggregate,
+            aggregate_schema_version=aggregate_schema_version,
+            artifacts=artifacts,
+        )
+        with self.engine.begin() as connection:
+            now = self._clock()
+            if not self._finish_succeeded(connection, job_id, now):
+                return False
+            connection.execute(
+                insert(report_results).values(job_id=job_id, created_at=now, **values)
+            )
+            return True
+
     def _finish_failed(
         self,
         connection: Connection,