Isolate staging Cloud Run service and add backend response header

Migration PR 4 stage 1. The staging Cloud Run track previously promoted
staging-configured revisions to 100% of the production policyengine-api
service URL on every push; once a load balancer fronts that service this
becomes a public incident per deploy. Staging now deploys and promotes
its own policyengine-api-staging service, with the image name decoupled
from the service name because production reuses the staging-built image.

Every API response now carries X-PolicyEngine-Backend (app_engine or
cloud_run) so traffic-weight changes and rollbacks during the host
cutover can be verified in-band with sampled requests.

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

Author: anth-volk

.github/scripts/cloud_run_env.sh

@@ -5,6 +5,9 @@ cloud_run_set_defaults() {
   CLOUD_RUN_REGION="${CLOUD_RUN_REGION:-us-central1}"
   CLOUD_RUN_SERVICE="${CLOUD_RUN_SERVICE:-policyengine-api}"
   CLOUD_RUN_ARTIFACT_REPOSITORY="${CLOUD_RUN_ARTIFACT_REPOSITORY:-policyengine-api}"
+  # Image name stays fixed across services: the production deploy reuses the
+  # image built by the staging track, so it must not embed the service name.
+  CLOUD_RUN_IMAGE_NAME="${CLOUD_RUN_IMAGE_NAME:-policyengine-api}"
   CLOUD_RUN_RUNTIME_SERVICE_ACCOUNT="${CLOUD_RUN_RUNTIME_SERVICE_ACCOUNT:-policyengine-api-cr-runtime@policyengine-api.iam.gserviceaccount.com}"
   CLOUD_RUN_CLOUD_SQL_INSTANCE="${CLOUD_RUN_CLOUD_SQL_INSTANCE:-policyengine-api:us-central1:policyengine-api-data}"
   CLOUD_RUN_CPU="${CLOUD_RUN_CPU:-4}"
@@ -22,7 +25,7 @@ cloud_run_set_defaults() {
   local sha
   sha="${GITHUB_SHA:-local}"
   CLOUD_RUN_IMAGE_TAG="${CLOUD_RUN_IMAGE_TAG:-${sha}}"
-  CLOUD_RUN_IMAGE_URI="${CLOUD_RUN_IMAGE_URI:-${CLOUD_RUN_REGION}-docker.pkg.dev/${CLOUD_RUN_PROJECT}/${CLOUD_RUN_ARTIFACT_REPOSITORY}/${CLOUD_RUN_SERVICE}:${CLOUD_RUN_IMAGE_TAG}}"
+  CLOUD_RUN_IMAGE_URI="${CLOUD_RUN_IMAGE_URI:-${CLOUD_RUN_REGION}-docker.pkg.dev/${CLOUD_RUN_PROJECT}/${CLOUD_RUN_ARTIFACT_REPOSITORY}/${CLOUD_RUN_IMAGE_NAME}:${CLOUD_RUN_IMAGE_TAG}}"
 
   local short_sha
   short_sha="${sha:0:7}"
@@ -32,6 +35,7 @@ cloud_run_set_defaults() {
   export CLOUD_RUN_REGION
   export CLOUD_RUN_SERVICE
   export CLOUD_RUN_ARTIFACT_REPOSITORY
+  export CLOUD_RUN_IMAGE_NAME
   export CLOUD_RUN_RUNTIME_SERVICE_ACCOUNT
   export CLOUD_RUN_CLOUD_SQL_INSTANCE
   export CLOUD_RUN_CPU

.github/workflows/push.yml

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

changelog.d/migration-pr4-stage1-staging-isolation.changed.md

@@ -0,0 +1 @@
+Isolated staging Cloud Run deploys onto a dedicated policyengine-api-staging service and tagged every API response with an X-PolicyEngine-Backend header.

policyengine_api/asgi_factory.py

@@ -9,6 +9,10 @@
 from a2wsgi import WSGIMiddleware
 from fastapi import FastAPI, HTTPException
 from policyengine_api.constants import VERSION
+from policyengine_api.migration_flags import (
+    BACKEND_RESPONSE_HEADER,
+    get_api_host_backend,
+)
 from policyengine_api.migration_logging import log_migration_request
 from pydantic import BaseModel
 from starlette.middleware.gzip import GZipMiddleware
@@ -56,6 +60,8 @@ async def add_cors_for_native_routes(request, call_next):
         started_at = time.time()
         request_id = request.headers.get("X-Request-ID") or uuid.uuid4().hex
         response = await call_next(request)
+        if BACKEND_RESPONSE_HEADER not in response.headers:
+            response.headers[BACKEND_RESPONSE_HEADER] = get_api_host_backend()
         origin = request.headers.get("origin")
         if origin and "access-control-allow-origin" not in response.headers:
             response.headers["Access-Control-Allow-Origin"] = origin

policyengine_api/migration_flags.py

@@ -30,6 +30,8 @@
 DEFAULT_SIM_FRONT_DOOR = "old_gateway_direct"
 DEFAULT_SIM_COMPUTE_BACKEND = "old_gateway"
 
+BACKEND_RESPONSE_HEADER = "X-PolicyEngine-Backend"
+
 
 @dataclass(frozen=True)
 class MigrationContext:
@@ -112,6 +114,18 @@ def get_db_read(entity: str) -> str:
     return _read_choice(env_name, DEFAULT_DB_SOURCE, DB_READ_SOURCES)
 
 
+def get_api_host_backend() -> str:
+    """Backend value for response tagging; must never break a response."""
+    try:
+        return _read_choice(
+            "API_HOST_BACKEND",
+            DEFAULT_API_HOST_BACKEND,
+            API_HOST_BACKENDS,
+        )
+    except ValueError:
+        return DEFAULT_API_HOST_BACKEND
+
+
 def get_sim_compute(flow: str) -> str:
     env_name = f"SIM_COMPUTE_{flow.upper()}"
     return _read_choice(

policyengine_api/migration_logging.py

@@ -9,6 +9,8 @@
 
 from policyengine_api.gcp_logging import logger
 from policyengine_api.migration_flags import (
+    BACKEND_RESPONSE_HEADER,
+    get_api_host_backend,
     get_migration_log_context,
     infer_route_group,
 )
@@ -26,6 +28,10 @@ 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
         try:
             log_migration_request(
                 request_id=getattr(flask.g, "request_id", None),

tests/unit/routes/test_migration_context_logging.py

@@ -4,6 +4,7 @@
 from flask import Flask, Response
 
 from policyengine_api.asgi_factory import create_asgi_app
+from policyengine_api.migration_flags import BACKEND_RESPONSE_HEADER
 from policyengine_api.migration_logging import register_migration_request_logging
 
 
@@ -108,3 +109,49 @@ def test_asgi_shell_does_not_log_unregistered_flask_fallback_routes():
     assert response.status_code == 200
     assert response.content == b"fallback"
     mock_logger.log_struct.assert_not_called()
+
+
+def test_flask_responses_include_backend_header_default():
+    with patch("policyengine_api.migration_logging.logger"):
+        response = _app().test_client().get("/readiness-check")
+
+    assert response.status_code == 200
+    assert response.headers[BACKEND_RESPONSE_HEADER] == "app_engine"
+
+
+def test_flask_responses_include_backend_header_for_cloud_run(monkeypatch):
+    monkeypatch.setenv("API_HOST_BACKEND", "cloud_run")
+
+    with patch("policyengine_api.migration_logging.logger"):
+        response = _app().test_client().get("/readiness-check")
+
+    assert response.headers[BACKEND_RESPONSE_HEADER] == "cloud_run"
+
+
+def test_backend_header_falls_back_to_default_on_invalid_flag(monkeypatch):
+    monkeypatch.setenv("API_HOST_BACKEND", "not-a-backend")
+
+    with patch("policyengine_api.migration_logging.logger"):
+        response = _app().test_client().get("/readiness-check")
+
+    assert response.status_code == 200
+    assert response.headers[BACKEND_RESPONSE_HEADER] == "app_engine"
+
+
+def test_fastapi_native_routes_include_backend_header(monkeypatch):
+    monkeypatch.setenv("API_HOST_BACKEND", "cloud_run")
+
+    with patch("policyengine_api.migration_logging.logger"):
+        response = TestClient(create_asgi_app(_app())).get("/health")
+
+    assert response.status_code == 200
+    assert response.headers[BACKEND_RESPONSE_HEADER] == "cloud_run"
+
+
+def test_asgi_shell_adds_backend_header_when_flask_hook_is_absent():
+    response = TestClient(create_asgi_app(_app_without_migration_logging())).get(
+        "/fallback"
+    )
+
+    assert response.status_code == 200
+    assert response.headers[BACKEND_RESPONSE_HEADER] == "app_engine"

tests/unit/test_cloud_run_deploy_scripts.py

@@ -9,6 +9,7 @@
 
 REPO = Path(__file__).resolve().parents[2]
 PRODUCTION_CLOUD_SQL_INSTANCE = "policyengine-api:us-central1:policyengine-api-data"
+STAGING_CLOUD_RUN_SERVICE = "policyengine-api-staging"
 DEDICATED_CLOUD_RUN_RUNTIME_SERVICE_ACCOUNT = (
     "policyengine-api-cr-runtime@policyengine-api.iam.gserviceaccount.com"
 )
@@ -384,6 +385,67 @@ def test_promote_cloud_run_tag_dry_run_shifts_service_traffic_to_tag():
     assert "--to-latest" not in result.stdout
 
 
+def test_push_workflow_isolates_staging_cloud_run_service():
+    workflow = _push_workflow()
+    staging_deploy = _workflow_job_block(workflow, "deploy-cloud-run-staging")
+    staging_promote = _workflow_job_block(workflow, "promote-cloud-run-staging")
+    production_deploy = _workflow_job_block(workflow, "deploy-cloud-run-candidate")
+
+    staging_service_env = f"CLOUD_RUN_SERVICE: {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
+
+
+def test_build_cloud_run_image_uri_is_independent_of_service_override():
+    result = _run_script(
+        ".github/scripts/build_cloud_run_image.sh",
+        _script_env(
+            GITHUB_SHA="1234567890abcdef",
+            GITHUB_RUN_NUMBER="42",
+            CLOUD_RUN_SERVICE=STAGING_CLOUD_RUN_SERVICE,
+        ),
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert (
+        "us-central1-docker.pkg.dev/policyengine-api/policyengine-api/"
+        "policyengine-api:1234567890abcdef"
+    ) in result.stdout
+    assert f"{STAGING_CLOUD_RUN_SERVICE}:" not in result.stdout
+
+
+def test_deploy_cloud_run_candidate_dry_run_targets_service_override():
+    result = _run_script(
+        ".github/scripts/deploy_cloud_run_candidate.sh",
+        _script_env(
+            **_required_runtime_env(),
+            CLOUD_RUN_IMAGE_URI="us-central1-docker.pkg.dev/project/repo/api:sha",
+            CLOUD_RUN_TAG="stage3-test",
+            CLOUD_RUN_SERVICE=STAGING_CLOUD_RUN_SERVICE,
+        ),
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert f"gcloud run deploy {STAGING_CLOUD_RUN_SERVICE}" in result.stdout
+
+
+def test_promote_cloud_run_tag_dry_run_targets_service_override():
+    result = _run_script(
+        ".github/scripts/promote_cloud_run_tag.sh",
+        _script_env(
+            CLOUD_RUN_TAG="stage3-test",
+            CLOUD_RUN_SERVICE=STAGING_CLOUD_RUN_SERVICE,
+        ),
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert (
+        f"gcloud run services update-traffic {STAGING_CLOUD_RUN_SERVICE}"
+        in result.stdout
+    )
+
+
 def test_push_workflow_tests_app_engine_and_cloud_run_staging_tracks():
     workflow = _push_workflow()
     app_engine_tests = _workflow_job_block(workflow, "integration-tests-staging")