Calculate and store households through the v2 household functions

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

@@ -25,6 +25,12 @@ previous_revision="${3:-}"
 #   and model.
 # This is the identity the gateway reports from the routing registry; the
 # numerical release sentinel checks what the worker computes.
+# REQUIRE_CURRENT_SIMULATION_DATABASE_REVISION=true (candidate checks only):
+# /ready must report the entrypoint's own database reachable and at the
+# migration revision the image expects, which shows that the runtime secret
+# reached the revision and that the migration job ran. Stable and rollback
+# checks leave it unset, because an older image may face a newer schema.
+require_database_revision="${REQUIRE_CURRENT_SIMULATION_DATABASE_REVISION:-}"
 expected_identity="${EXPECTED_IDENTITY:-none}"
 expected_policyengine_version="${EXPECTED_POLICYENGINE_VERSION:-}"
 expected_app_name="${EXPECTED_APP_NAME:-}"
@@ -39,6 +45,14 @@ case "${expected_identity}" in
     exit 2
     ;;
 esac
+case "${require_database_revision}" in
+  "" | true) ;;
+  *)
+    printf 'REQUIRE_CURRENT_SIMULATION_DATABASE_REVISION must be true or empty, not %s\n' \
+      "${require_database_revision}" >&2
+    exit 2
+    ;;
+esac
 case "${expected_latest}" in
   "" | true) ;;
   *)
@@ -63,6 +77,9 @@ if [ "${expected_identity}" = "required" ] ||
 fi
 routing_max_attempts="${SIMULATION_ENTRYPOINT_ROUTING_MAX_ATTEMPTS:-24}"
 routing_retry_delay_seconds="${SIMULATION_ENTRYPOINT_ROUTING_RETRY_DELAY_SECONDS:-5}"
+# A cold candidate may not reach its database within /ready's three-second
+# wait; an unreachable database is asked about again this many times.
+database_ready_max_attempts="${SIMULATION_DATABASE_READY_MAX_ATTEMPTS:-6}"
 
 if ! [[ "${routing_max_attempts}" =~ ^[1-9][0-9]*$ ]]; then
   printf 'SIMULATION_ENTRYPOINT_ROUTING_MAX_ATTEMPTS must be a positive integer\n' >&2
@@ -72,6 +89,10 @@ if ! [[ "${routing_retry_delay_seconds}" =~ ^[0-9]+$ ]]; then
   printf 'SIMULATION_ENTRYPOINT_ROUTING_RETRY_DELAY_SECONDS must be a non-negative integer\n' >&2
   exit 2
 fi
+if ! [[ "${database_ready_max_attempts}" =~ ^[1-9][0-9]*$ ]]; then
+  printf 'SIMULATION_DATABASE_READY_MAX_ATTEMPTS must be a positive integer\n' >&2
+  exit 2
+fi
 
 health_headers="$(mktemp)"
 health_body="$(mktemp)"
@@ -131,8 +152,29 @@ while true; do
   attempt=$((attempt + 1))
 done
 
-curl --fail --silent --show-error "${base_url}/ready" |
-  jq -e '.status == "ready"' >/dev/null
+ready_body="$(curl --fail --silent --show-error "${base_url}/ready")"
+jq -e '.status == "ready"' <<<"${ready_body}" >/dev/null
+if [ "${require_database_revision}" = "true" ]; then
+  database_attempt=1
+  while [ "${database_attempt}" -lt "${database_ready_max_attempts}" ] &&
+    jq -e '.simulation_database.reachable == false' <<<"${ready_body}" >/dev/null; do
+    printf '%s: the simulation database was not reached; asking again (%s/%s)\n' \
+      "${base_url}" "${database_attempt}" "${database_ready_max_attempts}" >&2
+    sleep "${routing_retry_delay_seconds}"
+    ready_body="$(curl --fail --silent --show-error "${base_url}/ready")"
+    database_attempt=$((database_attempt + 1))
+  done
+fi
+if [ "${require_database_revision}" = "true" ] &&
+  ! jq -e '
+    .simulation_database.reachable == true
+    and (.simulation_database.revision | type == "string")
+    and .simulation_database.revision == .simulation_database.expected_revision
+  ' <<<"${ready_body}" >/dev/null; then
+  printf '%s: the simulation database is not at the expected revision: %s\n' \
+    "${base_url}" "$(jq -c '.simulation_database' <<<"${ready_body}")" >&2
+  exit 1
+fi
 curl --fail --silent --show-error "${base_url}/versions" --output "${versions_body}"
 jq -e 'type == "object"' "${versions_body}" >/dev/null
 

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

@@ -58,6 +58,28 @@ resolve_secret_reference() {
   printf '%s:%s' "${reference}" "${highest_enabled_version}"
 }
 
+runtime_secrets=()
+
+# Resolve one secret per assignment: an array assignment takes its exit status
+# from its last command substitution only, so resolving several secrets in
+# one would let an earlier failure pass unnoticed.
+add_runtime_secret() {
+  local variable_name="$1"
+  local resolved_reference
+  resolved_reference="$(resolve_secret_reference "$2")"
+  runtime_secrets+=("${variable_name}=${resolved_reference}")
+}
+
+# The Stage 12 comparison path's own wrapper, removed with it, so the main
+# path's wrapper can change without touching the Stage 12 block. Both use the
+# secret lookups above, which predate the split.
+add_stage12_runtime_secret() {
+  local variable_name="$1"
+  local resolved_reference
+  resolved_reference="$(resolve_secret_reference "$2")"
+  runtime_secrets+=("${variable_name}=${resolved_reference}")
+}
+
 required_variables=(
   PROJECT_ID
   REGION
@@ -77,6 +99,10 @@ required_variables=(
   OLD_GATEWAY_AUTH_AUDIENCE_VALUE
   OLD_GATEWAY_AUTH_CLIENT_ID_VALUE
   OLD_GATEWAY_AUTH_CLIENT_SECRET_SECRET_NAME
+  SIMULATION_DATABASE_URL_SECRET_NAME
+  MODAL_TOKEN_ID_SECRET_NAME
+  MODAL_TOKEN_SECRET_SECRET_NAME
+  MODAL_WORKSPACE_ID_VALUE
   OBSERVABILITY_SERVICE_NAMESPACE
   OBSERVABILITY_TRACE_PROJECT_ID
   OTEL_EXPORTER_OTLP_ENDPOINT
@@ -103,6 +129,10 @@ esac
   || fail "TAG is invalid"
 [[ "${ENTRYPOINT_RUNTIME_SERVICE_ACCOUNT}" =~ ^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.gserviceaccount\.com$ ]] \
   || fail "ENTRYPOINT_RUNTIME_SERVICE_ACCOUNT is invalid"
+# The Modal workspace whose call-report function may report that calls
+# ended (the entrypoint also checks the environment, MODAL_ENVIRONMENT).
+[[ "${MODAL_WORKSPACE_ID_VALUE}" =~ ^ac-[A-Za-z0-9]+$ ]] \
+  || fail "MODAL_WORKSPACE_ID_VALUE must be a Modal workspace ID"
 [[ "${ENTRYPOINT_MIN_INSTANCES}" =~ ^[0-9]+$ ]] \
   || fail "ENTRYPOINT_MIN_INSTANCES must be a non-negative integer"
 [[ "${ENTRYPOINT_MAX_INSTANCES}" =~ ^[0-9]+$ ]] \
@@ -115,18 +145,21 @@ export SIMULATION_ENTRYPOINT_ENFORCE_PERMISSIONS_VALUE="${SIMULATION_ENTRYPOINT_
 [[ "${SIMULATION_ENTRYPOINT_ENFORCE_PERMISSIONS_VALUE}" == "0" || "${SIMULATION_ENTRYPOINT_ENFORCE_PERMISSIONS_VALUE}" == "1" ]] \
   || fail "SIMULATION_ENTRYPOINT_ENFORCE_PERMISSIONS_VALUE must be 0 or 1"
 
-runtime_secrets=(
-  "OLD_GATEWAY_AUTH_CLIENT_SECRET=$(
-    resolve_secret_reference "${OLD_GATEWAY_AUTH_CLIENT_SECRET_SECRET_NAME}"
-  )"
-)
+add_runtime_secret OLD_GATEWAY_AUTH_CLIENT_SECRET \
+  "${OLD_GATEWAY_AUTH_CLIENT_SECRET_SECRET_NAME}"
+# The runtime role's URL for the entrypoint's own simulation schema. The
+# migration job, which uses a separate migrator URL, runs before this deploy.
+add_runtime_secret SIMULATION_DATABASE_URL "${SIMULATION_DATABASE_URL_SECRET_NAME}"
+# sim's own Modal token, on every deploy: the entrypoint starts runs' Modal
+# calls and asks Modal about a call when Modal reports that it ended. Until
+# the Stage 12 comparison path is removed, its Modal calls use the same token.
+add_runtime_secret MODAL_TOKEN_ID "${MODAL_TOKEN_ID_SECRET_NAME}"
+add_runtime_secret MODAL_TOKEN_SECRET "${MODAL_TOKEN_SECRET_SECRET_NAME}"
 
 if [[ "${DEPLOY_STAGE12_V2}" == "true" ]]; then
   stage12_variables=(
     STAGE12_ENABLED_VALUE
     STAGE12_ARTIFACT_BUCKET_VALUE
-    MODAL_TOKEN_ID_SECRET_NAME
-    MODAL_TOKEN_SECRET_SECRET_NAME
     STAGE12_DATABASE_URL_SECRET_NAME
   )
   for variable_name in "${stage12_variables[@]}"; do
@@ -138,11 +171,7 @@ if [[ "${DEPLOY_STAGE12_V2}" == "true" ]]; then
     || fail "STAGE12_ARTIFACT_BUCKET_VALUE is invalid"
   validate_secret_reference "${STAGE12_DATABASE_URL_SECRET_NAME}"
 
-  runtime_secrets+=(
-    "MODAL_TOKEN_ID=$(resolve_secret_reference "${MODAL_TOKEN_ID_SECRET_NAME}")"
-    "MODAL_TOKEN_SECRET=$(resolve_secret_reference "${MODAL_TOKEN_SECRET_SECRET_NAME}")"
-    "STAGE12_DATABASE_URL=$(resolve_secret_reference "${STAGE12_DATABASE_URL_SECRET_NAME}")"
-  )
+  add_stage12_runtime_secret STAGE12_DATABASE_URL "${STAGE12_DATABASE_URL_SECRET_NAME}"
 fi
 
 runtime_environment_file="${RUNNER_TEMP}/simulation-entry-runtime-environment.yaml"
@@ -167,6 +196,8 @@ jq -n '
     OTEL_METRICS_EXPORTER: "otlp",
     OTEL_TRACES_SAMPLER_ARG: "1.0",
     POLICYENGINE_OTEL_GOOGLE_AUDIENCE: env.POLICYENGINE_OTEL_GOOGLE_AUDIENCE,
+    SIMULATION_CALL_REPORT_MODAL_WORKSPACE_ID: env.MODAL_WORKSPACE_ID_VALUE,
+    SIMULATION_CALL_REPORT_MODAL_ENVIRONMENT: env.MODAL_ENVIRONMENT,
     STAGE12_ENABLED: (
       if env.DEPLOY_STAGE12_V2 == "true" then env.STAGE12_ENABLED_VALUE else "0" end
     )

.github/scripts/deploy-cloud-run-simulation-safety-check.sh

@@ -0,0 +1,132 @@
+#!/usr/bin/env bash
+# Point the safety-check Cloud Run job at the entrypoint revision that serves
+# traffic, and keep its Cloud Scheduler entry.
+#
+# The job runs `python -m policyengine_simulation_entry.safety_check`: one
+# recording pass per execution. It is updated only from a revision that was
+# promoted (or restored by a rollback), so a rejected candidate never records
+# anything. It copies the revision's image digest, service account, Modal
+# token and database URL (at the secret versions the revision pins) and its
+# plain telemetry settings; nothing else. A revision deployed before the
+# safety check existed has no job code, so its schedule is paused instead.
+#
+# With the argument `pause`, only pause the schedule: a workflow does this
+# when pointing the job at a revision failed part way, so the job never keeps
+# running the image traffic just moved away from.
+#
+# Environment: PROJECT_ID, REGION, SERVICE, and unless pausing,
+# SAFETY_CHECK_REVISION, SAFETY_CHECK_SCHEDULER_SERVICE_ACCOUNT, RUNNER_TEMP.
+set -euo pipefail
+
+fail() {
+  printf '%s\n' "$*" >&2
+  exit 1
+}
+
+pause_schedule() {
+  if gcloud scheduler jobs describe "${job}" "${scheduler_location[@]}" >/dev/null 2>&1; then
+    gcloud scheduler jobs pause "${job}" "${scheduler_location[@]}" --quiet
+  fi
+}
+
+# Every five minutes: SAFETY_CHECK_INTERVAL in
+# policyengine_simulation_entry.run_recording.
+SCHEDULE="*/5 * * * *"
+SECRETS=(SIMULATION_DATABASE_URL MODAL_TOKEN_ID MODAL_TOKEN_SECRET)
+PLAIN_VARIABLES_FILTER='^(APP_ENVIRONMENT|CLOUD_RUN_REGION|OBSERVABILITY_[A-Z_]+|OTEL_[A-Z_]+|POLICYENGINE_OTEL_GOOGLE_AUDIENCE)$'
+
+for variable_name in PROJECT_ID REGION SERVICE; do
+  [[ -n "${!variable_name:-}" ]] || fail "${variable_name} is required"
+done
+[[ "${PROJECT_ID}" =~ ^[a-z][a-z0-9-]{4,28}[a-z0-9]$ ]] || fail "PROJECT_ID is invalid"
+[[ "${REGION}" =~ ^[a-z]+-[a-z]+[0-9]+$ ]] || fail "REGION is invalid"
+[[ "${SERVICE}" =~ ^[a-z]([a-z0-9-]{0,47}[a-z0-9])?$ ]] || fail "SERVICE is invalid"
+
+job="${SERVICE}-safety-check"
+location=(--project "${PROJECT_ID}" --region "${REGION}")
+scheduler_location=(--project "${PROJECT_ID}" --location "${REGION}")
+
+if [[ "${1:-}" == "pause" ]]; then
+  pause_schedule
+  printf 'The safety check schedule for %s is paused.\n' "${SERVICE}"
+  exit 0
+fi
+
+for variable_name in SAFETY_CHECK_REVISION SAFETY_CHECK_SCHEDULER_SERVICE_ACCOUNT RUNNER_TEMP; do
+  [[ -n "${!variable_name:-}" ]] || fail "${variable_name} is required"
+done
+[[ "${SAFETY_CHECK_REVISION}" =~ ^[a-z]([a-z0-9-]{0,61}[a-z0-9])?$ ]] \
+  || fail "SAFETY_CHECK_REVISION is invalid"
+[[ "${SAFETY_CHECK_SCHEDULER_SERVICE_ACCOUNT}" =~ ^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.gserviceaccount\.com$ ]] \
+  || fail "SAFETY_CHECK_SCHEDULER_SERVICE_ACCOUNT is invalid"
+
+revision_json="$(gcloud run revisions describe "${SAFETY_CHECK_REVISION}" "${location[@]}" --format=json)"
+revision_service="$(jq -r '.metadata.labels["serving.knative.dev/service"] // ""' <<<"${revision_json}")"
+[[ "${revision_service}" == "${SERVICE}" ]] \
+  || fail "${SAFETY_CHECK_REVISION} belongs to ${revision_service:-no service}, not ${SERVICE}"
+
+# Every revision deployed with the safety check has the simulation database
+# and the expected Modal caller (SIMULATION_CALL_REPORT_MODAL_WORKSPACE_ID,
+# which the deploy has required since then); an older one has no job code.
+if ! jq -e '
+  .spec.containers[0].env // []
+  | any(.name == "SIMULATION_DATABASE_URL")
+    and any(.name == "SIMULATION_CALL_REPORT_MODAL_WORKSPACE_ID")
+' >/dev/null <<<"${revision_json}"; then
+  pause_schedule
+  printf '%s predates the safety check, so its schedule is paused.\n' "${SAFETY_CHECK_REVISION}"
+  exit 0
+fi
+
+image="$(jq -er '.status.imageDigest // .spec.containers[0].image' <<<"${revision_json}")"
+service_account="$(jq -er '.spec.serviceAccountName' <<<"${revision_json}")"
+
+job_secrets=()
+for secret_variable in "${SECRETS[@]}"; do
+  reference="$(jq -er --arg name "${secret_variable}" '
+    [.spec.containers[0].env[] | select(.name == $name) | .valueFrom.secretKeyRef
+      | select(.name != null and .key != null) | "\(.name):\(.key)"]
+    | if length == 1 then .[0] else error("not one secret reference") end
+  ' <<<"${revision_json}")" || fail "${SAFETY_CHECK_REVISION} has no pinned secret for ${secret_variable}"
+  job_secrets+=("${secret_variable}=${reference}")
+done
+job_secrets_csv="$(IFS=,; printf '%s' "${job_secrets[*]}")"
+
+environment_file="${RUNNER_TEMP}/simulation-safety-check-environment.yaml"
+trap 'rm -f "${environment_file}"' EXIT
+jq --arg filter "${PLAIN_VARIABLES_FILTER}" '
+  [.spec.containers[0].env[] | select(.value != null and (.name | test($filter)))
+    | {(.name): .value}] | add // {}
+' <<<"${revision_json}" > "${environment_file}"
+
+gcloud run jobs deploy "${job}" "${location[@]}" \
+  --image "${image}" \
+  --service-account "${service_account}" \
+  --set-secrets "${job_secrets_csv}" \
+  --env-vars-file "${environment_file}" \
+  --command /opt/venv/bin/python \
+  --args=-m,policyengine_simulation_entry.safety_check \
+  --tasks 1 \
+  --max-retries 0 \
+  --task-timeout 600s \
+  --cpu 1 \
+  --memory 512Mi \
+  --labels "serving-revision=${SAFETY_CHECK_REVISION}" \
+  --quiet
+
+schedule_action=create
+if gcloud scheduler jobs describe "${job}" "${scheduler_location[@]}" >/dev/null 2>&1; then
+  schedule_action=update
+fi
+gcloud scheduler jobs "${schedule_action}" http "${job}" "${scheduler_location[@]}" \
+  --schedule "${SCHEDULE}" \
+  --time-zone Etc/UTC \
+  --uri "https://${REGION}-run.googleapis.com/apis/run.googleapis.com/v1/namespaces/${PROJECT_ID}/jobs/${job}:run" \
+  --http-method POST \
+  --oauth-service-account-email "${SAFETY_CHECK_SCHEDULER_SERVICE_ACCOUNT}" \
+  --attempt-deadline 60s \
+  --quiet
+if [[ "${schedule_action}" == "update" ]]; then
+  gcloud scheduler jobs resume "${job}" "${scheduler_location[@]}" --quiet
+fi
+printf 'The safety check runs %s from %s.\n' "${image}" "${SAFETY_CHECK_REVISION}"

.github/scripts/migrate-simulation-database.sh

@@ -0,0 +1,37 @@
+#!/usr/bin/env bash
+# Bring the simulation schema to the chain's head, or require it to be there.
+# Run from libs/policyengine-simulation-persistence, whose locked environment
+# (with the migrations extra) provides Alembic. migrations/env.py reads the
+# migrator URL from SIMULATION_MIGRATION_DATABASE_URL; this script never
+# prints it.
+#
+# SIMULATION_MIGRATION_MODE:
+#   upgrade  apply every pending revision, then require the database at the
+#            head (deployments from the default branch);
+#   verify   apply nothing and only require the database at this checkout's
+#            head, so a deployment dispatched from another branch can never
+#            apply that branch's revisions.
+
+set -euo pipefail
+set +x
+
+if [[ -z "${SIMULATION_MIGRATION_DATABASE_URL:-}" ]]; then
+  printf 'SIMULATION_MIGRATION_DATABASE_URL must name the simulation database\n' >&2
+  exit 1
+fi
+
+mode="${SIMULATION_MIGRATION_MODE:-}"
+case "${mode}" in
+  upgrade | verify) ;;
+  *)
+    printf 'SIMULATION_MIGRATION_MODE must be upgrade or verify, not %s\n' "${mode:-empty}" >&2
+    exit 1
+    ;;
+esac
+
+alembic_config="../../migrations/alembic.ini"
+
+if [[ "${mode}" == "upgrade" ]]; then
+  uv run --frozen --extra migrations alembic -c "${alembic_config}" upgrade head
+fi
+uv run --frozen --extra migrations alembic -c "${alembic_config}" current --check-heads

.github/scripts/modal-image-smoke.sh

@@ -3,8 +3,10 @@
 # Usage: ./modal-image-smoke.sh <modal-environment>
 #
 # Each smoke imports the true app entrypoints inside the real (or, for
-# the executor, prefix-identical) Modal image — catching in-image
-# dependency breakage (issue #602's class) before merge.
+# the executor and the v2 US worker, prefix-identical) Modal image —
+# catching in-image dependency breakage (issue #602's class) before merge.
+# The v2 image smoke builds every image of the v2 application: the report
+# coordinator, the country simulation workers and the household functions.
 
 set -euo pipefail
 
@@ -47,4 +49,10 @@ echo "=== Stage 12 static runtime file smoke (env: $MODAL_ENV) ==="
         src/modal/v2_static_runtime_files_smoke.py
 )
 
+echo "=== v2 image smoke: every v2 image and its entry points (env: $MODAL_ENV) ==="
+(
+    cd "$REPO_ROOT/projects/policyengine-simulation-executor"
+    uv run --frozen modal run --env="$MODAL_ENV" src/modal/v2_image_smoke.py
+)
+
 echo "=== Image smokes passed ==="