Address Stage 1 review findings

Pin CLOUD_RUN_SERVICE explicitly on the production Cloud Run job and add
workflow invariants: every service-targeting Cloud Run job must pin a
service, and only the production job may promote the production service.
The cloud_run_env.sh default silently targeting production was the
mechanism behind the original staging-promotes-to-prod bug.

Drop the dead try/except around the backend header assignment;
get_api_host_backend() is fail-safe by design, so the except could only
mask real header-assignment bugs, and the ASGI path never had one.

Document the one-time staging service bootstrap (gcloud rejects
--no-traffic when creating a new service, and the deploy service account
cannot set the public invoker binding) in a Stage 1 runbook.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: anth-volk

.github/workflows/push.yml

@@ -475,6 +475,8 @@ jobs:
       (github.repository == 'PolicyEngine/policyengine-api')
       && (github.event.head_commit.message == 'Update PolicyEngine API')
     environment: production
+    env:
+      CLOUD_RUN_SERVICE: policyengine-api
     permissions:
       contents: read
       id-token: write

docs/migration-pr4-staging-service-runbook.md

@@ -0,0 +1,79 @@
+# PR 4 Stage 1 Staging Cloud Run Service Runbook
+
+Stage 1 of the public host cutover splits staging Cloud Run deploys onto a dedicated
+`policyengine-api-staging` service. Previously, both the staging and production CI tracks
+targeted the single `policyengine-api` service, and the staging promote step put a
+staging-configured revision on 100% of the production service URL during every push. That
+is harmless while nothing public routes to the service URL, and unacceptable once a load
+balancer fronts it.
+
+## One-Time Bootstrap (completed 2026-07-02)
+
+The CI pipeline cannot create this service itself, for two reasons:
+
+1. `deploy_cloud_run_candidate.sh` always deploys with `--no-traffic` (the PR 3
+   candidate-then-promote pattern). gcloud rejects `--no-traffic` when creating a new
+   service, because a new service must route 100% of traffic to its first revision.
+2. The GitHub deploy service account holds `roles/run.developer`, which can create and
+   deploy services but cannot set IAM policy. The `allUsers` -> `roles/run.invoker`
+   binding that `--allow-unauthenticated` needs on a new service must be created by a
+   project owner or `roles/run.admin` principal. (The production service's binding was
+   created manually during PR 3.)
+
+Bootstrap command, run once by a project owner before merging the Stage 1 PR:
+
+```bash
+gcloud run deploy policyengine-api-staging \
+  --project policyengine-api \
+  --region us-central1 \
+  --image us-docker.pkg.dev/cloudrun/container/hello \
+  --allow-unauthenticated \
+  --service-account policyengine-api-cr-runtime@policyengine-api.iam.gserviceaccount.com
+```
+
+The placeholder image is intentional: the first CI staging deploy replaces the revision
+entirely, since `deploy_cloud_run_candidate.sh` sets every flag (image, env vars, secrets,
+resources, service account, Cloud SQL attachment) explicitly on each deploy.
+
+## Verification
+
+```bash
+gcloud run services describe policyengine-api-staging \
+  --project policyengine-api --region us-central1 \
+  --format "value(status.url, spec.template.spec.serviceAccountName)"
+gcloud run services get-iam-policy policyengine-api-staging \
+  --project policyengine-api --region us-central1
+curl -s -o /dev/null -w '%{http_code}\n' <service URL>
+```
+
+Expected: service URL resolves, runtime service account is
+`policyengine-api-cr-runtime@policyengine-api.iam.gserviceaccount.com`, IAM policy contains
+`allUsers` with `roles/run.invoker`, and the curl returns 200.
+
+## Steady State After Stage 1
+
+- Staging jobs (`deploy-cloud-run-staging`, `promote-cloud-run-staging`) pin
+  `CLOUD_RUN_SERVICE: policyengine-api-staging`; the production job
+  (`deploy-cloud-run-candidate`) pins `CLOUD_RUN_SERVICE: policyengine-api`. Unit tests
+  enforce that every service-targeting Cloud Run job carries an explicit pin.
+- The Docker image name is decoupled from the service name (`CLOUD_RUN_IMAGE_NAME`):
+  production reuses the image built by the staging track, so both tracks push and pull
+  `policyengine-api:<sha>` regardless of service.
+- Every API response carries `X-PolicyEngine-Backend` (`app_engine` or `cloud_run`) for
+  in-band backend verification during the host cutover traffic ramps.
+- Known follow-up (out of Stage 1 scope): the staging service still uses the prod-named
+  Secret Manager secrets and the production Cloud SQL instance with staging DB user/name
+  values, as the shared-service track always did. The service split makes a per-service
+  secret set possible; migrate in a later stage.
+
+## Exit Gates
+
+Evaluated on the first post-merge push cycle (see the Stage 1 section of
+`api-v1-pr4-cloud-run-host-cutover-execution-plan.md` in the planning folder):
+
+- `gcloud run services describe policyengine-api --region us-central1` shows only
+  `stage3-prod-*` tags at 100% at all times, including after the staging promote step.
+- `policyengine-api-staging` is healthy and passes the staging integration suite via its
+  own URL.
+- `X-PolicyEngine-Backend` is present and correct on the App Engine production URL, the
+  Cloud Run staging service URL, and the Cloud Run production service URL.

policyengine_api/migration_logging.py

@@ -28,10 +28,7 @@ def set_request_migration_context():
 
     @app.after_request
     def log_request_migration_context(response):
-        try:
-            response.headers[BACKEND_RESPONSE_HEADER] = get_api_host_backend()
-        except Exception:
-            pass
+        response.headers[BACKEND_RESPONSE_HEADER] = get_api_host_backend()
         try:
             log_migration_request(
                 request_id=getattr(flask.g, "request_id", None),

tests/unit/test_cloud_run_deploy_scripts.py

@@ -9,7 +9,14 @@
 
 REPO = Path(__file__).resolve().parents[2]
 PRODUCTION_CLOUD_SQL_INSTANCE = "policyengine-api:us-central1:policyengine-api-data"
+PRODUCTION_CLOUD_RUN_SERVICE = "policyengine-api"
 STAGING_CLOUD_RUN_SERVICE = "policyengine-api-staging"
+CLOUD_RUN_SERVICE_SCRIPTS = (
+    "scripts/deploy_cloud_run_candidate.sh",
+    "scripts/promote_cloud_run_tag.sh",
+    "scripts/get_cloud_run_tag_url.sh",
+    "scripts/get_cloud_run_service_url.sh",
+)
 DEDICATED_CLOUD_RUN_RUNTIME_SERVICE_ACCOUNT = (
     "policyengine-api-cr-runtime@policyengine-api.iam.gserviceaccount.com"
 )
@@ -385,6 +392,10 @@ def test_promote_cloud_run_tag_dry_run_shifts_service_traffic_to_tag():
     assert "--to-latest" not in result.stdout
 
 
+def _workflow_job_names(workflow: str) -> list[str]:
+    return re.findall(r"^  ([a-zA-Z0-9_-]+):$", workflow, flags=re.MULTILINE)
+
+
 def test_push_workflow_isolates_staging_cloud_run_service():
     workflow = _push_workflow()
     staging_deploy = _workflow_job_block(workflow, "deploy-cloud-run-staging")
@@ -395,6 +406,38 @@ def test_push_workflow_isolates_staging_cloud_run_service():
     assert staging_service_env in staging_deploy
     assert staging_service_env in staging_promote
     assert STAGING_CLOUD_RUN_SERVICE not in production_deploy
+    assert f"CLOUD_RUN_SERVICE: {PRODUCTION_CLOUD_RUN_SERVICE}\n" in production_deploy
+
+
+def test_every_cloud_run_job_pins_a_service():
+    workflow = _push_workflow()
+
+    for job_name in _workflow_job_names(workflow):
+        block = _workflow_job_block(workflow, job_name)
+        if not any(script in block for script in CLOUD_RUN_SERVICE_SCRIPTS):
+            continue
+        assert re.search(r"CLOUD_RUN_SERVICE: \S+", block), (
+            f"Job {job_name} uses a service-targeting Cloud Run script without "
+            "pinning CLOUD_RUN_SERVICE; the cloud_run_env.sh default silently "
+            "targets production"
+        )
+
+
+def test_only_production_job_promotes_the_production_cloud_run_service():
+    workflow = _push_workflow()
+
+    for job_name in _workflow_job_names(workflow):
+        block = _workflow_job_block(workflow, job_name)
+        if "scripts/promote_cloud_run_tag.sh" not in block:
+            continue
+        if job_name == "deploy-cloud-run-candidate":
+            expected = f"CLOUD_RUN_SERVICE: {PRODUCTION_CLOUD_RUN_SERVICE}\n"
+        else:
+            expected = f"CLOUD_RUN_SERVICE: {STAGING_CLOUD_RUN_SERVICE}"
+        assert expected in block, (
+            f"Job {job_name} promotes Cloud Run traffic without pinning the "
+            "expected service"
+        )
 
 
 def test_build_cloud_run_image_uri_is_independent_of_service_override():