[anth-volk]

Fixes #845

What this changes

Step 1 of the API v2 migration, item 1A-4 (result collection option B: Cloud Tasks, no Postgres leases). Second of three stacked PRs: #847, then #848 (this one), then #849. It needs #847.

Tables (revision 0002). simulation.jobs has one row per calculation: the submitter, input_sha256 (a SHA-256 of the job kind, the policyengine.py bundle version and the request as canonical JSON), the bundle version, the Modal application, function and call ID, the status, created_at/updated_at/dispatched_at/last_checked_at/completed_at, the number of collection checks, the persisted observability_id and any error. simulation.job_results holds the validated result, its schema version and SHA-256 for each succeeded job that is not a report (#849 adds report_results).

Conditional writes, no leases (JobStore in the persistence library; rule sim.persistence.conditional-job-writes):

• Submission is INSERT ... ON CONFLICT DO NOTHING on a unique index over input_sha256 that excludes failed jobs. Identical requests share one job; a failed job never blocks a new attempt.
• Only the instance whose UPDATE ... SET status = 'dispatching' WHERE status = 'pending' returns a row spawns the Modal call.
• A result is written in the same transaction as UPDATE ... SET status = 'succeeded' WHERE status = 'running', and a failure with ... WHERE status = 'running'. When two collectors race, PostgreSQL makes the second wait for the first's row lock, re-check the condition after the first commits, and update nothing.

Tests for concurrent collectors (tests/test_job_store_concurrency.py, PostgreSQL). Five tests force the interleaving: writer A's transaction holds the row lock, and the test waits until pg_stat_activity shows writer B blocked on it before A commits or rolls back. They cover result vs result, result vs failure (both orders), rollback of the first writer, and dispatch vs dispatch. Four more race eight threads through the public methods per job: results, mixed results and failures, dispatch claims and identical submissions. Removing the status condition from the result write makes four of them fail.

Simulation Entrypoint (simulation_jobs.py, google_oidc.py):

• SimulationJobService.submit records the job, claims dispatch, spawns the Modal call with the captured observability context, records the call ID and schedules the first collection task. A refused spawn fails the job; a refused task is left to the sweep.
• POST /internal/simulation-jobs/{job_id}/collect, called by Cloud Tasks: checks the Modal call with FunctionCall.get(timeout=0). If it is running, it counts the check and enqueues the next task after 2, 4, 8, 16, 32 and then 60 seconds. A finished result is validated against the job kind's Pydantic model and written conditionally. A worker failure is stored with an error code and only the exception type. Transient Modal, Cloud Tasks or database failures return 503 so that Cloud Tasks retries.
• POST /internal/simulation-jobs/sweep, called by Cloud Scheduler: enqueues a task for running jobs nobody has checked for five minutes and fails jobs stuck before their Modal call was recorded for ten minutes (dispatch_abandoned). It returns 503 if any task was refused, so the failed run shows in Cloud Scheduler.
• Both internal routes accept only a Google-signed OIDC token whose audience is the stable service URL and whose email is the invoker service account (rule sim.persistence.internal-routes-google-oidc).
• GET /simulation-jobs/{job_reference} (permission simulation:read) reads a job by UUID or Modal fc- call ID from the database only, returning 202 with Retry-After while unfinished and the stored result once succeeded (rule sim.persistence.status-reads-never-advance-work).
• None of these routes is in the OpenAPI document; the route table stays frozen.

Not included. No job kind is registered (REGISTERED_JOB_KINDS is empty) and no route submits jobs; Step 2 adds both. Nothing in production uses the new tables or routes. The old-gateway pass-through routes and the Stage 12 comparison are unchanged.

Deploy. The deploy step now requires and passes RECONCILIATION_QUEUE and RECONCILIATION_INVOKER_SERVICE_ACCOUNT (GitHub environment variables), RECONCILIATION_TARGET_URL (the stable service URL, which the deploy job already reads from Cloud Run before deploying) and SIMULATION_MODAL_ENVIRONMENT.

Before merging: operator steps (1A-3) in both staging and production

None of these has been performed. The deploy fails without step 5, and collection cannot work without the others.

1. Create the Cloud Tasks queue in the Simulation Entrypoint project, with a retry policy for 503 responses.
2. Create the invoker service account. Grant the Simulation Entrypoint runtime service accounts (sim-entry-beta-runtime, sim-entry-prod-runtime) roles/cloudtasks.enqueuer on the queue and roles/iam.serviceAccountUser on the invoker (to attach its OIDC token to tasks). Grant the invoker roles/run.invoker on the service.
3. Create the Cloud Scheduler job that sends POST <stable service URL>/internal/simulation-jobs/sweep every few minutes with an OIDC token of the invoker service account and audience equal to the stable service URL (the value of steps.previous.outputs.stable_url in the deploy job).
4. Confirm the Cloud Run ingress setting accepts requests from Cloud Tasks and Cloud Scheduler on the stable URL (it is deployed with --allow-unauthenticated; the routes authenticate the token themselves).
5. Set the RECONCILIATION_QUEUE (projects/<project>/locations/<location>/queues/<queue>) and RECONCILIATION_INVOKER_SERVICE_ACCOUNT variables in the beta and prod GitHub environments.

After deployment, the plan's staging load check of Cloud Tasks and the conditional writes is still to be done before acceptance.

Environment variables

Name	Read by	Provider	How it reaches the reader	Status
RECONCILIATION_QUEUE	Simulation Entrypoint (config.py, CloudTasksCollectionQueue)	GitHub environment variable RECONCILIATION_QUEUE (beta, prod)	step env: RECONCILIATION_QUEUE_VALUE → validated by the deploy script → Cloud Run environment file	Wiring checked; the variable and queue do not exist yet: unverified
RECONCILIATION_INVOKER_SERVICE_ACCOUNT	Simulation Entrypoint (task OIDC tokens and GoogleOidcVerifier)	GitHub environment variable (beta, prod)	step env: RECONCILIATION_INVOKER_SERVICE_ACCOUNT_VALUE → deploy script → Cloud Run environment file	Wiring checked; the account and IAM do not exist yet: unverified
RECONCILIATION_TARGET_URL	Simulation Entrypoint (task URL and token audience)	Cloud Run: the service's status.url, read by cloud-run-simulation-entry-stable-state.sh	steps.previous.outputs.stable_url → step env: ENTRYPOINT_STABLE_URL → deploy script → Cloud Run environment file	Verified in the repository; no operator configuration
SIMULATION_MODAL_ENVIRONMENT	Simulation Entrypoint (ModalCallDispatcher)	Reusable workflow input modal_environment	step env: MODAL_ENVIRONMENT → deploy script → Cloud Run environment file	Verified in the repository
Google credentials for Cloud Tasks	GoogleAccessTokens (google.auth.default)	The Cloud Run runtime service account	Cloud Run metadata server	Enqueue permission is step 2: unverified
MODAL_TOKEN_ID, MODAL_TOKEN_SECRET (existing)	Modal client in the collection route	Secret Manager	Mounted by the deploy script only when DEPLOY_STAGE12_V2 is true, which both deploy workflows set	Existing wiring; unchanged

Checks run

• libs/policyengine-simulation-persistence: against a local PostgreSQL 17 with SIMULATION_REQUIRE_DATABASE_TESTS=1, 46 passed.
• projects/policyengine-simulation-entry: uv run pytest tests/, 654 passed.
• projects/policyengine-simulation-executor: uv run pytest tests/, 1292 passed and 22 skipped.
• ruff check ., ruff format --check, the rules guard, the quality guards (with shellcheck).
• Ran the built entrypoint image with the job store configured against a local database migrated to 0002: a job written as simulation_runtime read back by UUID and by fc- call ID, the internal routes refused requests without a valid Google token, and /ready reported 0002.
• Not run: anything that needs Modal, Cloud Tasks, Cloud Scheduler or Google's token-signing keys; those adapters are tested with fakes (httpx.MockTransport, patched Modal handles, locally signed RS256 tokens).

Deviations from the plan

• RECONCILIATION_TARGET_URL and SIMULATION_MODAL_ENVIRONMENT are new runtime variables not listed in the plan's configuration table. Both come from values the deploy job already has, so they add no operator configuration.
• One invoker service account serves both Cloud Tasks and the Cloud Scheduler job.
• There is no submission route yet, because no job kind exists until Step 2; submission is implemented and tested in SimulationJobService.

Notes for review

• Changes under rules/ need approval from a core developer other than the author.
• Likely conflicts: Register weighted-microdata rules, weight the builder tests, and check change weights and household counts #807, Enable ruff argument, idiom and pytest-style rules (ARG, B, C4, FBT, PERF, PLW, PT, PTH, RUF, SIM and others) #802 and Refactor within ruff's complexity limits, then select every ruff rule (ALL) #804 (see Add the simulation schema's persistence library, migration chain and pre-deploy migration job #847).

🤖 Generated with Claude Code