diff --git a/Dockerfile b/Dockerfile
index bd4b8ada..48485843 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -12,45 +12,50 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# --- Build stage -------------------------------------------------------------
-# The full python:3.13 image carries the compilers/headers that a dependency
-# without a prebuilt cp313 wheel would need. Install everything into a
-# relocatable prefix (/install) that the slim runtime can drop in as-is, so the
-# build can never fail for lack of a compiler on the slim base.
-FROM python:3.13@sha256:e72bfff2ccf413e3c329074d643fac616d7e1dfe85ac57e527f1d13cd8e0ee6c AS builder
+# ==============================================================================
+# Stage 1: OS Runtime + FFmpeg (Runs concurrently with Stage 2 under BuildKit!)
+# ==============================================================================
+FROM python:3.13-slim@sha256:c33f0bc4364a6881bed1ec0cc2665e6c53c87a43e774aaeab88e6f17af105e4f AS runtime-base
 
 ENV PYTHONUNBUFFERED=1
 
-COPY requirements.txt .
-RUN pip install --no-cache-dir --require-hashes --prefix=/install -r requirements.txt
-
-# --- Runtime stage -----------------------------------------------------------
-# python:3.13-slim is ~850 MB smaller than the full image: faster to push to the
-# registry and faster to cold-start. It carries only ffmpeg, the dependencies
-# built above, and the app — no compilers or build cruft.
-FROM python:3.13-slim@sha256:c33f0bc4364a6881bed1ec0cc2665e6c53c87a43e774aaeab88e6f17af105e4f
-
-ENV PYTHONUNBUFFERED=1
-
-# ffmpeg is required by the worker's video actions (combine/convert). One layer,
-# no recommended extras, apt lists dropped to keep the image small.
+# Exclude Debian recommended GUI/X11/Mesa packages.
 RUN apt-get update \
   && apt-get install -y --no-install-recommends ffmpeg \
-  && rm -rf /var/lib/apt/lists/*
-
-# Run as a non-root user with a real home, and give it a writable app dir it
-# owns. The worker's video actions write temp files using bare relative names
-# into the process CWD (== WORKDIR), so WORKDIR must be owned by this user.
-RUN useradd --create-home --uid 10001 --shell /usr/sbin/nologin appuser \
+  && rm -rf /var/lib/apt/lists/* \
+  && useradd --create-home --uid 10001 --shell /usr/sbin/nologin appuser \
   && mkdir -p /app \
   && chown appuser:appuser /app
 
+# ==============================================================================
+# Stage 2: Python Dependency Builder via official Astral uv (digest-pinned)
+#          (Executes in ~3-6s *while* Stage 1 is still running apt-get!)
+# ==============================================================================
+FROM python:3.13-slim@sha256:c33f0bc4364a6881bed1ec0cc2665e6c53c87a43e774aaeab88e6f17af105e4f AS venv-builder
+
+COPY --from=ghcr.io/astral-sh/uv:0.12.18@sha256:3adc3706091ce7c2fe595e669628caedd6d951551b92b258b7e7dbe06d9440bc /uv /bin/uv
+ENV UV_COMPILE_BYTECODE=1 \
+    UV_LINK_MODE=copy \
+    VIRTUAL_ENV=/opt/venv \
+    PATH="/opt/venv/bin:$PATH"
+
+WORKDIR /app
+COPY requirements.txt .
+RUN uv venv /opt/venv \
+  && uv pip install --no-cache --only-binary :all: --require-hashes -r requirements.txt
+
+# ==============================================================================
+# Stage 3: Final Image Assembly (< 1 second merge)
+# ==============================================================================
+FROM runtime-base AS final
+
+ENV VIRTUAL_ENV=/opt/venv \
+    PATH="/opt/venv/bin:$PATH" \
+    PYTHONUNBUFFERED=1
+
 WORKDIR /app
+COPY --from=venv-builder /opt/venv /opt/venv
 
-# Drop in the dependencies built in the full image (same python 3.13, so the
-# installed packages and gunicorn entry point land on /usr/local and PATH).
-# Left root-owned and world-readable — import/exec only need read access.
-COPY --from=builder /install /usr/local
 
 # Runtime files only (not the whole repo): explicit copies keep docs, examples,
 # tests, deploy scripts, and .git out of the image. Root-owned but world-readable
diff --git a/cloudbuild.yaml b/cloudbuild.yaml
index b4d8b725..50c61aaf 100644
--- a/cloudbuild.yaml
+++ b/cloudbuild.yaml
@@ -9,7 +9,7 @@
 # Layer caching (faster repeat deploys): by default the build pulls the
 # previously pushed :latest image and passes it to `docker build --cache-from`,
 # so unchanged layers are reused instead of rebuilt. The slow builder-stage
-# `pip install` only re-runs when requirements.txt changes. Two details make
+# `uv pip install` only re-runs when requirements.txt changes. Two details make
 # this actually work:
 #   * the pull is best-effort (`|| true`): the first build on a fresh project has
 #     nothing to pull and must not fail;
@@ -18,7 +18,8 @@
 #     build can read these layers back. Without the inline-cache arg,
 #     --cache-from finds the image but gets zero cache hits.
 # Set _USE_CACHE=0 (deploy.sh --no-build-cache) to force a clean cold rebuild,
-# e.g. a release or a dependency/CVE refresh.
+# e.g. a dependency/CVE refresh. Keep inline-cache metadata on that fresh image
+# so the next ordinary redeploy can reuse its unchanged layers.
 steps:
   - name: 'gcr.io/cloud-builders/docker'
     entrypoint: 'bash'
@@ -36,7 +37,11 @@ steps:
             .
         else
           echo "Layer cache OFF: forced cold rebuild (--no-build-cache)."
-          DOCKER_BUILDKIT=1 docker build -t "${_IMAGE}" .
+          DOCKER_BUILDKIT=1 docker build \
+            -t "${_IMAGE}" \
+            --no-cache \
+            --build-arg BUILDKIT_INLINE_CACHE=1 \
+            .
         fi
 substitutions:
   _USE_CACHE: '1'
diff --git a/deploy.sh b/deploy.sh
index 602b7611..d04214bc 100755
--- a/deploy.sh
+++ b/deploy.sh
@@ -102,26 +102,55 @@ fmt_hms() {
 # terminal shows nothing for minutes during the image build. Returns the wrapped
 # command's own exit code, so `set -e` still aborts the deploy if the build fails.
 run_with_heartbeat() {
-  local label=$1; shift
+  local label=$1 log_file=$2; shift 2
   local hb_secs=${HEARTBEAT_SECS:-20}
   local start rc=0
   start=$(date +%s)
-  "$@" &
-  local cmd_pid=$!
-  # Heartbeat in a background subshell: it watches the command's PID and exits
-  # when the command does. cmd_pid/start are inherited from this function scope.
-  (
-    while kill -0 "$cmd_pid" 2>/dev/null; do
-      sleep "$hb_secs"
-      kill -0 "$cmd_pid" 2>/dev/null || break
-      echo "    … ${label} still running ($(fmt_hms $(( $(date +%s) - start ))) elapsed)"
-    done
-  ) &
-  local hb_pid=$!
+  # Put the build command and any children in their own process group so an
+  # interrupted deploy can stop the entire upload/build CLI tree without ps.
+  python3 -c 'import os, sys; os.setsid(); os.execvp(sys.argv[1], sys.argv[1:])' \
+    "$@" >"$log_file" 2>&1 &
+  HEARTBEAT_CMD_PID=$!
+  local cmd_pid=$HEARTBEAT_CMD_PID
+  # A single Python watcher has no sleep/tail child to orphan on interruption.
+  # Show gcloud's build-log link as soon as it appears, even though the complete
+  # command output is replayed only after the build finishes.
+  python3 -c '
+import os, sys, time
+pid, interval, label, start, log = int(sys.argv[1]), int(sys.argv[2]), sys.argv[3], int(sys.argv[4]), sys.argv[5]
+next_heartbeat = time.monotonic() + interval
+shown_link = False
+while True:
+    if not shown_link:
+        try:
+            with open(log, encoding="utf-8", errors="replace") as output:
+                for line in output:
+                    if line.startswith("Logs are available at ["):
+                        print(f"    {line.rstrip()}", flush=True)
+                        shown_link = True
+                        break
+        except FileNotFoundError:
+            pass
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        break
+    now = time.monotonic()
+    if now >= next_heartbeat:
+        elapsed = int(time.time()) - start
+        print(f"    … {label} still running ({elapsed // 3600}h {elapsed % 3600 // 60:02}m {elapsed % 60:02}s elapsed)", flush=True)
+        next_heartbeat = now + interval
+    time.sleep(0.5)
+' "$cmd_pid" "$hb_secs" "$label" "$start" "$log_file" &
+  HEARTBEAT_PID=$!
+  local hb_pid=$HEARTBEAT_PID
   wait "$cmd_pid" || rc=$?
   kill "$hb_pid" 2>/dev/null || true
   wait "$hb_pid" 2>/dev/null || true
-  return $rc
+  HEARTBEAT_CMD_PID=""
+  HEARTBEAT_PID=""
+  cat "$log_file"
+  return "$rc"
 }
 
 # Emits the closing timing line for the current phase: wall-clock time of day,
@@ -407,6 +436,82 @@ echo "════════════════════════
 # human think-time at the prompt doesn't pollute the timing deliverable.
 SCRIPT_START=$(date +%s)
 
+UI_BUILD_PID=""; INFRA_SETUP_PID=""; HEARTBEAT_CMD_PID=""; HEARTBEAT_PID=""
+UI_BUILD_LOG=""; INFRA_SETUP_LOG=""; BUILD_SUBMIT_LOG=""
+
+cleanup() {
+  local pid log
+  # The UI and infra jobs start with Bash monitor mode, and the Cloud Build
+  # command starts with setsid. Each PID is therefore its own process-group
+  # leader; signalling its negative PGID stops descendants without ps/pgrep.
+  for pid in "${UI_BUILD_PID:-}" "${INFRA_SETUP_PID:-}" "${HEARTBEAT_CMD_PID:-}"; do
+    [ -n "$pid" ] || continue
+    kill -TERM -- "-$pid" 2>/dev/null || true
+  done
+  if [ -n "${UI_BUILD_PID:-}${INFRA_SETUP_PID:-}${HEARTBEAT_CMD_PID:-}" ]; then
+    sleep 1
+  fi
+  for pid in "${UI_BUILD_PID:-}" "${INFRA_SETUP_PID:-}" "${HEARTBEAT_CMD_PID:-}"; do
+    [ -n "$pid" ] || continue
+    kill -KILL -- "-$pid" 2>/dev/null || true
+    wait "$pid" 2>/dev/null || true
+  done
+  if [ -n "${HEARTBEAT_PID:-}" ]; then
+    kill "$HEARTBEAT_PID" 2>/dev/null || true
+    wait "$HEARTBEAT_PID" 2>/dev/null || true
+  fi
+  for log in "${UI_BUILD_LOG:-}" "${INFRA_SETUP_LOG:-}"; do
+    [ -n "$log" ] || continue
+    if [ -s "$log" ]; then
+      echo "--- background log: $log ---" >&2
+      cat "$log" >&2
+    fi
+    rm -f "$log"
+  done
+  if [ -n "${BUILD_SUBMIT_LOG:-}" ]; then
+    if [ -s "$BUILD_SUBMIT_LOG" ]; then
+      echo "--- background log: $BUILD_SUBMIT_LOG ---" >&2
+      cat "$BUILD_SUBMIT_LOG" >&2
+    fi
+    rm -f "$BUILD_SUBMIT_LOG"
+  fi
+}
+trap cleanup EXIT
+trap 'exit 130' INT
+trap 'exit 143' TERM
+
+# Render UI env + config and kick off the local Angular UI build in the
+# background right away so local CPU work (npm ci / ng build) overlaps with
+# cloud API, service-account, IAM, Cloud Tasks, bucket, and Firestore setup.
+export UI_CONTROL_PLANE_MODE="iap"
+envsubst < ./ui/src/env.template.txt > ./ui/src/env.ts
+generate_config
+if grep -q "controlPlaneMode: 'none'" ./ui/src/env.ts; then
+  echo "ERROR: ui/src/env.ts rendered with controlPlaneMode 'none' (sign-in disabled)." >&2
+  exit 1
+fi
+if [ "$SKIP_UI_BUILD" != "1" ]; then
+  UI_BUILD_LOG=$(mktemp)
+  # A separate job-control process group lets EXIT cleanup stop npm/ng children.
+  set -m
+  (
+    export NG_CLI_ANALYTICS=ci
+    ( cd ui && npm ci )
+    ( cd ui && npx ng build --configuration production )
+  ) </dev/null >"$UI_BUILD_LOG" 2>&1 &
+  UI_BUILD_PID=$!
+  set +m
+elif [ ! -d ui/dist ]; then
+  echo "ERROR: --skip-ui-build given but ui/dist does not exist." >&2
+  echo "       Run a normal deploy once (or 'cd ui && npx ng build') first." >&2
+  exit 1
+elif grep -rqs 'controlPlaneMode:"none"' ui/dist || grep -rqs "controlPlaneMode:'none'" ui/dist; then
+  echo "ERROR: the existing ui/dist was built for local dev (controlPlaneMode 'none'," >&2
+  echo "       sign-in disabled). Refusing to deploy it. Drop --skip-ui-build and run a" >&2
+  echo "       normal deploy to rebuild the UI first." >&2
+  exit 1
+fi
+
 # --- Enable services ---------------------------------------------------------
 # Note: compute.googleapis.com is enabled here so the default Compute Engine
 # service account (used for role bindings below) is guaranteed to exist.
@@ -524,13 +629,14 @@ gcloud services identity create --service=iap.googleapis.com --project=$PROJECT
 # --- Derived values ----------------------------------------------------------
 phase "Resolving project number and runtime service account..."
 PROJECT_NUMBER=$(gcloud projects describe $PROJECT --format="value(projectNumber)")
-# Two distinct identities (least privilege, P2#1):
+# Separate build and runtime identities (P2#1):
 #   BUILD_SA   - the default Compute Engine SA, which is also the default Cloud
 #                Build identity. Used ONLY to build and push the container image
 #                (`gcloud builds submit`, no --service-account). It holds the
 #                build-time roles (artifactregistry.writer, logging.logWriter,
 #                storage.objectUser for the source) and is NOT a request-serving
-#                identity.
+#                identity. On some projects it also inherits project Editor;
+#                these grants do not remove that pre-existing broad access.
 #   RUNTIME_SA - a dedicated SA that the app + worker Cloud Run services run as.
 #                It carries only the roles the running app needs. Crucially it
 #                does NOT get roles/artifactregistry.writer, so a compromise of
@@ -650,8 +756,12 @@ echo "Granting the runtime SA self-impersonation (signBlob + Cloud Tasks OIDC)..
 add_sa_iam_binding "${RUNTIME_SA}" "serviceAccount:${RUNTIME_SA}" "roles/iam.serviceAccountTokenCreator" "$PROJECT"
 add_sa_iam_binding "${RUNTIME_SA}" "serviceAccount:${RUNTIME_SA}" "roles/iam.serviceAccountUser" "$PROJECT"
 
+# --- Non-IAM runtime infrastructure (overlapped with Cloud Build) ---
+INFRA_SETUP_LOG=$(mktemp)
+set -m
+(
 # --- Cloud Tasks queues -------------------------------------------------------
-phase "Setting up Cloud Tasks queues..."
+echo "[>] Setting up Cloud Tasks queues..."
 QUEUES=("Other" "Gemini" "Veo")
 for QUEUE_SUFFIX in "${QUEUES[@]}"; do
   QUEUE_NAME="${TASKS_QUEUE_PREFIX}${QUEUE_SUFFIX}"
@@ -698,7 +808,7 @@ echo "  ✓ ${#QUEUES[@]} Cloud Tasks queues ready (${QUEUES[*]/#/${TASKS_QUEUE_
 # deployer sets ADOPT_EXISTING_BUCKET=1 — so a shared bucket cannot have its
 # contents exposed through signed URLs. (Needs roles/storage.admin, already
 # required.)
-phase "Setting up GCS bucket..."
+echo "[>] Setting up GCS bucket..."
 DEFAULT_BUCKET="${PROJECT}-scene-machine"
 if ! gcloud storage buckets describe "gs://$GCS_BUCKET" --project=$PROJECT &> /dev/null; then
     echo "Creating dedicated GCS bucket gs://$GCS_BUCKET in ${REGION}..."
@@ -739,7 +849,7 @@ else
 fi
 
 # --- Firestore databases (two) -------------------------------------------------
-phase "Setting up Firestore databases..."
+echo "[>] Setting up Firestore databases..."
 if ! gcloud firestore databases describe --database="$FIRESTORE_DB" --project=$PROJECT &> /dev/null; then
     echo "Creating Firestore database: $FIRESTORE_DB"
     gcloud firestore databases create --database="$FIRESTORE_DB" --project=$PROJECT --location="$REGION"
@@ -764,38 +874,24 @@ else
     gcloud firestore databases describe --database="$FIRESTORE_DB_UI" --project=$PROJECT --format="value(locationId)"
 fi
 
-# --- Render UI env + config (must precede the single image build) -------------
-# Order matters: these artifacts are baked into the image (Dockerfile
-# `COPY . .`), so they must exist before `gcloud builds submit`.
+# --- SceneMachineUser custom role -----------------------------------------------
+echo "[>] Ensuring SceneMachineUser custom role matches user-role.yaml..."
+if ! gcloud iam roles describe SceneMachineUser --project=$PROJECT &> /dev/null; then
+  echo "SceneMachineUser role doesn't exist. Creating it..."
+  gcloud iam roles create SceneMachineUser --project=$PROJECT --file=./user-role.yaml
+else
+  echo "SceneMachineUser role exists. Syncing it to user-role.yaml..."
+  gcloud iam roles update SceneMachineUser --project=$PROJECT --file=./user-role.yaml --quiet || true
+fi
+
+) </dev/null >"$INFRA_SETUP_LOG" 2>&1 &
+INFRA_SETUP_PID=$!
+set +m
+
+# --- UI env + config (rendered at SCRIPT_START before UI_BUILD_PID) -----------
 phase "Rendering ui/src/env.ts and ui/definitions/config.json..."
-# IAP is the only deployable front-door mode (controlPlaneMode 'none' is local
-# dev only), and the data plane is always mediated, so there is nothing to
-# choose here — the UI is always built for IAP.
-export UI_CONTROL_PLANE_MODE="iap"
 echo "  Front-door auth: IAP (the only deployable mode)"
-# env.ts: UI_CONTROL_PLANE_MODE
-# is additionally exported for the front-door env.template.txt field
-# (controlPlaneMode) — a no-op against templates that don't reference it.
-envsubst < ./ui/src/env.template.txt > ./ui/src/env.ts
-# config.json: read by the backend (orch.py) for the project/bucket/database
-# params and rendered into the deploy. The app serves the SPA, /api and the
-# status viewer from one Cloud Run service, so the browser always calls /api
-# RELATIVE to wherever the page loaded; no app host is baked in. Only
-# $FIRESTORE_DB / $GCS_BUCKET / $PROJECT / $REGION / $TASKS_QUEUE_PREFIX are
-# substituted.
-generate_config
 
-# Safety: never build or ship a UI rendered for LOCAL DEV (controlPlaneMode
-# 'none' turns the sign-in gate off). The line above always sets 'iap', so this
-# only trips on a stray UI_CONTROL_PLANE_MODE override; fail loudly rather than
-# deploy an app with authentication disabled.
-if grep -q "controlPlaneMode: 'none'" ./ui/src/env.ts; then
-  echo "ERROR: ui/src/env.ts rendered with controlPlaneMode 'none' (sign-in disabled)." >&2
-  echo "       Refusing to build a deploy with the front-door auth gate off." >&2
-  echo "       This should not happen on a normal deploy; check for a stray" >&2
-  echo "       UI_CONTROL_PLANE_MODE in your environment, then re-run $0." >&2
-  exit 1
-fi
 
 # --- UI build ------------------------------------------------------------------
 if [ "$SKIP_UI_BUILD" = "1" ]; then
@@ -817,12 +913,14 @@ if [ "$SKIP_UI_BUILD" = "1" ]; then
   echo "[skip] Building the Angular UI — skipped (--skip-ui-build); reusing ui/dist."
 else
   phase "Building the Angular UI (npm ci + ng build)..."
-  export NG_CLI_ANALYTICS=ci
-  (
-    cd ui \
-      && npm ci \
-      && npx ng build --configuration production
-  )
+  if ! wait "$UI_BUILD_PID"; then
+    UI_BUILD_PID=""
+    exit 1
+  fi
+  UI_BUILD_PID=""
+  cat "$UI_BUILD_LOG"
+  rm -f "$UI_BUILD_LOG"
+  UI_BUILD_LOG=""
 fi
 
 # --- Version stamp + Artifact Registry + ONE image build -----------------------
@@ -837,6 +935,8 @@ if ! gcloud artifacts repositories describe "${ARTIFACT_REPO}" --project=$PROJEC
   echo "Creating artifact repository: $ARTIFACT_REPO"
   gcloud artifacts repositories create "${ARTIFACT_REPO}" --repository-format=docker --project=$PROJECT --location="$REGION"
 fi
+# The default public Cloud Build pool avoids the high-CPU pool's measured
+# provisioning queue on both cached and no-cache builds.
 # NOTE: the repo's .gcloudignore excludes ui/* but re-includes ui/dist/ and
 # ui/remix-engine-status-viewer/ — both are LOAD-BEARING for this build: the
 # front-door app service serves the built SPA (ui/dist/ui/browser) and the
@@ -862,40 +962,87 @@ if [ "$NO_BUILD_CACHE" = "1" ]; then
 else
   echo "  Docker layer cache: ON (reuses unchanged layers from the previous image)."
 fi
-run_with_heartbeat "Cloud Build" \
-  gcloud builds submit . --config=cloudbuild.yaml --substitutions="$BUILD_SUBS" \
-    --project=$PROJECT --region=$REGION
+BUILD_ATTEMPT=0
+BUILD_MAX_ATTEMPTS=4
+while true; do
+  BUILD_ATTEMPT=$((BUILD_ATTEMPT + 1))
+  BUILD_SUBMIT_LOG=$(mktemp)
+  if run_with_heartbeat "Cloud Build" "$BUILD_SUBMIT_LOG" \
+    gcloud builds submit . --config=cloudbuild.yaml --substitutions="$BUILD_SUBS" \
+      --project=$PROJECT --region=$REGION; then
+    # Identify this build, not whatever a concurrent deploy later writes to
+    # :latest. Fail closed if gcloud's success output lacks its build URL.
+    if ! BUILD_ID=$(python3 deploy/resolve_build_image.py build-id \
+      "$PROJECT" "$REGION" "$BUILD_SUBMIT_LOG"); then
+      rm -f "$BUILD_SUBMIT_LOG"
+      BUILD_SUBMIT_LOG=""
+      echo "ERROR: Cloud Build succeeded but its ID could not be verified; refusing to deploy a mutable image tag." >&2
+      exit 1
+    fi
+    rm -f "$BUILD_SUBMIT_LOG"
+    BUILD_SUBMIT_LOG=""
+    break
+  fi
+  if [ "$BUILD_ATTEMPT" -lt "$BUILD_MAX_ATTEMPTS" ] \
+      && grep -qiE 'PERMISSION_DENIED|permission_denied|HTTPError 403|HTTP[[:space:]/:]+403|status[[:space:]:=]+403|does not have storage\.objects' "$BUILD_SUBMIT_LOG"; then
+    rm -f "$BUILD_SUBMIT_LOG"
+    BUILD_SUBMIT_LOG=""
+    BUILD_RETRY_DELAY=$((BUILD_ATTEMPT * 15))
+    echo "  ⚠ Cloud Build hit transient IAM propagation delay (attempt ${BUILD_ATTEMPT}/${BUILD_MAX_ATTEMPTS}); retrying in ${BUILD_RETRY_DELAY}s..."
+    sleep "$BUILD_RETRY_DELAY"
+    continue
+  fi
+  rm -f "$BUILD_SUBMIT_LOG"
+  BUILD_SUBMIT_LOG=""
+  exit 1
+done
+
+# Cloud Build reports the digest of the image it actually pushed. Using that
+# immutable image for both services prevents a later :latest push from changing
+# which code this deploy promotes between the worker and app steps.
+if ! IMAGE_DIGEST=$(gcloud builds describe "$BUILD_ID" \
+  --project="$PROJECT" --region="$REGION" --format=json \
+  | python3 deploy/resolve_build_image.py digest "$IMAGE"); then
+  echo "ERROR: could not verify the image digest from build ${BUILD_ID}; refusing to deploy a mutable image tag." >&2
+  exit 1
+fi
+DEPLOY_IMAGE="${IMAGE%:*}@${IMAGE_DIGEST}"
+echo "✓ Verified build ${BUILD_ID}; deploying immutable image ${DEPLOY_IMAGE}"
+
+phase "Completing overlapped infrastructure setup..."
+if ! wait "$INFRA_SETUP_PID"; then
+  INFRA_SETUP_PID=""
+  exit 1
+fi
+INFRA_SETUP_PID=""
+cat "$INFRA_SETUP_LOG"
+rm -f "$INFRA_SETUP_LOG"
+INFRA_SETUP_LOG=""
 
 # --- Cloud Run: worker (private, Cloud-Tasks-invoked) --------------------------
+# Cloud Run natively serves https://worker-${PROJECT_NUMBER}.${REGION}.run.app
+# on both first deploy and all subsequent deploys (listed in
+# metadata.annotations."run.googleapis.com/urls"). Using the deterministic
+# regional URL consistently avoids a second app revision rollout on cold deploy
+# and keeps warm redeploys 100% idempotent.
+WORKER_URL="https://worker-${PROJECT_NUMBER}.${REGION}.run.app"
 if [ "$APP_ONLY" = "1" ]; then
   phase "Reusing existing 'worker' Cloud Run service (--app-only)..."
   echo "[skip] Deploying 'worker' — skipped (--app-only); reusing the live service."
-  # The app deploy below needs WORKER_URL; read it from the live worker.
-  WORKER_URL=$(gcloud run services describe worker --region=$REGION --project=$PROJECT --format='value(status.url)' 2>/dev/null || true)
-  if [ -z "$WORKER_URL" ]; then
+  if ! gcloud run services describe worker --region=$REGION --project=$PROJECT >/dev/null 2>&1; then
     echo "ERROR: --app-only given but no existing 'worker' service in ${PROJECT}/${REGION}." >&2
     echo "       Deploy once without --app-only, then re-run with --app-only." >&2
     exit 1
   fi
   echo "  Reusing worker: ${WORKER_URL}"
 else
-  phase "Deploying 'worker' Cloud Run service (private)..."
-  # GUNICORN_TIMEOUT just above the worker's 1800s Cloud Run request timeout so
-  # gunicorn reaps a thread only AFTER Cloud Run has already returned, never
-  # killing a legitimate long render mid-flight. (D7)
-  gcloud run deploy worker --image "$IMAGE" --region $REGION --project $PROJECT \
+  phase "Deploying 'worker' Cloud Run service..."
+  gcloud run deploy worker --image "$DEPLOY_IMAGE" --region "$REGION" --project "$PROJECT" \
     --cpu=8 --memory=16G --timeout=1800 --no-allow-unauthenticated \
     --service-account="$RUNTIME_SA" \
     --set-env-vars=ROLE=worker,GUNICORN_TIMEOUT=1830
-  WORKER_URL=$(gcloud run services describe worker --region=$REGION --project=$PROJECT --format='value(status.url)')
-  echo "✓ Worker deployed: ${WORKER_URL}"
-
-  # The only run.invoker grant the runtime SA gets: service-scoped to the
-  # worker, exactly what the Cloud-Tasks-minted OIDC tokens need to invoke it.
-  # (There is no project-wide run.invoker, so the app cannot invoke other
-  # Cloud Run services.)
-  echo "Granting service-scoped run.invoker on 'worker' to ${RUNTIME_SA}..."
   add_run_invoker_binding worker "$REGION" "$PROJECT" "serviceAccount:${RUNTIME_SA}"
+  echo "✓ Worker deployed: ${WORKER_URL}"
 fi
 
 # --- Cloud Run: app (UI + same-origin /api control plane) -----------------------
@@ -904,11 +1051,11 @@ IAP_FLAG_AVAILABLE=true
 # IAP front door. The --iap flag (built-in IAP for Cloud Run, GA March 2026) may
 # not exist on older gcloud installs — gate it behind a CLI capability check and
 # fall back to a private deploy + manual enable instruction.
-if ! gcloud run deploy --help 2>/dev/null | grep -q -- '--iap'; then
+if ! gcloud run deploy --help 2>/dev/null | grep -- '--iap' >/dev/null; then
   IAP_FLAG_AVAILABLE=false
 fi
 if [ "$IAP_FLAG_AVAILABLE" = "true" ]; then
-  gcloud run deploy app --image "$IMAGE" --region $REGION --project $PROJECT \
+  gcloud run deploy app --image "$DEPLOY_IMAGE" --region $REGION --project $PROJECT \
     --cpu=2 --memory=2Gi --timeout=300 --min-instances=${APP_MIN_INSTANCES} --no-allow-unauthenticated --iap \
     --service-account="$RUNTIME_SA" \
     --set-env-vars=ROLE=app,AUTH_MODE=iap,WORKER_URL=${WORKER_URL},IAP_AUDIENCE=${IAP_AUDIENCE},FIRESTORE_DB_UI=${FIRESTORE_DB_UI},DICTATION_ENABLED=${DICTATION_ENABLED},DICTATION_MODE=${DICTATION_MODE}
@@ -920,7 +1067,7 @@ else
   # step if 'services update' lacks --iap too.
   echo "⚠ 'gcloud run deploy' lacks --iap; deploying the app private, then"
   echo "  enabling IAP via 'gcloud run services update'."
-  gcloud run deploy app --image "$IMAGE" --region $REGION --project $PROJECT \
+  gcloud run deploy app --image "$DEPLOY_IMAGE" --region $REGION --project $PROJECT \
     --cpu=2 --memory=2Gi --timeout=300 --min-instances=${APP_MIN_INSTANCES} --no-allow-unauthenticated \
     --service-account="$RUNTIME_SA" \
     --set-env-vars=ROLE=app,AUTH_MODE=iap,WORKER_URL=${WORKER_URL},IAP_AUDIENCE=${IAP_AUDIENCE},FIRESTORE_DB_UI=${FIRESTORE_DB_UI},DICTATION_ENABLED=${DICTATION_ENABLED},DICTATION_MODE=${DICTATION_MODE}
@@ -941,49 +1088,14 @@ echo "Granting service-scoped run.invoker on 'app' to the IAP service agent..."
 add_run_invoker_binding app "$REGION" "$PROJECT" "serviceAccount:${IAP_SA}"
 APP_URL=$(gcloud run services describe app --region=$REGION --project=$PROJECT --format='value(status.url)')
 echo "✓ App deployed: ${APP_URL}"
-# Nothing was predicted: the image carries only same-origin URLs, so the app and
-# its status viewer work on this first deploy. The actual Cloud Run hosts,
-# known only now, feed the GCS bucket CORS list below, so the browser
-# can fetch signed media URLs cross-origin from every actual app service origin.
-APP_URL_METADATA=$(gcloud run services describe app \
-  --region=$REGION --project=$PROJECT \
-  --format='json(metadata.annotations."run.googleapis.com/urls")')
-UI_CORS_ORIGINS=$(printf '%s' "$APP_URL_METADATA" \
-  | python3 ./deploy/render_cors_origins.py)
-
-# --- Bucket CORS (needs the actual app service origins) -------------------------
-phase "Applying GCS bucket CORS for returned Cloud Run origins..."
-export UI_CORS_ORIGINS
-envsubst < ./gcs-cors-config.template.json > ./gcs-cors-config.json
-gcloud storage buckets update gs://$GCS_BUCKET --cors-file=./gcs-cors-config.json --project=$PROJECT
-
-# --- SceneMachineUser custom role -----------------------------------------------
-phase "Ensuring SceneMachineUser custom role matches user-role.yaml..."
-if ! gcloud iam roles describe SceneMachineUser --project=$PROJECT &> /dev/null; then
-  echo "SceneMachineUser role doesn't exist. Creating it..."
-  gcloud iam roles create SceneMachineUser --project=$PROJECT --file=./user-role.yaml
-else
-  # Update (not skip) so an edited user-role.yaml — e.g. the slimmed
-  # IAP-access-only permission set — actually takes effect on a project where the
-  # role already exists, instead of being silently ignored. '|| true' tolerates
-  # the benign "no changes to apply" case on a re-deploy; the role keeps its
-  # IAP-access permission regardless, so user admission is never at risk here.
-  echo "SceneMachineUser role exists. Syncing it to user-role.yaml..."
-  gcloud iam roles update SceneMachineUser --project=$PROJECT --file=./user-role.yaml --quiet || true
-fi
 
+phase "Seeding Firestore config after both services are deployed..."
 # --- Seed Firestore config (front-door topology) --------------------------------
-# Uses firestore_config_frontdoor.template.json: the UI Firestore config doc with
-# the backend base fixed to the same-origin '/api' and no API key. Owner-
-# credential REST writes bypass the (deliberately read-only) config rules — by
-# design.
-phase "Adding default Scene Machine configurations to Firestore..."
-# Capture the HTTP status (as the other REST calls in this script do): a non-200
-# here means the UI's same-origin '/api' wiring was NOT written, so fail loudly
-# instead of reporting a successful deploy with a broken app config.
+echo "[>] Adding default Scene Machine configurations to Firestore..."
+ADC_TOKEN=$(gcloud auth application-default print-access-token)
 CONFIG_SEED_STATUS=$(curl -s -X PATCH \
 "https://firestore.googleapis.com/v1/projects/${PROJECT}/databases/${FIRESTORE_DB_UI}/documents/config/global" \
-  -H "Authorization: Bearer $(gcloud auth application-default print-access-token)" \
+  -H "Authorization: Bearer ${ADC_TOKEN}" \
   -H "x-goog-user-project: ${PROJECT}" \
   -H "Content-Type: application/json" \
   -o /dev/null -w '%{http_code}' \
@@ -994,13 +1106,9 @@ if [ "$CONFIG_SEED_STATUS" != "200" ]; then
   exit 1
 fi
 
-# The model catalog: config/models is overwritten from the repo file on every
-# deploy. Operators may edit the live doc between deploys; the pre-flight
-# preview above showed what this write replaces. Same fail-loudly contract as
-# the config/global seed.
 MODELS_SEED_STATUS=$(python3 scripts/seed_config_models.py convert < ui/definitions/models.json | curl -s -X PATCH \
 "https://firestore.googleapis.com/v1/projects/${PROJECT}/databases/${FIRESTORE_DB_UI}/documents/config/models" \
-  -H "Authorization: Bearer $(gcloud auth application-default print-access-token)" \
+  -H "Authorization: Bearer ${ADC_TOKEN}" \
   -H "x-goog-user-project: ${PROJECT}" \
   -H "Content-Type: application/json" \
   -o /dev/null -w '%{http_code}' \
@@ -1011,11 +1119,8 @@ if [ "$MODELS_SEED_STATUS" != "200" ]; then
   exit 1
 fi
 
-# The announcement is operator-authored after the first deploy. Firestore's
-# create operation makes the initial seed race-safe and returns 409 when an
-# operator document already exists; either result is a successful deploy.
 if ! ANNOUNCEMENT_SEED_STATUS=$(GOOGLE_CLOUD_PROJECT="$PROJECT" \
-  GOOGLE_OAUTH_ACCESS_TOKEN="$(gcloud auth application-default print-access-token)" \
+  GOOGLE_OAUTH_ACCESS_TOKEN="${ADC_TOKEN}" \
   python3 scripts/seed_announcement.py seed \
   "https://firestore.googleapis.com/v1/projects/${PROJECT}/databases/${FIRESTORE_DB_UI}/documents/config?documentId=announcement" \
   "$ANNOUNCEMENT_MARKDOWN_FILE" "$ANNOUNCEMENT_ENABLED"); then
@@ -1025,14 +1130,12 @@ if ! ANNOUNCEMENT_SEED_STATUS=$(GOOGLE_CLOUD_PROJECT="$PROJECT" \
 fi
 
 for template in creative_templates/*.json; do
-  # Skip cleanly if the directory is empty/absent: without 'nullglob' the glob
-  # would otherwise stay literal and run the body once on a non-existent file.
   [ -e "$template" ] || continue
   template_name=$(basename "$template" .json)
 
   TEMPLATE_SEED_STATUS=$(curl -s -X PATCH \
   "https://firestore.googleapis.com/v1/projects/${PROJECT}/databases/${FIRESTORE_DB_UI}/documents/creativeTemplates/${template_name}" \
-    -H "Authorization: Bearer $(gcloud auth application-default print-access-token)" \
+    -H "Authorization: Bearer ${ADC_TOKEN}" \
     -H "x-goog-user-project: ${PROJECT}" \
     -H "Content-Type: application/json" \
     -o /dev/null -w '%{http_code}' \
@@ -1042,6 +1145,21 @@ for template in creative_templates/*.json; do
     exit 1
   fi
 done
+# Nothing was predicted: the image carries only same-origin URLs, so the app and
+# its status viewer work on this first deploy. The actual Cloud Run hosts,
+# known only now, feed the GCS bucket CORS list below, so the browser
+# can fetch signed media URLs cross-origin from every actual app service origin.
+APP_URL_METADATA=$(gcloud run services describe app \
+  --region=$REGION --project=$PROJECT \
+  --format='json(metadata.annotations."run.googleapis.com/urls")')
+UI_CORS_ORIGINS=$(printf '%s' "$APP_URL_METADATA" \
+  | python3 ./deploy/render_cors_origins.py)
+
+# --- Bucket CORS (needs the actual app service origins) -------------------------
+phase "Applying GCS bucket CORS for returned Cloud Run origins..."
+export UI_CORS_ORIGINS
+envsubst < ./gcs-cors-config.template.json > ./gcs-cors-config.json
+gcloud storage buckets update gs://$GCS_BUCKET --cors-file=./gcs-cors-config.json --project=$PROJECT
 
 # --- Automated provisioning complete: timing checkpoint ----------------------------
 close_phase
diff --git a/deploy/libs.sh b/deploy/libs.sh
index 928bc444..05b8de31 100644
--- a/deploy/libs.sh
+++ b/deploy/libs.sh
@@ -200,17 +200,28 @@ add_iam_binding() {
   fi
 
   if [ -n "$role" ] && [ -n "$member" ] && [ -n "$project" ]; then
-    if gcloud projects get-iam-policy "$project" \
-        --flatten="bindings[].members" \
-        --filter="bindings.role=${role} AND bindings.members=${member}" \
-        --format="value(bindings.role)" 2>/dev/null | grep -q .; then
+    if [ "${_CACHED_IAM_PROJECT:-}" != "$project" ]; then
+      if _CACHED_PROJECT_IAM_POLICY=$(gcloud projects get-iam-policy "$project" \
+          --flatten="bindings[].members" \
+          --format="value(bindings.role,bindings.members)" 2>/dev/null); then
+        _CACHED_IAM_PROJECT="$project"
+      else
+        _CACHED_IAM_PROJECT=""
+        _CACHED_PROJECT_IAM_POLICY=""
+      fi
+    fi
+    if [ "${_CACHED_IAM_PROJECT:-}" = "$project" ] \
+        && grep -Fqx "${role}"$'\t'"${member}" <<<"$_CACHED_PROJECT_IAM_POLICY"; then
       echo "  ✓ ${member} already has ${role} — skipping."
       return 0
     fi
   fi
 
   _retry_iam_write "$label" "$propagating_runtime_sa" \
-    gcloud projects add-iam-policy-binding "$@"
+    gcloud projects add-iam-policy-binding "$@" || return $?
+  if [ "${_CACHED_IAM_PROJECT:-}" = "$project" ] && [ -n "$role" ] && [ -n "$member" ]; then
+    _CACHED_PROJECT_IAM_POLICY="${_CACHED_PROJECT_IAM_POLICY}"$'\n'"${role}"$'\t'"${member}"
+  fi
 }
 
 # Service-scoped run.invoker on a Cloud Run SERVICE, with the same pre-check +
diff --git a/deploy/resolve_build_image.py b/deploy/resolve_build_image.py
new file mode 100644
index 00000000..8114c25b
--- /dev/null
+++ b/deploy/resolve_build_image.py
@@ -0,0 +1,111 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Resolve the immutable image produced by this deployment's Cloud Build.
+
+The build still pushes :latest for the next build's layer cache. Cloud Run gets
+the digest from the successful build receipt, so a later :latest push cannot
+change which image the worker and app use within this deployment.
+"""
+
+import json
+import pathlib
+import re
+import sys
+from typing import Any
+
+
+def build_id(project: str, region: str, log: str) -> str:
+  """Read the sole build ID emitted for this project and region."""
+  output = pathlib.Path(log).read_text(encoding="utf-8", errors="replace")
+  # gcloud 568.0.0 emits this status line. It is not a documented output API,
+  # so reject missing or ambiguous lines instead of deploying a mutable tag.
+  pattern = re.compile(
+      r"^Created \[https://cloudbuild\.googleapis\.com/v1/projects/"
+      + re.escape(project)
+      + r"/locations/"
+      + re.escape(region)
+      + r"/builds/([0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12})\]\.\s*$",
+      re.MULTILINE,
+  )
+  matches = pattern.findall(output)
+  if len(matches) != 1:
+    raise ValueError(
+        "expected exactly one 'Created [https://cloudbuild.googleapis.com/"
+        f"v1/projects/{project}/locations/{region}/builds/BUILD_ID].' line "
+        "(verified with gcloud 568.0.0). This build was not promoted, "
+        "and background infrastructure setup may be incomplete. "
+        "Recommended recovery: verify gcloud output and rerun this "
+        "idempotent script after updating gcloud or its parser. Manual "
+        "recovery requires first verifying/completing every infrastructure "
+        "step before the Cloud Run phase in deploy.sh; use only the exact "
+        "build ID in this run's unique Cloud Console log URL, and confirm "
+        "SUCCESS plus the matching image digest with 'gcloud builds "
+        f"describe BUILD_ID --project={project} --region={region} "
+        "--format=json'. Then follow all worker, app, invoker, seed and "
+        "CORS steps in deploy.sh in order, using the image name without "
+        "':latest' plus '@sha256:DIGEST'. If any step cannot be verified, "
+        "stop and rerun. Never deploy :latest or guess a build from a list."
+    )
+  return matches[0]
+
+
+def digest(image: str, receipt: dict[str, Any]) -> str:
+  """Return the exact tagged image digest from a successful build."""
+  if not isinstance(receipt, dict):
+    raise ValueError("Cloud Build receipt must be an object")
+  if receipt.get("status") != "SUCCESS":
+    raise ValueError("Cloud Build is not successful")
+  results = receipt.get("results")
+  if not isinstance(results, dict) or not isinstance(
+      results.get("images"), list
+  ):
+    raise ValueError("Cloud Build receipt has no image results")
+  images = results["images"]
+  if any(not isinstance(entry, dict) for entry in images):
+    raise ValueError("Cloud Build receipt contains an invalid image result")
+  matches = [
+      entry.get("digest", "") for entry in images if entry.get("name") == image
+  ]
+  if (
+      len(matches) != 1
+      or not isinstance(matches[0], str)
+      or not re.fullmatch(r"sha256:[0-9a-f]{64}", matches[0])
+  ):
+    raise ValueError(
+        "expected exactly one valid digest for the requested image"
+    )
+  return matches[0]
+
+
+def main() -> int:
+  """Resolve one build ID or image digest, failing closed on bad input."""
+  try:
+    if len(sys.argv) == 5 and sys.argv[1] == "build-id":
+      print(build_id(sys.argv[2], sys.argv[3], sys.argv[4]))
+    elif len(sys.argv) == 3 and sys.argv[1] == "digest":
+      print(digest(sys.argv[2], json.load(sys.stdin)))
+    else:
+      raise ValueError(
+          "usage: resolve_build_image.py build-id PROJECT REGION LOG | digest"
+          " IMAGE"
+      )
+  except (OSError, ValueError, KeyError, TypeError) as exc:
+    print(f"ERROR: {exc}", file=sys.stderr)
+    return 1
+  return 0
+
+
+if __name__ == "__main__":
+  sys.exit(main())
diff --git a/test/test_deploy_safety.py b/test/test_deploy_safety.py
index f007b131..99ce42be 100644
--- a/test/test_deploy_safety.py
+++ b/test/test_deploy_safety.py
@@ -11,6 +11,9 @@
 import os
 import subprocess
 import sys
+import signal
+import time
+import textwrap
 
 import pytest
 
@@ -769,3 +772,600 @@ def test_config_validation_regex_accepts_quoted_and_rejects_empty(
   )
   assert proc.stderr == "", f"Bash error evaluating grep command: {proc.stderr}"
   assert (proc.returncode == 0) == expected_match
+
+
+def _verify_dockerfile_security_invariants(dockerfile: str) -> None:
+  """Validate digest pinning and uv hash/wheel flags on a Dockerfile string."""
+  for syntax_ref in re.findall(
+      r"^\s*#\s*syntax\s*=\s*(\S+)", dockerfile, re.MULTILINE | re.IGNORECASE
+  ):
+    assert (
+        "@sha256:" in syntax_ref
+    ), f"Unpinned # syntax= frontend image in Dockerfile: {syntax_ref}"
+
+  # Strip comment lines and join backslash continuations before parsing rules.
+  uncommented_lines = [
+      line
+      for line in dockerfile.splitlines()
+      if not line.lstrip().startswith("#")
+  ]
+  normalized = re.sub(r"\\\s*\n", " ", "\n".join(uncommented_lines))
+
+  stage_names = set()
+  stage_count = 0
+  for line in normalized.splitlines():
+    from_match = re.match(
+        r"^\s*FROM\s+(?:--\S+\s+)*(\S+)(?:\s+AS\s+(\S+))?",
+        line,
+        re.IGNORECASE,
+    )
+    if from_match:
+      ref, alias = from_match.groups()
+      is_prior_stage = ref.lower() in stage_names or (
+          ref.isdigit() and int(ref) < stage_count
+      )
+      if not is_prior_stage:
+        assert re.search(r"@sha256:[0-9a-f]{64}$", ref), (
+            f"Unpinned FROM image in Dockerfile: {ref}"
+        )
+      stage_count += 1
+      if alias:
+        stage_names.add(alias.lower())
+
+    for ref in re.findall(
+        r"\bCOPY\s+(?:--(?!from=)\S+\s+)*--from=(\S+)", line, re.IGNORECASE
+    ):
+      is_prior_stage = ref.lower() in stage_names or (
+          ref.isdigit() and int(ref) < stage_count
+      )
+      if not is_prior_stage:
+        assert re.search(r"@sha256:[0-9a-f]{64}$", ref), (
+            f"Unpinned COPY --from image in Dockerfile: {ref}"
+        )
+  assert stage_count, "Expected at least one FROM instruction in Dockerfile"
+
+  uv_install_cmds = [
+      line
+      for line in normalized.splitlines()
+      if re.search(r"^\s*RUN\b.*\buv\s+pip\s+install\b", line, re.IGNORECASE)
+  ]
+  assert uv_install_cmds, "Expected a RUN ... uv pip install instruction"
+  for cmd in uv_install_cmds:
+    assert "--require-hashes" in cmd, f"Missing --require-hashes in: {cmd}"
+    assert (
+        "--only-binary :all:" in cmd or "--only-binary=:all:" in cmd
+    ), f"Missing --only-binary :all: in: {cmd}"
+
+
+def test_dockerfile_external_images_are_digest_pinned_and_hash_verified():
+  """External Dockerfile images must be @sha256-pinned and uv verified."""
+  dockerfile = _dockerfile()
+  _verify_dockerfile_security_invariants(dockerfile)
+
+  # Self-verifying mutation checks: ensure each regression fails the check.
+  with pytest.raises(AssertionError, match="Unpinned # syntax="):
+    _verify_dockerfile_security_invariants(
+        "# syntax=docker/dockerfile:1\n" + dockerfile
+    )
+  with pytest.raises(AssertionError, match="Missing --require-hashes"):
+    _verify_dockerfile_security_invariants(
+        dockerfile.replace("--require-hashes", "") + "\n# --require-hashes\n"
+    )
+  with pytest.raises(AssertionError, match="Missing --only-binary :all:"):
+    _verify_dockerfile_security_invariants(
+        dockerfile.replace("--only-binary :all:", "")
+    )
+  with pytest.raises(AssertionError, match="Unpinned FROM image"):
+    _verify_dockerfile_security_invariants(
+        dockerfile.replace(
+            "FROM runtime-base AS final", "from alpine:latest as final"
+        )
+    )
+  with pytest.raises(AssertionError, match="Unpinned FROM image"):
+    _verify_dockerfile_security_invariants(
+        dockerfile.replace(
+            "FROM runtime-base AS final",
+            "FROM --platform=linux/amd64@sha256:0000 unpinned:latest AS final",
+        )
+    )
+  with pytest.raises(AssertionError, match="Unpinned FROM image"):
+    _verify_dockerfile_security_invariants(
+        "FROM alpine AS injected\n"
+        + dockerfile.replace("FROM runtime-base AS final", "FROM runtime-base AS alpine")
+    )
+  for bad_ref in ("alpine:latest@sha256:", "alpine:latest@sha256:not-a-digest"):
+    with pytest.raises(AssertionError, match="Unpinned FROM image"):
+      _verify_dockerfile_security_invariants(
+          dockerfile.replace(
+              "FROM runtime-base AS final", f"FROM {bad_ref} AS final"
+          )
+      )
+
+
+def test_add_iam_binding_caches_policy_and_recovers_after_fetch_error(tmp_path):
+  """add_iam_binding caches get-iam-policy without poisoning on error."""
+  fake_bin = tmp_path / "bin"
+  fake_bin.mkdir()
+  calls_log = tmp_path / "gcloud_calls.log"
+  fail_flag = tmp_path / "fail_first_get"
+  fail_flag.write_text("1")
+
+  sa_member = "serviceAccount:sm-runtime@p1.iam.gserviceaccount.com"
+  fake_gcloud = fake_bin / "gcloud"
+  fake_gcloud.write_text(
+      "#!/usr/bin/env bash\n"
+      f'echo "$*" >> "{calls_log}"\n'
+      'if [[ "$1 $2" == "projects get-iam-policy" ]]; then\n'
+      f'  if [[ -f "{fail_flag}" ]]; then\n'
+      f'    rm -f "{fail_flag}"\n'
+      "    exit 1\n"
+      "  fi\n"
+      f'  printf "roles/datastore.user\\t{sa_member}\\n"\n'
+      "  exit 0\n"
+      "fi\n"
+      'if [[ "$1 $2" == "projects add-iam-policy-binding"'
+      ' && "$*" == *"roles/run.admin"* ]]; then\n'
+      "  exit 1\n"
+      "fi\n"
+      "exit 0\n"
+  )
+  fake_gcloud.chmod(0o755)
+
+  libs_sh = _REPO / "deploy" / "libs.sh"
+  script = f"""
+    set -euo pipefail
+    export PATH="{fake_bin}:$PATH"
+    source "{libs_sh}"
+    add_iam_binding p1 --member="{sa_member}" --role="roles/logging.logWriter"
+    add_iam_binding p1 --member="{sa_member}" --role="roles/datastore.user"
+    add_iam_binding p1 --member="{sa_member}" --role="roles/aiplatform.user"
+    add_iam_binding p1 --member="{sa_member}" --role="roles/aiplatform.user"
+    if add_iam_binding p1 --member="{sa_member}" --role="roles/run.admin"; then
+      echo "UNEXPECTED_ADD_SUCCESS" >&2
+      exit 1
+    fi
+    if grep -Fq "roles/run.admin" <<<"$_CACHED_PROJECT_IAM_POLICY"; then
+      echo "CACHE_POISONED" >&2
+      exit 1
+    fi
+  """
+  proc = subprocess.run(
+      ["bash", "-c", script], capture_output=True, text=True, check=False
+  )
+  assert proc.returncode == 0, f"Script failed: {proc.stderr}"
+  calls = calls_log.read_text().splitlines()
+  get_calls = [c for c in calls if c.startswith("projects get-iam-policy")]
+  add_calls = [
+      c
+      for c in calls
+      if c.startswith("projects add-iam-policy-binding")
+      and "roles/run.admin" not in c
+  ]
+  assert len(get_calls) == 2, f"Expected 2 get-iam-policy calls, got: {calls}"
+  assert (
+      len(add_calls) == 2
+  ), f"Expected 2 add-iam-policy-binding calls, got: {calls}"
+
+
+def test_deploy_cleanup_trap_dumps_logs_and_unlinks_on_abort(tmp_path):
+  """The EXIT trap in deploy.sh dumps background logs and reaps jobs."""
+  deploy_sh = (_REPO / "deploy.sh").read_text(encoding="utf-8")
+  match = re.search(
+      r"(UI_BUILD_PID=\"\";.*?trap cleanup EXIT)", deploy_sh, re.DOTALL
+  )
+  assert match, "Expected cleanup() and trap cleanup EXIT in deploy.sh"
+  trap_block = match.group(1)
+  assert re.search(r'UI_BUILD_LOG=\$\(mktemp\)\s+.*?set -m\s+\(', deploy_sh, re.DOTALL)
+  assert re.search(r'UI_BUILD_PID=\$!\s+set \+m', deploy_sh)
+  assert re.search(r'INFRA_SETUP_LOG=\$\(mktemp\)\s+set -m\s+\(', deploy_sh)
+  assert re.search(r'INFRA_SETUP_PID=\$!\s+set \+m', deploy_sh)
+
+  # 1. Pre-flight abort (PIDs/logs still empty) must not fail under set -u.
+  preflight = subprocess.run(
+      ["bash", "-c", f"set -euo pipefail\n{trap_block}\nexit 1\n"],
+      capture_output=True,
+      text=True,
+      check=False,
+  )
+  assert preflight.returncode == 1
+  assert "unbound variable" not in preflight.stderr
+
+  # 2. Mid-deploy abort dumps non-empty background log to stderr and unlinks it.
+  bg_log = tmp_path / "infra.log"
+  child_pid_file = tmp_path / 'infra-child.pid'
+  bg_log.write_text("Firestore DB 1 created\nSeed failed HTTP 404\n")
+  abort_run = subprocess.run(
+      [
+          "bash",
+          "-c",
+          f'set -euo pipefail\n{trap_block}\nINFRA_SETUP_LOG="{bg_log}"\n'
+          'set -m\n'
+          f'( sleep 30 & echo "$!" > "{child_pid_file}"; wait )'
+          f' </dev/null >>"{bg_log}" 2>&1 &\n'
+          'INFRA_SETUP_PID=$!\nset +m\n'
+          f'while [ ! -s "{child_pid_file}" ]; do sleep 0.01; done\n'
+          'exit 1\n',
+      ],
+      capture_output=True,
+      text=True,
+      check=False,
+      timeout=10,
+  )
+  assert abort_run.returncode == 1
+  assert "--- background log:" in abort_run.stderr
+  assert "Seed failed HTTP 404" in abort_run.stderr
+  assert not bg_log.exists(), "Expected cleanup trap to unlink background log"
+  assert not _pid_exists(int(child_pid_file.read_text())), (
+      'Expected cleanup trap to stop the infra job child'
+  )
+
+
+def test_deploy_failure_gates_app_and_config_seeding():
+  """A failed worker rollout must stop before app promotion or config writes."""
+  text = _deploy_sh()
+  worker = text.index('gcloud run deploy worker --image')
+  worker_binding = text.index('add_run_invoker_binding worker', worker)
+  app = text.index('gcloud run deploy app --image', worker_binding)
+  app_binding = text.index('add_run_invoker_binding app', app)
+  seed = text.index('CONFIG_SEED_STATUS=$(curl', app_binding)
+  assert worker < worker_binding < app < app_binding < seed
+  assert '${IMAGE}:latest' not in text
+  assert 'BUILD_MACHINE_ARGS' not in text
+  assert '--machine-type=' not in text
+  assert '.package-lock.stamp' not in text
+  assert '( cd ui && npm ci )' in text
+  assert ') </dev/null >"$UI_BUILD_LOG" 2>&1 &' in text
+  assert ') </dev/null >"$INFRA_SETUP_LOG" 2>&1 &' in text
+  assert not re.search(r'run_with_heartbeat[^\n]*\|\s*tee', text)
+  assert '--image "$DEPLOY_IMAGE"' in text
+  assert 'DEPLOY_IMAGE="${IMAGE%:*}@${IMAGE_DIGEST}"' in text
+
+
+@pytest.mark.parametrize('failure', ['deploy', 'binding'])
+def test_worker_failure_stops_before_app_rollout_or_seeding(tmp_path, failure):
+  """Execute the real worker phase with a failing fake gcloud/binding."""
+  text = _deploy_sh()
+  start = text.index('WORKER_URL="https://worker-')
+  end = text.index('# --- Cloud Run: app', start)
+  worker_phase = text[start:end]
+  calls = tmp_path / 'calls'
+  script = f'''
+set -euo pipefail
+APP_ONLY=0 PROJECT=p REGION=us-central1 PROJECT_NUMBER=123
+IMAGE=pkg:latest DEPLOY_IMAGE=pkg@sha256:{'a' * 64} RUNTIME_SA=runtime@example.com
+CALLS="{calls}" FAILURE="{failure}"
+phase() {{ :; }}
+gcloud() {{
+  printf '%s\n' "$*" >> "$CALLS"
+  if [ "$FAILURE" = deploy ] && [ "$1 $2 $3" = 'run deploy worker' ]; then return 7; fi
+}}
+add_run_invoker_binding() {{
+  printf 'binding %s\n' "$*" >> "$CALLS"
+  if [ "$FAILURE" = binding ]; then return 8; fi
+}}
+{worker_phase}
+printf 'app phase reached\n' >> "$CALLS"
+'''
+  proc = subprocess.run(['/bin/bash', '-c', script], capture_output=True, text=True)
+  assert proc.returncode != 0, proc.stdout + proc.stderr
+  recorded = calls.read_text()
+  assert 'run deploy worker' in recorded
+  assert f'--image pkg@sha256:{"a" * 64}' in recorded
+  assert '--image pkg:latest' not in recorded
+  assert 'app phase reached' not in recorded
+  assert 'run deploy app' not in recorded
+  assert 'seed' not in recorded
+
+
+def test_build_receipt_resolves_only_its_own_successful_image(tmp_path):
+  helper = _REPO / 'deploy' / 'resolve_build_image.py'
+  build = '957c2a44-0012-4e80-8666-3fecfc199401'
+  project, region = 'pr206-test', 'us-central1'
+  log = tmp_path / 'build.log'
+  log.write_text(
+      f'Created [https://cloudbuild.googleapis.com/v1/projects/{project}/locations/{region}/builds/{build}].\n'
+  )
+  found = subprocess.run(
+      [sys.executable, str(helper), 'build-id', project, region, str(log)],
+      capture_output=True, text=True,
+  )
+  assert found.returncode == 0, found.stderr
+  assert found.stdout.strip() == build
+  wrong_project = subprocess.run(
+      [sys.executable, str(helper), 'build-id', 'other-project', region, str(log)],
+      capture_output=True, text=True,
+  )
+  assert wrong_project.returncode != 0
+  log.write_text(log.read_text() + log.read_text())
+  duplicate_build = subprocess.run(
+      [sys.executable, str(helper), 'build-id', project, region, str(log)],
+      capture_output=True, text=True,
+  )
+  assert duplicate_build.returncode != 0
+  assert 'expected exactly one' in duplicate_build.stderr
+
+  image = f'{region}-docker.pkg.dev/{project}/repo/app:latest'
+  receipt = {
+      'status': 'SUCCESS',
+      'results': {'images': [
+          {'name': image, 'digest': 'sha256:' + 'a' * 64},
+          {'name': image.removesuffix(':latest'), 'digest': 'sha256:' + 'a' * 64},
+      ]},
+  }
+  resolved = subprocess.run(
+      [sys.executable, str(helper), 'digest', image], input=json.dumps(receipt),
+      capture_output=True, text=True,
+  )
+  assert resolved.returncode == 0, resolved.stderr
+  assert resolved.stdout.strip() == 'sha256:' + 'a' * 64
+  for bad in (
+      {**receipt, 'status': 'FAILURE'},
+      {**receipt, 'results': {'images': [{'name': image, 'digest': 'sha256:short'}]}},
+      {
+          **receipt,
+          'results': {'images': [
+              {'name': image, 'digest': 'sha256:' + 'a' * 64 + 'suffix'}
+          ]},
+      },
+      {**receipt, 'results': {'images': []}},
+      {**receipt, 'results': {'images': [{'name': image, 'digest': None}]}},
+      {**receipt, 'results': {'images': [{'name': image, 'digest': 123}]}},
+      None,
+      {**receipt, 'results': None},
+      {**receipt, 'results': {'images': [None]}},
+  ):
+    rejected = subprocess.run(
+        [sys.executable, str(helper), 'digest', image], input=json.dumps(bad),
+        capture_output=True, text=True,
+    )
+    assert rejected.returncode != 0
+    assert 'Traceback' not in rejected.stderr
+    assert 'expected string or bytes-like' not in rejected.stderr
+
+
+def test_skip_ui_build_fails_fast_and_build_id_failure_clears_log(tmp_path):
+  """--skip-ui-build validates ui/dist early; build-id error clears log."""
+  text = _deploy_sh()
+  start = text.index('if [ "$SKIP_UI_BUILD" != "1" ]; then')
+  end = text.index('# --- Enable services', start)
+  assert end > start
+  skip_block = text[start:end]
+
+  missing = subprocess.run(
+      ['bash', '-c', f'set -euo pipefail\nSKIP_UI_BUILD=1\n{skip_block}\n'],
+      cwd=tmp_path,
+      capture_output=True,
+      text=True,
+      check=False,
+  )
+  assert missing.returncode == 1
+  assert '--skip-ui-build given but ui/dist does not exist' in missing.stderr
+
+  dist = tmp_path / 'ui' / 'dist'
+  dist.mkdir(parents=True)
+  (dist / 'main.js').write_text('const c = {controlPlaneMode:"none"};')
+  dev_dist = subprocess.run(
+      ['bash', '-c', f'set -euo pipefail\nSKIP_UI_BUILD=1\n{skip_block}\n'],
+      cwd=tmp_path,
+      capture_output=True,
+      text=True,
+      check=False,
+  )
+  assert dev_dist.returncode == 1
+  assert 'existing ui/dist was built for local dev' in dev_dist.stderr
+
+  (dist / 'main.js').write_text('const c = {controlPlaneMode:"iap"};')
+  valid_dist = subprocess.run(
+      ['bash', '-c', f'set -euo pipefail\nSKIP_UI_BUILD=1\n{skip_block}\n'],
+      cwd=tmp_path,
+      capture_output=True,
+      text=True,
+      check=False,
+  )
+  assert valid_dist.returncode == 0
+
+  trap_match = re.search(
+      r'(UI_BUILD_PID="";.*?trap cleanup EXIT)', text, re.DOTALL
+  )
+  loop_start = text.index('BUILD_ATTEMPT=0')
+  loop_end = text.index('\n# Cloud Build reports the digest', loop_start)
+  loop_block = text[loop_start:loop_end]
+  script = (
+      'set -euo pipefail\n'
+      f'{trap_match.group(1)}\n'
+      'PROJECT=p REGION=us-central1 BUILD_SUBS=""\n'
+      'run_with_heartbeat() {\n'
+      '  local log="$2"\n'
+      '  echo "Cloud Build finished without receipt URL" > "$log"\n'
+      '  cat "$log"\n'
+      '  return 0\n'
+      '}\n'
+      f'{loop_block}\n'
+  )
+  res = subprocess.run(
+      ['bash', '-c', script],
+      cwd=_REPO,
+      capture_output=True,
+      text=True,
+      check=False,
+  )
+  assert res.returncode == 1
+  assert 'ID could not be verified' in res.stderr
+  assert '--- background log:' not in res.stderr
+
+
+def test_cloud_build_cold_image_still_exports_cache_for_the_next_warm_build(tmp_path):
+  """Run both actual Cloud Build shell branches against a recording docker."""
+  cloudbuild = (_REPO / 'cloudbuild.yaml').read_text(encoding='utf-8')
+  assert cloudbuild.count('      - |\n') == 1
+  script = textwrap.dedent(
+      cloudbuild.split('      - |\n', 1)[1].split('\nsubstitutions:', 1)[0]
+  )
+  fake_bin = tmp_path / 'bin'
+  fake_bin.mkdir()
+  docker = fake_bin / 'docker'
+  docker.write_text(
+      '#!/bin/sh\nprintf "%s\\n" "$*" >> "$DOCKER_LOG"\n'
+      'if [ "${DOCKER_PULL_FAIL:-0}" = 1 ] && [ "$1" = pull ]; then exit 1; fi\n'
+  )
+  docker.chmod(0o755)
+  image = 'us-central1-docker.pkg.dev/trial/repo/app:latest'
+  for use_cache, pull_fails in (('0', False), ('1', False), ('1', True)):
+    calls = tmp_path / f'docker-{use_cache}-{pull_fails}.log'
+    env = {
+        **os.environ,
+        'PATH': f'{fake_bin}:{os.environ["PATH"]}',
+        'DOCKER_LOG': str(calls),
+        '_IMAGE': image,
+        '_USE_CACHE': use_cache,
+        'DOCKER_PULL_FAIL': '1' if pull_fails else '0',
+    }
+    run = subprocess.run(['/bin/bash', '-c', script], env=env,
+                         capture_output=True, text=True)
+    assert run.returncode == 0, run.stderr
+    commands = calls.read_text().splitlines()
+    build = next((line for line in commands if line.startswith('build ')), '')
+    assert build, commands
+    assert f'-t {image}' in build
+    assert '--build-arg BUILDKIT_INLINE_CACHE=1' in build
+    if use_cache == '1':
+      assert commands[0] == f'pull {image}'
+      assert f'--cache-from {image}' in build
+      assert '--no-cache' not in build
+    else:
+      assert len(commands) == 1, commands
+      assert '--no-cache' in build
+      assert '--cache-from' not in build
+
+
+@pytest.mark.parametrize('no_build_cache, expected', [('0', None), ('1', '_USE_CACHE=0')])
+def test_deploy_cache_flag_reaches_cloud_build_substitution(no_build_cache, expected):
+  """Exercise the deploy flag's actual substitution assembly under Bash 3.2."""
+  deploy = _deploy_sh()
+  assert '--substitutions="$BUILD_SUBS"' in deploy
+  match = re.search(r'(BUILD_SUBS="_IMAGE=\$\{IMAGE\}".*?\nfi)', deploy, re.DOTALL)
+  assert match, 'Expected Cloud Build substitution assembly in deploy.sh'
+  script = (
+      'set -euo pipefail\nIMAGE=example:latest\n'
+      f'NO_BUILD_CACHE={no_build_cache}\n{match.group(1)}\n'
+      'printf "%s" "$BUILD_SUBS"\n'
+  )
+  run = subprocess.run(['/bin/bash', '-c', script], capture_output=True, text=True)
+  assert run.returncode == 0, run.stderr
+  substitutions = run.stdout.splitlines()[-1]
+  assert substitutions.startswith('_IMAGE=example:latest')
+  assert ('_USE_CACHE=0' in substitutions) == (expected is not None)
+
+
+def test_build_log_link_is_printed_before_the_build_finishes(tmp_path):
+  """The Cloud Build link should appear before the buffered output replay."""
+  heartbeat = re.search(
+      r'(run_with_heartbeat\(\) \{.*?\n\})', _deploy_sh(), re.DOTALL
+  )
+  assert heartbeat
+  link = (
+      'Logs are available at [ '
+      'https://console.cloud.google.com/cloud-build/builds/example ].'
+  )
+  fake_build = tmp_path / 'fake-build.sh'
+  fake_build.write_text(
+      f'#!/bin/bash\nprintf "%s\\n" "{link}"\nsleep 1.2\n'
+      'printf "%s\\n" "build complete"\n'
+  )
+  fake_build.chmod(0o755)
+  log = tmp_path / 'build.log'
+  script = (
+      'set -euo pipefail\n'
+      + heartbeat.group(1) + '\n'
+      + 'HEARTBEAT_SECS=30\n'
+      + f'run_with_heartbeat "Cloud Build" "{log}" "{fake_build}"\n'
+  )
+  run = subprocess.run(
+      ['/bin/bash', '-c', script], capture_output=True, text=True, timeout=10
+  )
+  assert run.returncode == 0, run.stderr
+  assert run.stdout.count(link) == 2, run.stdout
+  assert run.stdout.index(link) < run.stdout.index('build complete')
+
+
+def _pid_exists(pid: int) -> bool:
+  try:
+    os.kill(pid, 0)
+    return True
+  except ProcessLookupError:
+    return False
+
+
+def test_heartbeat_cleanup_kills_command_and_descendants_on_signal(tmp_path):
+  """The real Bash 3.2 EXIT trap must stop a heartbeat-wrapped process tree."""
+  text = _deploy_sh()
+  heartbeat = re.search(r'(run_with_heartbeat\(\) \{.*?\n\})', text, re.DOTALL)
+  cleanup = re.search(
+      r'(UI_BUILD_PID="";.*?trap \'exit 143\' TERM)', text, re.DOTALL
+  )
+  assert heartbeat and cleanup
+  heartbeat_block = heartbeat.group(1).replace(
+      'HEARTBEAT_PID=$!',
+      'HEARTBEAT_PID=$!; echo "$HEARTBEAT_PID" > "$PID_DIR/heartbeat"',
+  )
+  assert heartbeat_block != heartbeat.group(1)
+  log = tmp_path / 'build.log'
+  pid_dir = tmp_path / 'pids'
+  pid_dir.mkdir()
+  fake_command = tmp_path / 'fake-build.sh'
+  fake_command.write_text(
+      '#!/bin/bash\n'
+      'echo "$$" > "$PID_DIR/command"\n'
+      'echo "fake build started"\n'
+      'bash -c \'echo "$$" > "$PID_DIR/child"; '
+      'sleep 30 & echo "$!" > "$PID_DIR/grandchild"; wait\' &\n'
+      'wait\n',
+      encoding='utf-8',
+  )
+  fake_command.chmod(0o755)
+  script = (
+      'set -euo pipefail\n'
+      + 'fmt_hms() { printf "%ss" "$1"; }\n'
+      + heartbeat_block + '\n'
+      + cleanup.group(1) + '\n'
+      + 'HEARTBEAT_SECS=30\n'
+      + f'BUILD_SUBMIT_LOG="{log}"\n'
+      + f'run_with_heartbeat "Build" "{log}" "{fake_command}"\n'
+  )
+  env = {**os.environ, 'PID_DIR': str(pid_dir)}
+  proc = subprocess.Popen(
+      ['/bin/bash', '-c', script],
+      stdout=subprocess.PIPE,
+      stderr=subprocess.PIPE,
+      text=True,
+      env=env,
+  )
+  pids = []
+  try:
+    deadline = time.monotonic() + 8
+    while time.monotonic() < deadline:
+      if all((pid_dir / name).exists() for name in ('command', 'child', 'grandchild', 'heartbeat')):
+        break
+      time.sleep(0.05)
+    else:
+      pytest.fail('Fake build did not start its full process tree')
+    pids = [int((pid_dir / name).read_text()) for name in ('command', 'child', 'grandchild', 'heartbeat')]
+    os.kill(proc.pid, signal.SIGTERM)
+    stdout, stderr = proc.communicate(timeout=10)
+    assert proc.returncode == 143, (stdout, stderr)
+    assert '--- background log:' in stderr
+    deadline = time.monotonic() + 4
+    while time.monotonic() < deadline and any(_pid_exists(pid) for pid in pids):
+      time.sleep(0.05)
+    alive = [pid for pid in pids if _pid_exists(pid)]
+    assert not alive, f'Deploy cleanup left descendants alive: {alive}'
+    assert not log.exists(), 'Expected cleanup to remove the build log'
+  finally:
+    if proc.poll() is None:
+      proc.kill()
+      proc.communicate(timeout=5)
+    for pid in pids:
+      try:
+        os.kill(pid, signal.SIGKILL)
+      except ProcessLookupError:
+        pass