@@ -7,50 +7,74 @@
 # 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 without the simulation
-# database URL predates the safety check, so its schedule is paused.
+# plain telemetry settings; nothing else. A revision deployed before the
+# safety check existed has no job code, so its schedule is paused instead.
 #
-# Environment: PROJECT_ID, REGION, SERVICE, SAFETY_CHECK_REVISION,
-# SAFETY_CHECK_SCHEDULER_SERVICE_ACCOUNT, RUNNER_TEMP.
+# 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 SAFETY_CHECK_REVISION \
-  SAFETY_CHECK_SCHEDULER_SERVICE_ACCOUNT RUNNER_TEMP; do
+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"
-[[ "${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"
 
 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}"
 
-if ! jq -e '.spec.containers[0].env // [] | any(.name == "SIMULATION_DATABASE_URL")' >/dev/null <<<"${revision_json}"; then
-  if gcloud scheduler jobs describe "${job}" "${scheduler_location[@]}" >/dev/null 2>&1; then
-    gcloud scheduler jobs pause "${job}" "${scheduler_location[@]}" --quiet
-  fi
-  printf '%s has no simulation database, so the safety check is paused.\n' "${SAFETY_CHECK_REVISION}"
+# 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
 
 
@@ -1016,12 +1016,17 @@ jobs:
         run: .github/scripts/set-cloud-run-simulation-entry-revision.sh
 
       - name: Point the safety check back at the previous revision
+        id: safety_check_restore
         if: ${{ failure() && steps.safety_check.outcome == 'failure' && env.PREVIOUS_REVISION != '' }}
         env:
           SAFETY_CHECK_REVISION: ${{ env.PREVIOUS_REVISION }}
           SAFETY_CHECK_SCHEDULER_SERVICE_ACCOUNT: ${{ vars.SIMULATION_SAFETY_CHECK_SCHEDULER_SERVICE_ACCOUNT }}
         run: .github/scripts/deploy-cloud-run-simulation-safety-check.sh
 
+      - name: Pause the safety check after a failed restore
+        if: ${{ failure() && steps.safety_check_restore.outcome == 'failure' }}
+        run: .github/scripts/deploy-cloud-run-simulation-safety-check.sh pause
+
   finalize_routing:
     name: Finalize the Modal routing publication
     # Finish the durable journal after every gate, including failed uploads
 
@@ -217,12 +217,18 @@ jobs:
           fi
 
       - name: Point the safety check at the restored revision
+        id: safety_check
         if: ${{ inputs.entrypoint_revision != '' }}
         env:
           SAFETY_CHECK_REVISION: ${{ inputs.entrypoint_revision }}
           SAFETY_CHECK_SCHEDULER_SERVICE_ACCOUNT: ${{ vars.SIMULATION_SAFETY_CHECK_SCHEDULER_SERVICE_ACCOUNT }}
         run: .github/scripts/deploy-cloud-run-simulation-safety-check.sh
 
+      # The job must not keep running the image traffic just moved away from.
+      - name: Pause the safety check after a failed update
+        if: ${{ failure() && steps.safety_check.outcome == 'failure' }}
+        run: .github/scripts/deploy-cloud-run-simulation-safety-check.sh pause
+
       - name: Verify the stable service serves the restored release
         if: ${{ !inputs.resolve_pending_publication }}
         env:
 
@@ -112,10 +112,14 @@ records anything:
 
 - the promotion job, after the promoted revision passes every check; if this
   step fails, the previous revision is restored and the job is pointed back
-  at it;
-- the rollback workflow, after it restores entrypoint traffic. A restored
-  revision without the simulation database predates the safety check, so the
-  schedule is paused.
+  at it, and if that fails too, the schedule is paused;
+- the rollback workflow, after it restores entrypoint traffic; if this step
+  fails, the schedule is paused, so the job never keeps running the image
+  traffic just moved away from.
+
+A revision deployed before the safety check existed (without
+`SIMULATION_CALL_REPORT_MODAL_WORKSPACE_ID`) has no job code, so pointing the
+job at it pauses the schedule instead.
 
 `beta` does not promote its candidate yet, so staging has no safety check
 until it does (plan item 1.39).
 
@@ -56,6 +56,10 @@ def _revision(*, service=SERVICE, environment=None):
                     if environment is not None
                     else [
                         {"name": "APP_ENVIRONMENT", "value": "production"},
+                        {
+                            "name": "SIMULATION_CALL_REPORT_MODAL_WORKSPACE_ID",
+                            "value": "ac-workspace0001",
+                        },
                         {"name": "OLD_GATEWAY_URL", "value": "https://legacy.example"},
                         {"name": "OBSERVABILITY_SERVICE_NAMESPACE", "value": "sim"},
                         {"name": "CLOUD_RUN_REGION", "value": "us-central1"},
@@ -286,3 +290,68 @@ def test_a_rollback_points_the_job_at_the_restored_revision() -> None:
     assert steps[safety]["env"]["SAFETY_CHECK_REVISION"] == (
         "${{ inputs.entrypoint_revision }}"
     )
+
+
+def test_a_revision_from_before_the_safety_check_pauses_the_schedule(tmp_path) -> None:
+    """A 1.04 or 1.05 revision mounts the database and token but has no job code."""
+    environment = [
+        entry
+        for entry in _revision()["spec"]["containers"][0]["env"]
+        if entry["name"] != "SIMULATION_CALL_REPORT_MODAL_WORKSPACE_ID"
+    ]
+
+    result, fakes, _ = _deploy(
+        tmp_path, _revision(environment=environment), schedule_exists=True
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert [call[1:4] for call in fakes.calls("gcloud")] == [
+        ["run", "revisions", "describe"],
+        ["scheduler", "jobs", "describe"],
+        ["scheduler", "jobs", "pause"],
+    ]
+
+
+@pytest.mark.parametrize("schedule_exists", [True, False])
+def test_pausing_stops_the_schedule_without_a_revision(
+    tmp_path, schedule_exists
+) -> None:
+    fakes, _ = _fakes(tmp_path, _revision(), schedule_exists=schedule_exists)
+    settings = {
+        "PROJECT_ID": "policyengine",
+        "REGION": "us-central1",
+        "SERVICE": SERVICE,
+    }
+
+    result = run_script(SCRIPT, "pause", env=fakes.env(**settings))
+
+    assert result.returncode == 0, result.stderr
+    expected = [["scheduler", "jobs", "describe"]]
+    if schedule_exists:
+        expected.append(["scheduler", "jobs", "pause"])
+    assert [call[1:4] for call in fakes.calls("gcloud")] == expected
+
+
+def test_a_failed_update_in_a_rollback_pauses_the_safety_check() -> None:
+    """The job must not keep running the image traffic just moved away from."""
+    steps = _steps(_workflow_jobs(ROLLBACK_WORKFLOW)["rollback"])
+    safety = _step_named(steps, "Point the safety check at the restored revision")
+    pause = _step_named(steps, "Pause the safety check after a failed update")
+
+    assert safety < pause
+    assert steps[safety]["id"] == "safety_check"
+    assert steps[pause]["if"] == (
+        "${{ failure() && steps.safety_check.outcome == 'failure' }}"
+    )
+    assert steps[pause]["run"] == f".github/scripts/{SCRIPT} pause"
+
+
+def test_a_failed_restore_in_a_promotion_pauses_the_safety_check() -> None:
+    steps = _steps(_workflow_jobs(REUSABLE_DEPLOY_WORKFLOW)["promote_entrypoint"])
+    restore = _step_named(steps, "Point the safety check back at the previous revision")
+    pause = _step_named(steps, "Pause the safety check after a failed restore")
+
+    assert restore < pause
+    assert steps[restore]["id"] == "safety_check_restore"
+    assert "steps.safety_check_restore.outcome == 'failure'" in steps[pause]["if"]
+    assert steps[pause]["run"] == f".github/scripts/{SCRIPT} pause"