Add the simulation job store and background result collection through Cloud Tasks

.github/scripts/deploy-cloud-run-simulation-entry.sh

@@ -78,6 +78,9 @@ required_variables=(
   OLD_GATEWAY_AUTH_CLIENT_ID_VALUE
   OLD_GATEWAY_AUTH_CLIENT_SECRET_SECRET_NAME
   SIMULATION_DATABASE_URL_SECRET_NAME
+  RECONCILIATION_QUEUE_VALUE
+  RECONCILIATION_INVOKER_SERVICE_ACCOUNT_VALUE
+  ENTRYPOINT_STABLE_URL
   OBSERVABILITY_SERVICE_NAMESPACE
   OBSERVABILITY_TRACE_PROJECT_ID
   OTEL_EXPORTER_OTLP_ENDPOINT
@@ -110,6 +113,15 @@ esac
   || fail "ENTRYPOINT_MAX_INSTANCES must be a non-negative integer"
 ((10#${ENTRYPOINT_MAX_INSTANCES} >= 10#${ENTRYPOINT_MIN_INSTANCES})) \
   || fail "ENTRYPOINT_MAX_INSTANCES must be at least ENTRYPOINT_MIN_INSTANCES"
+# Background collection of simulation job results: Cloud Tasks calls the
+# stable service URL (never the candidate tag) with an OIDC token of the
+# invoker service account, which Cloud Scheduler also uses for the sweep.
+[[ "${RECONCILIATION_QUEUE_VALUE}" =~ ^projects/[a-z][a-z0-9-]{4,28}[a-z0-9]/locations/[a-z0-9-]+/queues/[A-Za-z0-9-]{1,100}$ ]] \
+  || fail "RECONCILIATION_QUEUE_VALUE must be a full Cloud Tasks queue name"
+[[ "${RECONCILIATION_INVOKER_SERVICE_ACCOUNT_VALUE}" =~ ^[a-z][a-z0-9-]{4,28}[a-z0-9]@[a-z][a-z0-9-]{4,28}[a-z0-9]\.iam\.gserviceaccount\.com$ ]] \
+  || fail "RECONCILIATION_INVOKER_SERVICE_ACCOUNT_VALUE must be a service account email"
+[[ "${ENTRYPOINT_STABLE_URL}" =~ ^https://[a-z0-9.-]+$ ]] \
+  || fail "ENTRYPOINT_STABLE_URL must be the service's HTTPS URL without a path"
 # An unset GitHub variable arrives as an empty value. Default it to log-only
 # permission checks; switching to 1 is a separate operator decision.
 export SIMULATION_ENTRYPOINT_ENFORCE_PERMISSIONS_VALUE="${SIMULATION_ENTRYPOINT_ENFORCE_PERMISSIONS_VALUE:-0}"
@@ -173,6 +185,10 @@ jq -n '
     OTEL_METRICS_EXPORTER: "otlp",
     OTEL_TRACES_SAMPLER_ARG: "1.0",
     POLICYENGINE_OTEL_GOOGLE_AUDIENCE: env.POLICYENGINE_OTEL_GOOGLE_AUDIENCE,
+    RECONCILIATION_QUEUE: env.RECONCILIATION_QUEUE_VALUE,
+    RECONCILIATION_INVOKER_SERVICE_ACCOUNT: env.RECONCILIATION_INVOKER_SERVICE_ACCOUNT_VALUE,
+    RECONCILIATION_TARGET_URL: env.ENTRYPOINT_STABLE_URL,
+    SIMULATION_MODAL_ENVIRONMENT: env.MODAL_ENVIRONMENT,
     STAGE12_ENABLED: (
       if env.DEPLOY_STAGE12_V2 == "true" then env.STAGE12_ENABLED_VALUE else "0" end
     )

.github/workflows/simulation-deploy.reusable.yml

@@ -265,6 +265,9 @@ jobs:
           OLD_GATEWAY_AUTH_CLIENT_ID_VALUE: ${{ secrets.OLD_GATEWAY_AUTH_CLIENT_ID }}
           OLD_GATEWAY_AUTH_CLIENT_SECRET_SECRET_NAME: ${{ vars.OLD_GATEWAY_AUTH_CLIENT_SECRET_SECRET_NAME }}
           SIMULATION_DATABASE_URL_SECRET_NAME: ${{ vars.SIMULATION_DATABASE_URL_SECRET_NAME }}
+          RECONCILIATION_QUEUE_VALUE: ${{ vars.RECONCILIATION_QUEUE }}
+          RECONCILIATION_INVOKER_SERVICE_ACCOUNT_VALUE: ${{ vars.RECONCILIATION_INVOKER_SERVICE_ACCOUNT }}
+          ENTRYPOINT_STABLE_URL: ${{ steps.previous.outputs.stable_url }}
           STAGE12_ENABLED_VALUE: ${{ vars.STAGE12_ENABLED }}
           STAGE12_ARTIFACT_BUCKET_VALUE: ${{ vars.STAGE12_ARTIFACT_BUCKET }}
           STAGE12_DATABASE_URL_SECRET_NAME: ${{ vars.STAGE12_DATABASE_URL_SECRET_NAME }}

docs/generated/rules.md

@@ -8,11 +8,11 @@ Every rule that governs `policyengine-sim-api` is listed here with how it is enf
 
 | Tier | enforced | ratcheted | advisory | doc-only | Total |
 | --- | ---: | ---: | ---: | ---: | ---: |
-| mechanical | 43 | 1 | 0 | 14 | 58 |
+| mechanical | 46 | 1 | 0 | 14 | 61 |
 | metric | 0 | 0 | 0 | 0 | 0 |
 | judgment | 0 | 0 | 0 | 0 | 0 |
 
-**Blocking:** 44 of 58 rules.
+**Blocking:** 47 of 61 rules.
 
 ## Gaps
 
@@ -54,8 +54,11 @@ Every rule that governs `policyengine-sim-api` is listed here with how it is enf
 - `sim.deploy.smoke-tests-strict`
 - `sim.deploy.traffic-changes-exact`
 - `sim.deploy.workflow-shell-safety`
+- `sim.persistence.conditional-job-writes`
+- `sim.persistence.internal-routes-google-oidc`
 - `sim.persistence.migrate-before-entrypoint`
 - `sim.persistence.migration-cycle`
+- `sim.persistence.status-reads-never-advance-work`
 - `sim.process.rules-registry`
 - `sim.stage12.deployment-scope`
 - `sim.stage12.deployment-uses-api-schema`
@@ -125,32 +128,47 @@ flowchart LR
 
 | Rule | Statement | Tier | Status | Blocking | Enforced by | CI | Proof |
 | --- | --- | --- | --- | --- | --- | --- | --- |
+| `sim.persistence.conditional-job-writes` | Every simulation job transition and result write is one conditional update on the job's expected status (and submission an insert that does nothing on a conflicting input hash), so concurrent collectors, dispatchers or submitters change a job at most once; no lease is taken. | mechanical | enforced | yes | pytest `libs/policyengine-simulation-persistence/tests/test_job_store_concurrency.py` | `pr.yml:simulation-database` | **unproven** |
 | `sim.persistence.executor-independent` | The Modal executor and gateway never declare, lock or import policyengine-simulation-persistence; only the Simulation Entrypoint and the migration job install it. | mechanical | enforced | yes | pytest `libs/policyengine-simulation-persistence/tests/test_ownership_boundaries.py` | `pr.yml:test` | yes |
+| `sim.persistence.internal-routes-google-oidc` | The simulation job collection and sweep routes accept only Google-signed OIDC tokens whose audience is the service URL and whose email is the configured invoker service account. | mechanical | enforced | yes | pytest `projects/policyengine-simulation-entry/tests/test_google_oidc.py`, `projects/policyengine-simulation-entry/tests/test_simulation_job_routes.py` | `pr.yml:test` | **unproven** |
 | `sim.persistence.migrate-before-entrypoint` | Each deployment applies the simulation chain in its own job, the only one that receives SIMULATION_MIGRATION_DATABASE_URL, before the entrypoint candidate deploys; the entrypoint receives SIMULATION_DATABASE_URL as a pinned Secret Manager version; and the candidate smoke requires the database at the image's revision. | mechanical | enforced | yes | pytest `projects/policyengine-simulation-entry/tests/test_simulation_database_deployment.py` | `pr.yml:test` | **unproven** |
 | `sim.persistence.migration-cycle` | On a fresh PostgreSQL, the simulation chain upgrades to its head, passes alembic check, downgrades to base and upgrades again, run by a non-superuser migrator, and the runtime role can read the revision but cannot change the schema. | mechanical | enforced | yes | pytest `libs/policyengine-simulation-persistence/tests/test_migration_cycle.py` | `pr.yml:simulation-database` | **unproven** |
 | `sim.persistence.no-modal-database-credentials` | No Modal secret, Modal deployment or Modal application receives SIMULATION_DATABASE_URL or SIMULATION_MIGRATION_DATABASE_URL; the Stage 12 comparison's own database secret is the only database credential Modal holds. | mechanical | enforced | yes | pytest `libs/policyengine-simulation-persistence/tests/test_ownership_boundaries.py` | `pr.yml:test` | yes |
 | `sim.persistence.one-migration-chain` | The simulation schema changes only through the one Alembic chain at migrations/ (configured by migrations/alembic.ini, versioned in simulation.alembic_version); no other Alembic configuration, environment or revision directory exists, and runtime source code contains no schema DDL. | mechanical | enforced | yes | pytest `libs/policyengine-simulation-persistence/tests/test_ownership_boundaries.py` | `pr.yml:test` | yes |
+| `sim.persistence.status-reads-never-advance-work` | Simulation job status reads use only the Simulation Entrypoint's database: they never inspect, spawn or advance a Modal call; results are collected by Cloud Tasks and the Cloud Scheduler sweep. | mechanical | enforced | yes | pytest `projects/policyengine-simulation-entry/tests/test_simulation_job_routes.py`, `projects/policyengine-simulation-entry/tests/test_simulation_jobs.py` | `pr.yml:test` | **unproven** |
 
 ```mermaid
 flowchart LR
+  r_sim_persistence_conditional_job_writes("sim.persistence.conditional-job-writes (enforced)")
+  e_pytest_libs_policyengine_simulation_persistence_tests_test_job_store_concurrency_py["pytest libs/policyengine-simulation-persistence/tests/test_job_store_concurrency.py"]
+  j_CI_pr_yml_simulation_database["CI pr.yml:simulation-database"]
   r_sim_persistence_executor_independent("sim.persistence.executor-independent (enforced)")
   e_pytest_libs_policyengine_simulation_persistence_tests_test_ownership_boundaries_py["pytest libs/policyengine-simulation-persistence/tests/test_ownership_boundaries.py"]
   j_CI_pr_yml_test["CI pr.yml:test"]
+  r_sim_persistence_internal_routes_google_oidc("sim.persistence.internal-routes-google-oidc (enforced)")
+  e_pytest_projects_policyengine_simulation_entry_tests_test_google_oidc_py__projects_policyengine_simulation_entry_tests_test_simulation_job_routes_py["pytest projects/policyengine-simulation-entry/tests/test_google_oidc.py, projects/policyengine-simulation-entry/tests/test_simulation_job_routes.py"]
   r_sim_persistence_migrate_before_entrypoint("sim.persistence.migrate-before-entrypoint (enforced)")
   e_pytest_projects_policyengine_simulation_entry_tests_test_simulation_database_deployment_py["pytest projects/policyengine-simulation-entry/tests/test_simulation_database_deployment.py"]
   r_sim_persistence_migration_cycle("sim.persistence.migration-cycle (enforced)")
   e_pytest_libs_policyengine_simulation_persistence_tests_test_migration_cycle_py["pytest libs/policyengine-simulation-persistence/tests/test_migration_cycle.py"]
-  j_CI_pr_yml_simulation_database["CI pr.yml:simulation-database"]
   r_sim_persistence_no_modal_database_credentials("sim.persistence.no-modal-database-credentials (enforced)")
   r_sim_persistence_one_migration_chain("sim.persistence.one-migration-chain (enforced)")
+  r_sim_persistence_status_reads_never_advance_work("sim.persistence.status-reads-never-advance-work (enforced)")
+  e_pytest_projects_policyengine_simulation_entry_tests_test_simulation_job_routes_py__projects_policyengine_simulation_entry_tests_test_simulation_jobs_py["pytest projects/policyengine-simulation-entry/tests/test_simulation_job_routes.py, projects/policyengine-simulation-entry/tests/test_simulation_jobs.py"]
+  r_sim_persistence_conditional_job_writes --> e_pytest_libs_policyengine_simulation_persistence_tests_test_job_store_concurrency_py
+  e_pytest_libs_policyengine_simulation_persistence_tests_test_job_store_concurrency_py --> j_CI_pr_yml_simulation_database
   r_sim_persistence_executor_independent --> e_pytest_libs_policyengine_simulation_persistence_tests_test_ownership_boundaries_py
   e_pytest_libs_policyengine_simulation_persistence_tests_test_ownership_boundaries_py --> j_CI_pr_yml_test
+  r_sim_persistence_internal_routes_google_oidc --> e_pytest_projects_policyengine_simulation_entry_tests_test_google_oidc_py__projects_policyengine_simulation_entry_tests_test_simulation_job_routes_py
+  e_pytest_projects_policyengine_simulation_entry_tests_test_google_oidc_py__projects_policyengine_simulation_entry_tests_test_simulation_job_routes_py --> j_CI_pr_yml_test
   r_sim_persistence_migrate_before_entrypoint --> e_pytest_projects_policyengine_simulation_entry_tests_test_simulation_database_deployment_py
   e_pytest_projects_policyengine_simulation_entry_tests_test_simulation_database_deployment_py --> j_CI_pr_yml_test
   r_sim_persistence_migration_cycle --> e_pytest_libs_policyengine_simulation_persistence_tests_test_migration_cycle_py
   e_pytest_libs_policyengine_simulation_persistence_tests_test_migration_cycle_py --> j_CI_pr_yml_simulation_database
   r_sim_persistence_no_modal_database_credentials --> e_pytest_libs_policyengine_simulation_persistence_tests_test_ownership_boundaries_py
   r_sim_persistence_one_migration_chain --> e_pytest_libs_policyengine_simulation_persistence_tests_test_ownership_boundaries_py
+  r_sim_persistence_status_reads_never_advance_work --> e_pytest_projects_policyengine_simulation_entry_tests_test_simulation_job_routes_py__projects_policyengine_simulation_entry_tests_test_simulation_jobs_py
+  e_pytest_projects_policyengine_simulation_entry_tests_test_simulation_job_routes_py__projects_policyengine_simulation_entry_tests_test_simulation_jobs_py --> j_CI_pr_yml_test
 ```
 
 ## Deployment pipeline

libs/policyengine-simulation-persistence/README.md

@@ -30,6 +30,38 @@ The temporary Stage 12 comparison tables are not part of this schema: they are
 owned and migrated by the Public API, and `libs/policyengine-stage12-persistence`
 writes them with DML only.
 
+## The job store
+
+`simulation.jobs` has one row per calculation the Simulation Entrypoint sends
+to Modal: the submitter, a SHA-256 of the job kind, policyengine.py bundle
+version and request (`input_sha256`), the Modal application, function and
+call ID, the status, timings, the number of collection checks and any error.
+`simulation.job_results` holds the validated result of each succeeded job that
+is not a report.
+
+A job moves `pending` -> `dispatching` -> `running` -> `succeeded` or
+`failed`. Every transition in `jobs.py` is one conditional
+`UPDATE ... WHERE status = <expected>` (`sim.persistence.conditional-job-writes`):
+
+- **Submission** inserts with `ON CONFLICT DO NOTHING` on a unique index over
+  `input_sha256` that excludes failed jobs, so identical requests share one
+  job and a failed job never blocks a new attempt.
+- **Dispatch** happens only after the conditional `pending` -> `dispatching`
+  update succeeds, so one Modal call starts per job.
+- **Collection** writes a result (in the same transaction as the
+  `running` -> `succeeded` update) or a failure only while the job is
+  `running`. When two collectors race, PostgreSQL makes the second wait for the
+  first's row lock, re-check the condition and change nothing. No lease is
+  taken. `tests/test_job_store_concurrency.py` forces that interleaving and
+  races many threads.
+
+The Simulation Entrypoint collects results in the background: Cloud Tasks
+calls its collection route, which checks the Modal call without waiting and
+either stores the outcome or schedules the next check with backoff; a Cloud
+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`).
+
 ## Database roles
 
 | Role | Used by | Privileges |
@@ -58,11 +90,14 @@ reads them.
 
 ## Adding a revision
 
-1. Change the tables in `src/policyengine_simulation_persistence/`.
+1. Change the tables in `src/policyengine_simulation_persistence/`. A new
+   module that defines tables must also be imported by `migrations/env.py`,
+   because the schema comparison sees only tables whose module was imported.
 2. Add a revision under `migrations/versions/` whose `down_revision` is the
    current head, and set `HEAD_REVISION` in `schema.py` to its revision ID.
 3. Run the tests against a disposable PostgreSQL server (below). They run the
-   chain's upgrade, `alembic check`, downgrade and upgrade cycle.
+   chain's upgrade, `alembic check`, downgrade and upgrade cycle, and run
+   `alembic check` again in a separate process, as the command line does.
 
 Only one migration PR in this chain may be open at a time; a second one rebases
 onto the new head after the first merges.