feat(relay): watch a reserve cell and flip it back on the first step 5 stop rule (#28698)

* feat(relay): watch a reserve cell and flip it back on the first step 5 stop rule

A manual, inert-until-dispatched workflow samples one reserve cell every minute for up to
300 minutes against the step 5 §9 stop rules, logs one JSON line per sample, and on the
first trip (or N unreadable samples in a row) runs the flag tool's own admitMode=db write
under the flag workflow's concurrency group. dry-run logs the flag tool's plan instead.

Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f

* fix(relay): keep the reserve watch's flip back out of the flag group, and baseline 503s before the flip

- The flip back holds no concurrency group: GitHub keeps one pending run per group and a
  newer one cancels it. The flag tool's CAS orders it instead; a generation conflict is
  re-read and retried (5 tries, backoff) while the switch file still says reserve.
- The 503 background comes from the 10 minutes before a required flipped-at, or an explicit
  value; neither is refused.
- A failed or cancelled write-mode watch flips back too. An unreachable cell fails with the
  exact break-glass inputs.
- Missing 503 data (empty series, unpublished minute, unparseable time) is unverifiable.
- Seat-behind-row key adds the director instance and cell; six segments share one command.

Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f

* fix(relay): read the reserve watch's 503 publishing front from every response code

A quiet fleet has no 503 series, so the watch read it as unverifiable and flipped a healthy
cell back within minutes. The 503 read now groups request_count by response code in one
query; the newest minute any code has a point for is how far the series is published, so a
503 minute at or before it with no point is zero and minutes past it are left out.

Also: the flip-back job's steps run on always(), flipped-at ages against the watch's start
in later segments, a pending pre-flip background no longer adds to the unverifiable streak,
the flip back retries runtime-status before calling for break glass, and the concurrency
rationale lives once in the docs.

Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f

* docs(relay): state what a cancel during the flip-back lease wait can leave, and the 503 publish-skew assumption

Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f

* fix(relay): fail a segment closed on a corrupt watch state file, and keep no checkout token on disk

Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f

* fix(relay): alarm on a failed flip back, refuse a watch under a rollout lease, wait for reserve to apply, and control the duplicate read

- A flip back whose write or plan never finished (no ID token, a failed login, a silent cell)
  raises an ::error:: annotation and a FLIP BACK FAILED summary, as the clock-skew monitor does.
- The first segment reads the Cloud SQL rollout lease through the lease action's own reader and
  refuses to start while any run holds it, or when it cannot be read.
- A cell not yet reporting reserve is waited on for up to 10 minutes, judging nothing, before the
  watch ends refused without a write.
- The duplicate-seat read takes the per-minute ledger summary in the same query and window; with
  no fresh summary in it the sample is unverifiable, not zero duplicates.

Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f

* fix(relay): page on a refused reserve watch or a failed flip back

A refused watch can leave a cell in reserve and unwatched, the same as a flip back that did not
land. Both now raise an escaped ::error:: annotation through one `alarm` command, and in write
mode write orca_relay_reserve_watch_alarm with gcloud logging write. A new log-based metric and
alert policy page on any such line, per cell, through relay_alert_notification_channels, and the
deploy identity gets roles/logging.logWriter in production to write it.

Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f

* fix(relay): name the log-writer risk, check the alarm's project, and say a failed login cannot page

Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f

* docs(relay): note a force-cancel can skip the reserve-watch flip-back write

.github/workflows/cloud-operate-relay-production-reserve-watch.yml

@@ -0,0 +1,327 @@
+name: Operate Relay Production Reserve Watch
+
+# Watches one cell already switched to reserve mode and flips it back to db on the first stop
+# rule it trips (step 5 design §9). Manual only. dry-run evaluates and logs, writes nothing.
+# An operator watches the run. A refused watch or a flip back that did not land raises an ::error::
+# annotation, as the clock-skew monitor does, and in write mode logs the line the reserve-watch
+# alert pages on. Refuses to start while a rollout holds the Cloud SQL rollout lease.
+on:
+  workflow_dispatch:
+    inputs:
+      mode:
+        description: dry-run logs what a trip would write; write flips the cell back on a trip
+        required: true
+        default: dry-run
+        type: choice
+        options: [dry-run, write]
+      cell-id:
+        description: Exact cell id already in reserve mode, e.g. production-gce-c26
+        required: true
+        type: string
+      duration-minutes:
+        description: How long to watch, 5 to 300
+        required: true
+        default: '120'
+        type: string
+      flipped-at:
+        description: >-
+          When the reserve write applied, ISO with a zone, e.g. 2026-10-14T12:00:00Z. The 503
+          background is the median of the 10 minutes before it
+        required: false
+        default: ''
+        type: string
+      background-503-per-minute:
+        description: >-
+          Director 503s per minute before the flip; overrides the one measured before flipped-at.
+          One of the two is required
+        required: false
+        default: ''
+        type: string
+      unverifiable-limit:
+        description: Consecutive samples with any unreadable signal that count as a trip
+        required: true
+        default: '3'
+        type: string
+      confirmation:
+        description: Required for write, exactly "WRITE_RELAY_CELL_FLAGS <cell-id>"
+        required: false
+        type: string
+
+permissions:
+  contents: read
+  id-token: write
+
+# One watch per cell; why not the flag workflow's group: cloud/docs/relay-workflows.md.
+concurrency:
+  group: relay-production-reserve-watch-${{ inputs.cell-id }}
+  cancel-in-progress: false
+
+defaults:
+  run:
+    # Explicit bash runs with pipefail, so the script's exit code survives the tee.
+    shell: bash
+    working-directory: cloud
+
+env:
+  MODE: ${{ inputs.mode }}
+  CELL_ID: ${{ inputs.cell-id }}
+  CONFIRMATION: ${{ inputs.confirmation }}
+  GCP_PROJECT_ID: onorca-cloud
+
+jobs:
+  watch:
+    if: ${{ vars.ORCA_CLOUD_OPERATIONS_ENABLED == 'true' }}
+    runs-on: ubuntu-latest
+    # Six 55-minute segments cover the longest watch; under the 360-minute job limit.
+    timeout-minutes: 350
+    environment: production
+    outputs:
+      state: ${{ steps.outcome.outputs.state }}
+      verdict: ${{ steps.outcome.outputs.verdict }}
+    env:
+      DURATION_MINUTES: ${{ inputs.duration-minutes }}
+      FLIPPED_AT: ${{ inputs.flipped-at }}
+      BACKGROUND_503_PER_MINUTE: ${{ inputs.background-503-per-minute }}
+      UNVERIFIABLE_LIMIT: ${{ inputs.unverifiable-limit }}
+      # One segment, after the auth step before it: fixed text, the inputs arrive as env vars.
+      WATCH_SEGMENT: |
+        set -euo pipefail
+        [[ "${CELL_ID}" =~ ^production-gce-c[0-9]{1,3}$ ]]
+        node -e "if (!/^[^.]+[.][^.]+[.][^.]+$/.test(process.env.ORCA_RELAY_ADMIN_ID_TOKEN ?? '')) process.exit(1)"
+        node dev/scripts/watch-relay-reserve-cell.mjs watch \
+          --mode "${MODE}" --cell-id "${CELL_ID}" \
+          --cell-origin "https://${CELL_ID#production-gce-}.relay.onorca.dev" \
+          --project-id "${GCP_PROJECT_ID}" --director-origin "https://relay.onorca.dev" \
+          --director-service orca-cloud-relay --confirmation "${CONFIRMATION}" \
+          --duration-minutes "${DURATION_MINUTES}" --segment-minutes 55 \
+          --flipped-at "${FLIPPED_AT}" --background-503-per-minute "${BACKGROUND_503_PER_MINUTE}" \
+          --unverifiable-limit "${UNVERIFIABLE_LIMIT}" \
+          --state-file "${RUNNER_TEMP}/relay-reserve-watch.json" --github-output "${GITHUB_OUTPUT}" \
+          | tee -a "${RUNNER_TEMP}/relay-reserve-watch.jsonl"
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+
+      - uses: actions/setup-node@v4
+        with:
+          node-version: 24
+
+      - id: google-auth-1
+        uses: google-github-actions/auth@v2
+        with:
+          workload_identity_provider: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_SERVICE_ACCOUNT }}
+          token_format: id_token
+          id_token_audience: https://relay.onorca.dev/v1/admin/drain
+          id_token_include_email: true
+
+      - uses: google-github-actions/setup-gcloud@v2
+        with:
+          skip_install: true
+
+      - id: watch-1
+        name: Watch segment 1 of 6
+        env:
+          ORCA_RELAY_ADMIN_ID_TOKEN: ${{ steps.google-auth-1.outputs.id_token }}
+        run: bash -c "${WATCH_SEGMENT}"
+
+      - id: google-auth-2
+        if: ${{ steps.watch-1.outputs.state == 'watching' }}
+        uses: google-github-actions/auth@v2
+        with:
+          workload_identity_provider: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_SERVICE_ACCOUNT }}
+          token_format: id_token
+          id_token_audience: https://relay.onorca.dev/v1/admin/drain
+          id_token_include_email: true
+
+      - id: watch-2
+        name: Watch segment 2 of 6
+        if: ${{ steps.watch-1.outputs.state == 'watching' }}
+        env:
+          ORCA_RELAY_ADMIN_ID_TOKEN: ${{ steps.google-auth-2.outputs.id_token }}
+        run: bash -c "${WATCH_SEGMENT}"
+
+      - id: google-auth-3
+        if: ${{ steps.watch-2.outputs.state == 'watching' }}
+        uses: google-github-actions/auth@v2
+        with:
+          workload_identity_provider: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_SERVICE_ACCOUNT }}
+          token_format: id_token
+          id_token_audience: https://relay.onorca.dev/v1/admin/drain
+          id_token_include_email: true
+
+      - id: watch-3
+        name: Watch segment 3 of 6
+        if: ${{ steps.watch-2.outputs.state == 'watching' }}
+        env:
+          ORCA_RELAY_ADMIN_ID_TOKEN: ${{ steps.google-auth-3.outputs.id_token }}
+        run: bash -c "${WATCH_SEGMENT}"
+
+      - id: google-auth-4
+        if: ${{ steps.watch-3.outputs.state == 'watching' }}
+        uses: google-github-actions/auth@v2
+        with:
+          workload_identity_provider: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_SERVICE_ACCOUNT }}
+          token_format: id_token
+          id_token_audience: https://relay.onorca.dev/v1/admin/drain
+          id_token_include_email: true
+
+      - id: watch-4
+        name: Watch segment 4 of 6
+        if: ${{ steps.watch-3.outputs.state == 'watching' }}
+        env:
+          ORCA_RELAY_ADMIN_ID_TOKEN: ${{ steps.google-auth-4.outputs.id_token }}
+        run: bash -c "${WATCH_SEGMENT}"
+
+      - id: google-auth-5
+        if: ${{ steps.watch-4.outputs.state == 'watching' }}
+        uses: google-github-actions/auth@v2
+        with:
+          workload_identity_provider: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_SERVICE_ACCOUNT }}
+          token_format: id_token
+          id_token_audience: https://relay.onorca.dev/v1/admin/drain
+          id_token_include_email: true
+
+      - id: watch-5
+        name: Watch segment 5 of 6
+        if: ${{ steps.watch-4.outputs.state == 'watching' }}
+        env:
+          ORCA_RELAY_ADMIN_ID_TOKEN: ${{ steps.google-auth-5.outputs.id_token }}
+        run: bash -c "${WATCH_SEGMENT}"
+
+      - id: google-auth-6
+        if: ${{ steps.watch-5.outputs.state == 'watching' }}
+        uses: google-github-actions/auth@v2
+        with:
+          workload_identity_provider: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_SERVICE_ACCOUNT }}
+          token_format: id_token
+          id_token_audience: https://relay.onorca.dev/v1/admin/drain
+          id_token_include_email: true
+
+      - id: watch-6
+        name: Watch segment 6 of 6
+        if: ${{ steps.watch-5.outputs.state == 'watching' }}
+        env:
+          ORCA_RELAY_ADMIN_ID_TOKEN: ${{ steps.google-auth-6.outputs.id_token }}
+        run: bash -c "${WATCH_SEGMENT}"
+
+      - id: outcome
+        name: Record the watch outcome
+        if: ${{ always() }}
+        run: |
+          node dev/scripts/watch-relay-reserve-cell.mjs outcome \
+            --state-file "${RUNNER_TEMP}/relay-reserve-watch.json" --github-output "${GITHUB_OUTPUT}"
+
+      # The documented flow writes reserve before dispatching, so a refusal can leave it unwatched.
+      - name: Alarm on a refused watch
+        if: ${{ always() && steps.outcome.outputs.state == 'refused' }}
+        env:
+          WATCH_VERDICT: ${{ steps.outcome.outputs.verdict }}
+        run: |
+          node dev/scripts/watch-relay-reserve-cell.mjs alarm --kind refused \
+            --mode "${MODE}" --cell-id "${CELL_ID}" --project-id "${GCP_PROJECT_ID}" --verdict "${WATCH_VERDICT}"
+
+      - name: Audit summary
+        if: ${{ always() }}
+        env:
+          WATCH_STATE: ${{ steps.outcome.outputs.state }}
+          WATCH_VERDICT: ${{ steps.outcome.outputs.verdict }}
+        run: |
+          {
+            echo "### Relay reserve watch: ${MODE} ${CELL_ID}"
+            echo "Triggered by: ${GITHUB_TRIGGERING_ACTOR}, run ${GITHUB_RUN_ID} attempt ${GITHUB_RUN_ATTEMPT}"
+            echo "Outcome: ${WATCH_STATE}: ${WATCH_VERDICT}"
+            echo '```json'
+            tail -n 20 "${RUNNER_TEMP}/relay-reserve-watch.jsonl" 2>/dev/null || echo '(no output)'
+            echo '```'
+          } >> "${GITHUB_STEP_SUMMARY}"
+
+  flip-back:
+    needs: watch
+    # Missing data is never healthy: a watch that crashed, timed out or was cancelled (by an
+    # operator too) flips back like a trip. In dry-run this only plans the write.
+    if: ${{ always() && vars.ORCA_CLOUD_OPERATIONS_ENABLED == 'true' && needs.watch.result != 'skipped' && !contains(fromJSON('["completed","stopped","refused"]'), needs.watch.outputs.state) }}
+    runs-on: ubuntu-latest
+    # A flip back waits for the cell to lease every control: 8-50 minutes, scaled to its controls.
+    timeout-minutes: 60
+    environment: production
+    # Only the run's per-cell group above, which a running run never yields (cancel-in-progress off).
+    # Every step runs on always() too, so an ordinary cancel cannot skip the switch-file write; a force-cancel
+    # can, and leaves the cell in reserve for the operator to flip back by hand. A cancel during the
+    # lease wait can still kill the process before Postgres records db: the file says db, Postgres reserve.
+    env:
+      VERDICT: ${{ needs.watch.outputs.verdict || format('the watch ended {0} without a verdict', needs.watch.result) }}
+    steps:
+      - uses: actions/checkout@v4
+        if: ${{ always() }}
+        with:
+          persist-credentials: false
+
+      - uses: actions/setup-node@v4
+        if: ${{ always() }}
+        with:
+          node-version: 24
+
+      - id: google-auth
+        if: ${{ always() }}
+        uses: google-github-actions/auth@v2
+        with:
+          workload_identity_provider: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ vars.PRODUCTION_GCP_RELAY_DEPLOY_SERVICE_ACCOUNT }}
+          token_format: id_token
+          id_token_audience: https://relay.onorca.dev/v1/admin/drain
+          id_token_include_email: true
+
+      - uses: google-github-actions/setup-gcloud@v2
+        if: ${{ always() }}
+        with:
+          skip_install: true
+
+      - name: Flip the cell back to db, or preview it
+        if: ${{ always() }}
+        env:
+          ORCA_RELAY_ADMIN_ID_TOKEN: ${{ steps.google-auth.outputs.id_token }}
+        run: |
+          [[ "${CELL_ID}" =~ ^production-gce-c[0-9]{1,3}$ ]]
+          ORCA_RELAY_CONTROL_ACCESS_TOKEN="$(gcloud auth print-access-token)"
+          echo "::add-mask::${ORCA_RELAY_CONTROL_ACCESS_TOKEN}"
+          export ORCA_RELAY_CONTROL_ACCESS_TOKEN
+          node dev/scripts/watch-relay-reserve-cell.mjs flip-back \
+            --mode "${MODE}" --cell-id "${CELL_ID}" \
+            --cell-origin "https://${CELL_ID#production-gce-}.relay.onorca.dev" \
+            --project-id "${GCP_PROJECT_ID}" --director-origin "https://relay.onorca.dev" \
+            --confirmation "${CONFIRMATION}" --verdict "${VERDICT}" \
+            | tee "${RUNNER_TEMP}/relay-reserve-flip-back.jsonl"
+
+      # The flip-back line is logged only once the write (or its dry-run plan) finished; without
+      # it a missing token, a failed login or a silent cell left the cell in reserve, unwatched.
+      - name: Alarm when the flip back did not land
+        if: ${{ always() }}
+        run: |
+          if grep -qs '"event":"orca_relay_reserve_watch_flip_back"' "${RUNNER_TEMP}/relay-reserve-flip-back.jsonl"; then
+            exit 0
+          fi
+          {
+            echo "## FLIP BACK FAILED: ${CELL_ID} may still be in reserve mode"
+            echo "Flip it to db by hand with Operate Relay Production Cell Flags."
+          } >> "${GITHUB_STEP_SUMMARY}"
+          node dev/scripts/watch-relay-reserve-cell.mjs alarm --kind flip-back-failed \
+            --mode "${MODE}" --cell-id "${CELL_ID}" --project-id "${GCP_PROJECT_ID}" --verdict "${VERDICT}"
+
+      - name: Audit summary
+        if: ${{ always() }}
+        run: |
+          {
+            echo "### Relay reserve watch flip back: ${MODE} ${CELL_ID}"
+            echo "Triggered by: ${GITHUB_TRIGGERING_ACTOR}, run ${GITHUB_RUN_ID} attempt ${GITHUB_RUN_ATTEMPT}"
+            echo "Verdict: ${VERDICT}"
+            echo '```json'
+            cat "${RUNNER_TEMP}/relay-reserve-flip-back.jsonl" 2>/dev/null || echo '(no output)'
+            echo '```'
+          } >> "${GITHUB_STEP_SUMMARY}"

cloud/dev/fixtures/terraform-root-partition/families.json

@@ -146,6 +146,7 @@
     "google_logging_metric.relay_event_loop_stall",
     "google_logging_metric.relay_long_cell_inventory_hold",
     "google_logging_metric.relay_reserve_placement_off_with_reserve_cells",
+    "google_logging_metric.relay_reserve_watch_alarm",
     "google_logging_metric.relay_snapshot",
     "google_monitoring_alert_policy.relay_assignment_5xx",
     "google_monitoring_alert_policy.relay_assignment_edge_429",
@@ -176,6 +177,7 @@
     "google_monitoring_alert_policy.relay_process_fatal",
     "google_monitoring_alert_policy.relay_region_hint_skew",
     "google_monitoring_alert_policy.relay_reserve_placement_off_with_reserve_cells",
+    "google_monitoring_alert_policy.relay_reserve_watch_alarm",
     "google_monitoring_dashboard.relay_incident",
     "google_project_iam_custom_role.github_production_relay_capacity_mutation",
     "google_project_iam_custom_role.github_relay_asia_topology_mutation",
@@ -304,6 +306,10 @@
       "production": "relay",
       "staging": "apps"
     },
+    "google_project_iam_member.github_log_writer": {
+      "production": "relay",
+      "staging": "apps"
+    },
     "google_project_iam_member.github_logging_viewer": {
       "production": "relay",
       "staging": "apps"

cloud/dev/scripts/operate-relay-cell-flags.mjs

@@ -244,7 +244,7 @@ function objectUrl(request, suffix = '') {
 }
 
 // One transient 5xx or network failure is retried once, as every relay admin step does.
-async function readRuntime(fetchImpl, request, idToken, wait) {
+export async function readRuntime(fetchImpl, request, idToken, wait) {
   const response = await fetchAdminOnceMore(
     fetchImpl,
     `${request.cellOrigin}/v1/admin/runtime-status`,
@@ -262,7 +262,7 @@ async function readRuntime(fetchImpl, request, idToken, wait) {
   return await response.json()
 }
 
-async function readCurrentObject(fetchImpl, request, accessToken) {
+export async function readCurrentObject(fetchImpl, request, accessToken) {
   const headers = { authorization: `Bearer ${accessToken}` }
   const metadata = await fetchImpl(objectUrl(request), { headers })
   if (metadata.status === 404) {
@@ -278,6 +278,12 @@ async function readCurrentObject(fetchImpl, request, accessToken) {
   return { generation: String(generation), object: await media.json().catch(() => null) }
 }
 
+function generationConflict(current, expected) {
+  return Object.assign(new Error(`current generation is ${current}, not expected ${expected}`), {
+    generationConflict: true
+  })
+}
+
 async function writeObject(fetchImpl, request, object, accessToken, audit) {
   const boundary = `relay-cell-flags-${Date.now()}`
   const metadata = {
@@ -315,7 +321,8 @@ async function writeObject(fetchImpl, request, object, accessToken, audit) {
   if (response.status === 412) {
     await response.body?.cancel().catch(() => undefined)
     throw Object.assign(new Error('the object changed since expected-generation; re-read and retry'), {
-      notWritten: true
+      notWritten: true,
+      generationConflict: true
     })
   }
   if (!response.ok) {
@@ -409,7 +416,7 @@ async function breakGlassToDatabase(request, dependencies, { fetchImpl, log, sle
   }
   const current = await readCurrentObject(fetchImpl, request, accessToken)
   if (current.generation !== request.expectedGeneration) {
-    throw new Error(`current generation is ${current.generation}, not expected ${request.expectedGeneration}`)
+    throw generationConflict(current.generation, request.expectedGeneration)
   }
   const object = desiredObject(request, current.object)
   const generation = await writeObject(fetchImpl, request, object, accessToken, audit)
@@ -459,9 +466,7 @@ export async function operateCellFlags(request, dependencies) {
   // Before the generation check, so a dry run always shows the generation to pass next.
   log(JSON.stringify(plan))
   if (current.generation !== request.expectedGeneration) {
-    throw new Error(
-      `current generation is ${current.generation}, not expected ${request.expectedGeneration}`
-    )
+    throw generationConflict(current.generation, request.expectedGeneration)
   }
   // The cell applies every readable write; a generation it did not apply was voided there. A
   // write of an object this image accepts whole (checked above) repairs it.

cloud/dev/scripts/relay-production-identity-boundaries.test.mjs

@@ -33,6 +33,7 @@ test('no workflow names the retired generic production deploy identity', async (
     'operate-relay-asia-admission.yml',
     'operate-relay-production-cell-flags.yml',
     'operate-relay-production-rehome-job.yml',
+    'operate-relay-production-reserve-watch.yml',
     'publish-relay-production.yml'
   ].map((name) => relayWorkflowFile(name)).sort())
 })

cloud/dev/scripts/relay-public-workflow-contract.test.mjs

@@ -20,7 +20,7 @@ const UNGATED = relayWorkflowFile('verify.yml')
 const relayWorkflows = () => workflowFiles().filter((file) => file !== UNGATED)
 
 test('the copy carries every relay workflow', () => {
-  assert.equal(relayWorkflows().length, 27)
+  assert.equal(relayWorkflows().length, 28)
 })
 
 // Why: workflow_run chains match by display name, not filename. Renaming a file is safe; renaming

cloud/dev/scripts/relay-reserve-dead-man-alert.test.mjs

@@ -12,6 +12,8 @@ const deadMan = read('../../apps/relay/src/cell-reserve-dead-man.ts')
 const registry = read('../../apps/relay/src/host-session-registry.ts')
 const server = read('../../apps/relay/src/relay-server.ts')
 const poller = read('../../apps/relay/src/shadow-seat-directory.ts')
+const watch = read('./watch-relay-reserve-cell.mjs')
+const githubActions = read('../../infra/terraform/relay-github-actions.tf')
 
 const block = (kind, name) => {
   const body = new RegExp(`resource "${kind}" "${name}" \\{([\\s\\S]*?)\\n\\}`).exec(terraform)?.[1]
@@ -84,3 +86,24 @@ test('the withheld dead-cell verdict alert counts every director withheld line,
   assert.match(read('../../infra/terraform/variables.tf'), /variable "relay_dead_cell_verdict_alert" \{[\s\S]*?default     = false/)
   assert.match(read('../../infra/terraform/environments/staging.tfvars'), /relay_dead_cell_verdict_alert\s+= true/)
 })
+
+test('the reserve watch alarm pages on any line the watch runner writes, per cell', () => {
+  // The runner writes a global-resource entry under its own log name.
+  assert.match(watch, /export const ALARM_EVENT = 'orca_relay_reserve_watch_alarm'/)
+  assert.match(watch, /export const ALARM_LOG_NAME = 'orca-relay-reserve-watch'/)
+  assert.match(watch, /JSON\.stringify\(\{ event: ALARM_EVENT, cellId, kind, reason \}\)/)
+  const metric = block('google_logging_metric', 'relay_reserve_watch_alarm')
+  assert.ok(
+    metric.includes(
+      'resource.type=\\"global\\" AND logName=\\"projects/${var.project_id}/logs/orca-relay-reserve-watch\\" AND jsonPayload.event=\\"orca_relay_reserve_watch_alarm\\"'
+    )
+  )
+  assert.ok(metric.includes('cell_id = "EXTRACT(jsonPayload.cellId)"'))
+  notifyingOnAnyLine('relay_reserve_watch_alarm', 'orca_relay_reserve_watch_alarm', 'global', 'cell_id')
+  // The deploy identity the workflow runs as can write that line, and only in production.
+  const grant = /resource "google_project_iam_member" "github_log_writer" \{([\s\S]*?)\n\}/.exec(githubActions)?.[1]
+  assert.ok(grant, 'github_log_writer grant not found')
+  assert.match(grant, /count = local\.relay_create_production_ops_identity \? 1 : 0/)
+  assert.match(grant, /role    = "roles\/logging\.logWriter"/)
+  assert.match(grant, /member  = google_service_account\.github_deploy\[0\]\.member/)
+})

cloud/dev/scripts/relay-reserve-watch-rules.mjs

@@ -0,0 +1,258 @@
+// Stop rules for a cell watched in reserve mode (step 5 design §9, "Auto-abort per flip"). Pure:
+// it takes one already-read sample and the watch state, and returns the next state and whether
+// to flip the cell back. The reader lives in watch-relay-reserve-cell.mjs.
+import { backgroundOf, SHADOW_GATE_THRESHOLDS } from './relay-same-cap-shadow-gate-verdict.mjs'
+
+const MINUTE_MS = 60_000
+
+export const RESERVE_WATCH_RULES = {
+  // The roll ladder's warn line: max(1.5x, background + 20), so a zero background is not tripped
+  // by one 503. Three complete minutes in a row, not two. The read spans more minutes than that so
+  // a late publishing front still leaves three published minutes to judge.
+  director503: {
+    multiple: SHADOW_GATE_THRESHOLDS.nonDrain503Budget.warnMultiple,
+    marginPerMinute: SHADOW_GATE_THRESHOLDS.nonDrain503Budget.warnMarginPerMinute,
+    consecutiveMinutes: 3,
+    readMinutes: 10,
+    // Cloud Run publishes a minute's request count up to 3 minutes late.
+    publishDelayMs: 3 * MINUTE_MS
+  },
+  // The cell closes a demoted seat after 60 s (DEMOTION_CLOSE_MS); 30 s more covers a director's
+  // map still holding the closed seat.
+  duplicateSeat: { graceMs: 60_000 + 30_000, lookbackMs: 15 * MINUTE_MS },
+  // The floor keeps an almost empty cell's ordinary churn from reading as a 5% drop.
+  hostDrop: { share: 0.05, windowMs: 5 * MINUTE_MS, minHosts: 10 },
+  ledgerDrift: { share: 0.005, sustainMs: 2 * MINUTE_MS },
+  // A placing director logs its ledger summary every 60 s; none in this long is no evidence.
+  ledgerSummaryMaxAgeMs: 150_000,
+  // A watch dispatched before the reserve write applies waits this long for the cell to report it.
+  reserveApplyWaitMs: 10 * MINUTE_MS
+}
+
+// Precedence: the first tripped rule names the verdict, so exact signals come before proxies. The
+// sample line's `rules` keys follow this order too.
+export const RULE_NAMES = ['deadMan', 'overrun', 'director503', 'duplicateSeat', 'hostDrop', 'ledgerDrift']
+
+export function initialWatchState({ cellId, startedAt, durationMinutes, background503PerMinute, flippedAt }) {
+  return {
+    v: 1,
+    cellId,
+    startedAt,
+    deadlineAt: startedAt + durationMinutes * MINUTE_MS,
+    state: 'watching',
+    verdict: null,
+    samples: 0,
+    unverifiableStreak: 0,
+    // The background is measured in the minutes before the flip, unless the operator supplied it.
+    flippedAt: flippedAt ?? null,
+    background:
+      background503PerMinute === null ? null : { medianPerMinute: background503PerMinute, source: 'input' },
+    hosts: [],
+    reserveSeenAt: null,
+    ledgerAboveSince: null,
+    seatBehindRowFirstSeen: {}
+  }
+}
+
+// The background from the per-minute 503 counts before the flip. `publishedThrough` is the newest
+// minute any response code has a point for: a 503 minute at or before it with no point had none,
+// and the background waits (null) until it covers the last minute.
+export function measuredBackground(perMinute, minutes, publishedThrough) {
+  if (minutes.length === 0 || publishedThrough === null || publishedThrough < minutes.at(-1)) return null
+  const background = backgroundOf({ minutes, series: minutes.map((minute) => perMinute[minute] ?? 0), unverified: false })
+  return { medianPerMinute: background.medianPerMinute, minutes: background.minutes, source: 'measured' }
+}
+
+const pass = (detail = {}) => ({ status: 'pass', ...detail })
+const trip = (reason, detail = {}) => ({ status: 'trip', reason, ...detail })
+const unverifiable = (reason) => ({ status: 'unverifiable', reason })
+// Not readable yet by design, so it does not count toward the unverifiable streak.
+const pending = (reason) => ({ status: 'pending', reason })
+
+function judgeDeadMan(cell) {
+  if (!cell.ok) return unverifiable(cell.error)
+  if (cell.admitModeEffective !== 'db' && cell.admitModeEffective !== 'reserve') {
+    return unverifiable('runtime-status reports no admitModeEffective')
+  }
+  if (cell.admitModeEffective === 'db') {
+    return trip('the cell tripped its dead-man (reserve configured, db effective); Postgres still records reserve')
+  }
+  return pass()
+}
+
+// The cell ledger's total (seats, bookings, in-flight and phone units) against its hard cap. The
+// ledger refuses past the cap, so this is an invariant check: it trips only if that refusal broke.
+function judgeOverrun(cell) {
+  if (!cell.ok) return unverifiable(cell.error)
+  if (!Number.isInteger(cell.hardCap) || !Number.isInteger(cell.enforcedUnits)) {
+    return unverifiable('runtime-status reports no hard cap or ledger units')
+  }
+  if (cell.enforcedUnits > cell.hardCap) {
+    return trip(`ledger units ${cell.enforcedUnits} over hard cap ${cell.hardCap}`, {
+      enforcedUnits: cell.enforcedUnits,
+      hardCap: cell.hardCap
+    })
+  }
+  return pass({ enforcedUnits: cell.enforcedUnits, hardCap: cell.hardCap })
+}
+
+// Fleet-wide director 503s: no existing signal attributes a 503 to the cell a host was bound for,
+// so a problem local to one cell can stay under the fleet's line. This proxy can miss.
+function judgeDirector503(read, state, at) {
+  const { multiple, marginPerMinute, consecutiveMinutes, publishDelayMs } = RESERVE_WATCH_RULES.director503
+  const { background } = state
+  if (!background) {
+    const publishesAt = state.flippedAt === null ? null : Math.floor(state.flippedAt / MINUTE_MS) * MINUTE_MS + publishDelayMs
+    if (publishesAt !== null && at < publishesAt) return pending('the 503 background before the flip is not published yet')
+    return unverifiable('no 503 background yet')
+  }
+  if (!read.ok) return unverifiable(read.error)
+  const perMinute = background.medianPerMinute
+  const above = Math.max(perMinute * multiple, perMinute + marginPerMinute)
+  // Minute keys sort as time. Only minutes the publishing front has reached are judged: a 503
+  // minute there with no point had none, one past it is not published yet and is left out.
+  if (read.publishedThrough === null) return unverifiable('the director request series is empty')
+  const recent = read.minutes.filter((minute) => minute <= read.publishedThrough).slice(-consecutiveMinutes)
+  if (recent.length < consecutiveMinutes) {
+    return unverifiable(`director requests are published only through ${read.publishedThrough}`)
+  }
+  const counts = recent.map((minute) => read.perMinute[minute] ?? 0)
+  const detail = { backgroundPerMinute: perMinute, above, recent: Object.fromEntries(recent.map((m, i) => [m, counts[i]])) }
+  if (counts.every((count) => count > above)) {
+    return trip(`director 503s above ${above}/min for ${consecutiveMinutes} min (${counts.join(', ')})`, detail)
+  }
+  return pass(detail)
+}
+
+// A seat the ledger reconcile found behind its row and found again after the demotion should
+// have closed it. The line names no host, so (director instance, cell, epoch, rowEpoch) stands in
+// for one: a proxy that can only over-count, when two hosts share that key.
+function judgeDuplicateSeat(read, firstSeen, at) {
+  if (!read.ok) return { verdict: unverifiable(read.error), firstSeen }
+  const { graceMs, lookbackMs } = RESERVE_WATCH_RULES.duplicateSeat
+  const next = Object.fromEntries(
+    Object.entries(firstSeen).filter(([, seenAt]) => seenAt >= at - 2 * lookbackMs)
+  )
+  for (const report of [...read.reports].sort((left, right) => left.at - right.at)) {
+    next[report.key] = Math.min(next[report.key] ?? report.at, report.at)
+  }
+  for (const report of read.reports) {
+    const ageMs = report.at - next[report.key]
+    if (ageMs > graceMs) {
+      return {
+        verdict: trip(`a seat behind its ledger row (${report.key}) is still seated ${Math.round(ageMs / 1000)} s after first report`),
+        firstSeen: next
+      }
+    }
+  }
+  return { verdict: pass({ reportsInLookback: read.reports.length }), firstSeen: next }
+}
+
+function judgeHostDrop(hosts) {
+  const { share, minHosts, windowMs } = RESERVE_WATCH_RULES.hostDrop
+  const current = hosts.at(-1)
+  if (!current) return unverifiable('no host count')
+  if (hosts.some((entry) => entry.draining)) return pass({ controls: current.controls, draining: true })
+  const peak = Math.max(...hosts.map((entry) => entry.controls))
+  const drop = peak - current.controls
+  const detail = { controls: current.controls, peakInWindow: peak }
+  if (peak > 0 && drop >= minHosts && drop / peak > share) {
+    return trip(`connected hosts fell ${peak} -> ${current.controls} within ${windowMs / MINUTE_MS} min outside a drain`, detail)
+  }
+  return pass(detail)
+}
+
+// Rows the placing directors have not written (queued) or gave up on (dropped), per seat. The
+// summaries are fleet-wide: exact while this is the only reserve cell, an over-count otherwise.
+function judgeLedgerDrift(cell, ledger, aboveSince, at) {
+  if (!cell.ok) return { verdict: unverifiable(cell.error), aboveSince }
+  if (!ledger.ok) return { verdict: unverifiable(ledger.error), aboveSince }
+  const { share, sustainMs } = RESERVE_WATCH_RULES.ledgerDrift
+  const ratio = ledger.unwrittenRows / Math.max(cell.controls, 1)
+  const detail = { unwrittenRows: ledger.unwrittenRows, directors: ledger.directors, ratio }
+  if (ratio <= share) return { verdict: pass(detail), aboveSince: null }
+  const since = aboveSince ?? at
+  if (at - since >= sustainMs) {
+    return {
+      verdict: trip(`ledger drift ${(ratio * 100).toFixed(2)}% for ${Math.round((at - since) / 1000)} s`, detail),
+      aboveSince: since
+    }
+  }
+  return { verdict: pass({ ...detail, aboveForMs: at - since }), aboveSince: since }
+}
+
+/**
+ * One sample. `sample.cell` is the cell's runtime-status, `director503` the complete minutes of
+ * director 503s with the request series' publishing front (`publishedThrough`), `ledger` the placing directors' newest ledger summaries, `seatBehindRow` this
+ * cell's reconcile reports in the lookback. Each is `{ ok: false, error }` when it could not be read.
+ */
+export function evaluateWatchSample(state, sample, { unverifiableLimit }) {
+  const at = sample.at
+  const next = { ...state, samples: state.samples + 1 }
+  const { cell } = sample
+  if (cell.ok && cell.admitModeConfigured !== 'reserve') {
+    const seen = state.reserveSeenAt !== null
+    if (!seen && at < state.startedAt + RESERVE_WATCH_RULES.reserveApplyWaitMs) {
+      // Not judged, and the streak holds: the reserve write may not have applied yet.
+      const rules = { reserve: pending(`switch file says ${cell.admitModeConfigured}; waiting for reserve to apply`) }
+      return { state: next, rules, line: sampleLine(next, sample, rules) }
+    }
+    // Someone else flipped it back (or it never became reserve): nothing left to watch.
+    next.state = seen ? 'stopped' : 'refused'
+    next.verdict = seen
+      ? `${state.cellId} is not in reserve mode (switch file says ${cell.admitModeConfigured})`
+      : `${state.cellId} did not report reserve mode within ${RESERVE_WATCH_RULES.reserveApplyWaitMs / MINUTE_MS} min (switch file says ${cell.admitModeConfigured})`
+    return { state: next, rules: {}, line: sampleLine(next, sample, {}) }
+  }
+  if (cell.ok) {
+    next.reserveSeenAt = state.reserveSeenAt ?? at
+    next.hosts = [...state.hosts, { at, controls: cell.controls, draining: cell.draining }].filter(
+      (entry) => entry.at > at - RESERVE_WATCH_RULES.hostDrop.windowMs
+    )
+  }
+  const duplicate = judgeDuplicateSeat(sample.seatBehindRow, state.seatBehindRowFirstSeen, at)
+  next.seatBehindRowFirstSeen = duplicate.firstSeen
+  const ledger = judgeLedgerDrift(cell, sample.ledger, state.ledgerAboveSince, at)
+  next.ledgerAboveSince = ledger.aboveSince
+  const rules = {
+    deadMan: judgeDeadMan(cell),
+    overrun: judgeOverrun(cell),
+    director503: judgeDirector503(sample.director503, state, at),
+    duplicateSeat: duplicate.verdict,
+    hostDrop: cell.ok ? judgeHostDrop(next.hosts) : unverifiable(cell.error),
+    ledgerDrift: ledger.verdict
+  }
+  const tripped = RULE_NAMES.find((name) => rules[name].status === 'trip')
+  if (tripped) {
+    next.state = 'tripped'
+    next.verdict = `flip back ${state.cellId}: ${tripped}: ${rules[tripped].reason}`
+    return { state: next, rules, line: sampleLine(next, sample, rules) }
+  }
+  const missing = RULE_NAMES.filter((name) => rules[name].status === 'unverifiable')
+  const waiting = RULE_NAMES.some((name) => rules[name].status === 'pending')
+  // A pending rule holds the streak where it was: neither evidence of health nor a gap.
+  next.unverifiableStreak = missing.length > 0 ? state.unverifiableStreak + 1 : waiting ? state.unverifiableStreak : 0
+  if (next.unverifiableStreak >= unverifiableLimit) {
+    next.state = 'tripped'
+    next.verdict =
+      `flip back ${state.cellId}: unverifiable: ${next.unverifiableStreak} samples in a row could not ` +
+      `read ${missing.join(', ')}`
+  } else if (at >= state.deadlineAt) {
+    next.state = 'completed'
+    next.verdict = `${state.cellId} stayed healthy in reserve mode for the whole watch`
+  }
+  return { state: next, rules, line: sampleLine(next, sample, rules) }
+}
+
+function sampleLine(state, sample, rules) {
+  return JSON.stringify({
+    event: 'orca_relay_reserve_watch_sample',
+    cellId: state.cellId,
+    at: new Date(sample.at).toISOString(),
+    sample: state.samples,
+    state: state.state,
+    unverifiableStreak: state.unverifiableStreak,
+    rules,
+    ...(state.verdict ? { verdict: state.verdict } : {})
+  })
+}

cloud/dev/scripts/relay-same-cap-shadow-gate.mjs

@@ -122,7 +122,7 @@ async function retryingRead(reader, read) {
   return { failed: true, error: String(lastError?.message ?? lastError) }
 }
 
-async function readLogEntries(reader, { filter, projection, limit = ENTRY_LIMIT }) {
+export async function readLogEntries(reader, { filter, projection, limit = ENTRY_LIMIT }) {
   const args = [
     'logging', 'read', filter,
     '--project', reader.projectId,
@@ -154,21 +154,22 @@ async function readTimestampsOverWindow(reader, { filter, window }) {
 
 // Cloud Run's own request counter, aligned per minute server-side: a brownout's tens of thousands
 // of 503s are counted, where a log read of them stops at its entry limit (10-01 c29).
-async function readDirector503PerMinute(reader, { config, span }) {
+async function readDirectorRequestCountSeries(reader, { config, span, filters = [], groupByFields }) {
   const url = new URL(`https://monitoring.googleapis.com/v3/projects/${config.projectId}/timeSeries`)
   url.searchParams.set('filter', [
     'metric.type="run.googleapis.com/request_count"',
     'resource.type="cloud_run_revision"',
     `resource.label."service_name"="${config.directorService}"`,
-    'metric.label."response_code"="503"'
+    ...filters
   ].join(' AND '))
   url.searchParams.set('interval.startTime', span.startedAt.toISOString())
   url.searchParams.set('interval.endTime', span.endedAt.toISOString())
   url.searchParams.set('aggregation.alignmentPeriod', '60s')
   url.searchParams.set('aggregation.perSeriesAligner', 'ALIGN_DELTA')
   url.searchParams.set('aggregation.crossSeriesReducer', 'REDUCE_SUM')
+  if (groupByFields) url.searchParams.set('aggregation.groupByFields', groupByFields)
   url.searchParams.set('pageSize', '1000')
-  const read = await retryingRead(reader, async (timeoutMs) => {
+  return await retryingRead(reader, async (timeoutMs) => {
     const { stdout } = await reader.runGcloud(['auth', 'print-access-token'], { timeoutMs })
     const response = await reader.fetch(url, {
       headers: { authorization: `Bearer ${stdout.trim()}` },
@@ -179,17 +180,44 @@ async function readDirector503PerMinute(reader, { config, span }) {
     if (body.nextPageToken) throw new Error('Cloud Monitoring pagination is incomplete')
     return { timeSeries: body.timeSeries ?? [] }
   })
-  const perMinute = {}
-  for (const series of read.timeSeries ?? []) {
-    for (const point of series.points ?? []) {
-      // A delta point covers the minute that ends at its end time; a minute with no point had none.
-      const minute = minuteKey(Date.parse(point.interval.endTime) - 60_000)
-      perMinute[minute] = (perMinute[minute] ?? 0) + Number(point.value.int64Value ?? 0)
-    }
+}
+
+function addPointsPerMinute(perMinute, series) {
+  for (const point of series.points ?? []) {
+    // A delta point covers the minute that ends at its end time; a minute with no point had none.
+    const minute = minuteKey(Date.parse(point.interval.endTime) - 60_000)
+    perMinute[minute] = (perMinute[minute] ?? 0) + Number(point.value.int64Value ?? 0)
   }
+  return perMinute
+}
+
+export async function readDirector503PerMinute(reader, { config, span }) {
+  const read = await readDirectorRequestCountSeries(reader, {
+    config,
+    span,
+    filters: ['metric.label."response_code"="503"']
+  })
+  const perMinute = {}
+  for (const series of read.timeSeries ?? []) addPointsPerMinute(perMinute, series)
   return { perMinute, failed: read.failed }
 }
 
+// Every response code in one read, so a minute missing from the 503 series can be told apart from
+// one not published yet: the other codes' series carry the same publishing front.
+export async function readDirectorRequestsByCodePerMinute(reader, { config, span }) {
+  const read = await readDirectorRequestCountSeries(reader, {
+    config,
+    span,
+    groupByFields: 'metric.label."response_code"'
+  })
+  const byCode = {}
+  for (const series of read.timeSeries ?? []) {
+    const code = series.metric?.labels?.response_code ?? 'unknown'
+    byCode[code] = addPointsPerMinute(byCode[code] ?? {}, series)
+  }
+  return { byCode, failed: read.failed }
+}
+
 // Cells log through the COS container agent, so the text lives in jsonPayload.message; a
 // textPayload filter matches nothing here and returns zero without saying so.
 const CELL_LOG_SCOPE = 'resource.type="gce_instance" AND logName:"cos_containers"'

cloud/dev/scripts/watch-relay-reserve-cell.mjs

@@ -0,0 +1,581 @@
+#!/usr/bin/env node
+// Watches one cell already in reserve mode and, on the first stop rule it trips, flips it back to
+// db through the flag tool's own write (CAS, read-back, re-registration wait). Step 5 design §9.
+//
+// watch:     one segment of the watch (the workflow re-authenticates between segments, since an
+//            ID token lasts an hour). State carries over in --state-file. Writes nothing.
+// outcome:   the job's state and verdict, from the state file.
+// flip-back: the flag tool's admitMode=db write (or its dry run), then exits non-zero with the
+//            watch's verdict. Why it holds no flag-workflow group: docs/relay-workflows.md.
+// alarm:     a refused watch or a flip back that did not land: an error annotation, and in write
+//            mode a log line the reserve-watch alert pages on.
+import { execFile } from 'node:child_process'
+import { appendFile, readFile, writeFile } from 'node:fs/promises'
+import { parseArgs, promisify } from 'node:util'
+import { pathToFileURL } from 'node:url'
+import {
+  BREAK_GLASS_CONFIRMATION,
+  operateCellFlags,
+  parseCellFlagsRequest,
+  readCurrentObject,
+  readRuntime,
+  WRITE_CONFIRMATION
+} from './operate-relay-cell-flags.mjs'
+import { readDirectorRequestsByCodePerMinute, readLogEntries } from './relay-same-cap-shadow-gate.mjs'
+import { minutesOf } from './relay-same-cap-shadow-gate-verdict.mjs'
+import { PRODUCTION_LEASE } from './cloud-sql-rollout-lock-census.mjs'
+import { RELAY_REPOSITORY_ROOT } from './relay-repository.mjs'
+import {
+  evaluateWatchSample,
+  initialWatchState,
+  measuredBackground,
+  RESERVE_WATCH_RULES
+} from './relay-reserve-watch-rules.mjs'
+
+const execFileAsync = promisify(execFile)
+const MINUTE_MS = 60_000
+export const SAMPLE_INTERVAL_MS = MINUTE_MS
+export const MAX_WATCH_MINUTES = 300
+// Below the hour an ID token lasts, with room for the segment's last sample.
+export const MAX_SEGMENT_MINUTES = 55
+const METRIC_DELAY_MS = RESERVE_WATCH_RULES.director503.publishDelayMs
+const BACKGROUND_MINUTES = 10
+// A flipped-at older than this is a typo or a stale dispatch, not this flip.
+const MAX_FLIPPED_AT_AGE_MS = 24 * 60 * MINUTE_MS
+export const FLIP_BACK_ATTEMPTS = 5
+const FLIP_BACK_BACKOFF_MS = 2_000
+// One slow runtime-status reply is not a dead cell: retried this often before break glass.
+export const FLIP_BACK_RUNTIME_ATTEMPTS = 4
+const SAMPLE_READ_BUDGET_MS = 40_000
+const LOG_ENTRY_LIMIT = 1_000
+
+function integerIn(name, raw, min, max) {
+  const value = Number(raw)
+  if (!/^[0-9]+$/.test(raw ?? '') || value < min || value > max) {
+    throw new Error(`--${name} must be an integer from ${min} to ${max}`)
+  }
+  return value
+}
+
+// The flip back's request. Its generation is a placeholder: the flip back re-reads the current one.
+export function dbFlipRequest(values) {
+  return parseCellFlagsRequest({ ...values, set: 'admitMode=db', 'expected-generation': '0' })
+}
+
+// `now` is the watch's start: a later segment of a long watch must not age flipped-at out.
+function parseFlippedAt(raw, now) {
+  if (raw === '') return null
+  const at = Date.parse(raw)
+  if (!/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}(:\d{2}(\.\d{1,3})?)?(Z|[+-]\d{2}:\d{2})$/.test(raw) || !Number.isFinite(at)) {
+    throw new Error('--flipped-at must be an ISO time with a zone, e.g. 2026-10-14T12:00:00Z')
+  }
+  if (at > now) throw new Error('--flipped-at is in the future')
+  if (at < now - MAX_FLIPPED_AT_AGE_MS) throw new Error('--flipped-at is more than 24 hours ago')
+  return at
+}
+
+export function parseWatchArguments(values, now = Date.now()) {
+  // The flag tool's own checks, so the watch is of exactly the cell the flip back would write;
+  // a write-mode watch refuses a missing confirmation now, not hours later.
+  const request = dbFlipRequest(values)
+  if (!/^[a-z][a-z0-9-]{0,62}$/.test(values['director-service'] ?? '')) {
+    throw new Error('--director-service is not a Cloud Run service name')
+  }
+  const background = values['background-503-per-minute'] ?? ''
+  if (background !== '' && !/^(0|[1-9][0-9]{0,6})(\.[0-9]+)?$/.test(background)) {
+    throw new Error('--background-503-per-minute must be empty (measure it) or a non-negative number')
+  }
+  // The background must predate the flip: a watch starts after it, so its own past is not one.
+  const flippedAt = parseFlippedAt(values['flipped-at'] ?? '', now)
+  if (background === '' && flippedAt === null) {
+    throw new Error('--flipped-at (the background is the 10 minutes before it) or --background-503-per-minute is required')
+  }
+  if (!values['state-file']) throw new Error('--state-file is required')
+  return {
+    request,
+    directorService: values['director-service'],
+    durationMinutes: integerIn('duration-minutes', values['duration-minutes'], 5, MAX_WATCH_MINUTES),
+    segmentMinutes: integerIn('segment-minutes', values['segment-minutes'], 1, MAX_SEGMENT_MINUTES),
+    unverifiableLimit: integerIn('unverifiable-limit', values['unverifiable-limit'], 1, 30),
+    background503PerMinute: background === '' ? null : Number(background),
+    flippedAt,
+    stateFile: values['state-file'],
+    githubOutput: values['github-output'] || null
+  }
+}
+
+const failedRead = (error) => ({ ok: false, error: error instanceof Error ? error.message : String(error) })
+
+export async function readCellSample(fetchImpl, request, idToken, wait) {
+  try {
+    const runtime = await readRuntime(fetchImpl, request, idToken, wait)
+    if (runtime.role !== 'cell' || runtime.cellId !== request.cellId) {
+      return failedRead(`origin answers as ${runtime.role}/${runtime.cellId}`)
+    }
+    if (!runtime.flagsApplied) return failedRead('runtime-status reports no flagsApplied')
+    const controls = runtime.runtime?.controls
+    if (!Number.isInteger(controls)) return failedRead('runtime-status reports no control count')
+    return {
+      ok: true,
+      admitModeConfigured: runtime.flagsApplied.flags?.admitMode ?? 'db',
+      admitModeEffective: runtime.admitModeEffective,
+      draining: runtime.draining === true,
+      controls,
+      enforcedUnits: runtime.runtime?.enforcedConnectionUnits,
+      hardCap: runtime.connectionCapacity?.hardCap
+    }
+  } catch (error) {
+    return failedRead(error)
+  }
+}
+
+const minuteFloor = (at) => Math.floor(at / MINUTE_MS) * MINUTE_MS
+const failed503Read = (error) => ({ ok: false, error, perMinute: {}, minutes: [], publishedThrough: null })
+
+// The `minutes` whole minutes that end at the minute boundary `endedAt`: the 503 counts, and the
+// newest minute any response code has a point for. The director always serves something, so that
+// front is how far the 503 series is published; no front at all is no evidence. A code publishing
+// late only delays a trip: a later read sees its points.
+export async function readDirector503Minutes(reader, { projectId, directorService, endedAt, minutes }) {
+  const span = { startedAt: new Date(endedAt - minutes * MINUTE_MS), endedAt: new Date(endedAt) }
+  let read
+  try {
+    read = await readDirectorRequestsByCodePerMinute(reader, { config: { projectId, directorService }, span })
+  } catch (error) {
+    // A point with an unparseable end time has no minute.
+    return failed503Read(`director 503 metric unreadable: ${error instanceof Error ? error.message : error}`)
+  }
+  if (read.failed) return failed503Read('director 503 metric read failed')
+  const published = Object.values(read.byCode).flatMap((perMinute) => Object.keys(perMinute)).sort()
+  return {
+    ok: true,
+    perMinute: read.byCode['503'] ?? {},
+    minutes: minutesOf(span),
+    publishedThrough: published.at(-1) ?? null
+  }
+}
+
+const LEDGER_SUMMARY_EVENT = 'orca_relay_reserve_ledger_summary'
+const SEAT_BEHIND_ROW_EVENT = 'orca_relay_reserve_ledger_seat_behind_row'
+
+function directorLogFilter(directorService, eventClause, sinceAt) {
+  return [
+    'resource.type="cloud_run_revision"',
+    `resource.labels.service_name="${directorService}"`,
+    eventClause,
+    `timestamp>="${new Date(sinceAt).toISOString()}"`
+  ].join(' AND ')
+}
+
+// Newest summary per director instance; queued plus dropped rows are the ones not in Postgres.
+export async function readLedgerSample(reader, { directorService, at }) {
+  const read = await readLogEntries(reader, {
+    filter: directorLogFilter(directorService, `jsonPayload.event="${LEDGER_SUMMARY_EVENT}"`, at - RESERVE_WATCH_RULES.ledgerSummaryMaxAgeMs),
+    projection: 'json(timestamp,labels.instanceId,jsonPayload.pending,jsonPayload.dropped)',
+    limit: LOG_ENTRY_LIMIT
+  })
+  if (read.failed) return failedRead(read.error)
+  if (read.entries.length >= LOG_ENTRY_LIMIT) return failedRead('ledger summary read truncated')
+  if (read.entries.length === 0) return failedRead('no placing director logged a ledger summary')
+  if (read.entries.some((entry) => !Number.isFinite(Date.parse(entry.timestamp)))) {
+    return failedRead('ledger summary with an unparseable timestamp')
+  }
+  const newest = new Map()
+  for (const entry of read.entries) {
+    const instance = entry.labels?.instanceId ?? 'unknown'
+    const previous = newest.get(instance)
+    if (!previous || Date.parse(entry.timestamp) > Date.parse(previous.timestamp)) newest.set(instance, entry)
+  }
+  let unwrittenRows = 0
+  for (const entry of newest.values()) {
+    const pending = Number(entry.jsonPayload?.pending)
+    const dropped = Number(entry.jsonPayload?.dropped)
+    if (!Number.isFinite(pending) || !Number.isFinite(dropped)) return failedRead('ledger summary without counts')
+    unwrittenRows += pending + dropped
+  }
+  return { ok: true, unwrittenRows, directors: newest.size }
+}
+
+// The same read also takes the ledger summary every placing director logs each minute, as a
+// positive control: with no fresh summary in it, no seat-behind-row line is blindness, not zero.
+export async function readSeatBehindRowSample(reader, { directorService, cellId, at }) {
+  const read = await readLogEntries(reader, {
+    filter: directorLogFilter(
+      directorService,
+      `((jsonPayload.event="${SEAT_BEHIND_ROW_EVENT}" AND jsonPayload.cellId="${cellId}") OR jsonPayload.event="${LEDGER_SUMMARY_EVENT}")`,
+      at - RESERVE_WATCH_RULES.duplicateSeat.lookbackMs
+    ),
+    projection: 'json(timestamp,labels.instanceId,jsonPayload.event,jsonPayload.epoch,jsonPayload.rowEpoch)',
+    limit: LOG_ENTRY_LIMIT
+  })
+  if (read.failed) return failedRead(read.error)
+  if (read.entries.length >= LOG_ENTRY_LIMIT) return failedRead('seat-behind-row read truncated')
+  const reports = []
+  let controlSeen = false
+  for (const entry of read.entries) {
+    const reportedAt = Date.parse(entry.timestamp)
+    if (!Number.isFinite(reportedAt)) return failedRead('seat-behind-row line with an unparseable timestamp')
+    if (entry.jsonPayload?.event === LEDGER_SUMMARY_EVENT) {
+      controlSeen ||= reportedAt >= at - RESERVE_WATCH_RULES.ledgerSummaryMaxAgeMs
+      continue
+    }
+    const { epoch, rowEpoch } = entry.jsonPayload ?? {}
+    if (epoch === undefined || rowEpoch === undefined) return failedRead('seat-behind-row line without epochs')
+    // Everything the line carries but its time, which differs on every report of one seat. A
+    // proxy for the host: two hosts sharing it read as one seat, so it can only over-count.
+    reports.push({ at: reportedAt, key: `${entry.labels?.instanceId ?? 'unknown'}/${cellId}/${epoch}:${rowEpoch}` })
+  }
+  if (!controlSeen) return failedRead('seat-behind-row read holds no fresh ledger summary (positive control)')
+  return { ok: true, reports }
+}
+
+function createReader(dependencies, projectId) {
+  const now = dependencies.now
+  return {
+    runGcloud: dependencies.runGcloud,
+    fetch: dependencies.fetchImpl,
+    retryDelayMs: dependencies.retryDelayMs ?? 5_000,
+    readTimeoutMs: 20_000,
+    now,
+    // Every sample gets its own budget, so a slow read costs one sample, not the watch.
+    deadlineAt: now() + SAMPLE_READ_BUDGET_MS,
+    projectId
+  }
+}
+
+/**
+ * Same-cap and capacity rolls hold the Cloud SQL rollout lease for the whole run; it names no
+ * cell, so any live holder refuses the watch's start. Read only: holding it for hours would block
+ * every rollout. An unreadable lease refuses too (fail closed, as the lease action does).
+ */
+export async function rolloutLeaseRefusal(readRolloutLease, now) {
+  try {
+    const existing = await readRolloutLease()
+    if (!existing || existing.record.expires_at <= now) return null
+    return (
+      `a rollout holds the Cloud SQL rollout lease until ${new Date(existing.record.expires_at).toISOString()} ` +
+      `(${existing.holder}); start the watch after it ends`
+    )
+  } catch (error) {
+    return `the Cloud SQL rollout lease is unreadable (${error instanceof Error ? error.message : error}); refusing to start`
+  }
+}
+
+// The lease action's own reader, resolved from the repository root because this tree's depth
+// differs between the two repositories' copies.
+async function readProductionRolloutLease() {
+  const action = new URL('.github/actions/cloud-sql-rollout-lease/', RELAY_REPOSITORY_ROOT)
+  const [{ CloudSqlRolloutLease, describeHolder }, { createAccessTokenSource }] = await Promise.all([
+    import(new URL('storage-lease.mjs', action).href),
+    import(new URL('gcloud-access-token.mjs', action).href)
+  ])
+  const lease = new CloudSqlRolloutLease({
+    bucket: PRODUCTION_LEASE.bucket,
+    objectName: PRODUCTION_LEASE.object,
+    accessToken: createAccessTokenSource()
+  })
+  const existing = await lease.read()
+  return existing && { record: existing.record, holder: describeHolder(existing.record) }
+}
+
+// Only a missing file is a fresh watch. Anything else that is not a state fails the segment, which
+// the outcome step reads as failed and flips back: a corrupt file must not restart or refuse the watch.
+export async function loadState(path) {
+  const text = await readFile(path, 'utf8').catch((error) => (error?.code === 'ENOENT' ? null : Promise.reject(error)))
+  if (text === null) return null
+  const state = JSON.parse(text)
+  const isObject = state !== null && typeof state === 'object' && !Array.isArray(state)
+  if (!isObject || typeof state.state !== 'string') throw new Error(`state file ${path} holds no watch state`)
+  if (state.state === 'watching' && (typeof state.cellId !== 'string' || !Number.isFinite(state.startedAt))) {
+    throw new Error(`state file ${path} has no cell or start`)
+  }
+  return state
+}
+
+// A segment's config. Later segments check flipped-at against the watch's start; bad inputs before
+// the first sample are refused, so no flip back follows a typo.
+export async function parseWatchSegment(values, now = Date.now()) {
+  const existing = values['state-file'] ? await loadState(values['state-file']) : null
+  try {
+    return parseWatchArguments(values, Number.isFinite(existing?.startedAt) ? existing.startedAt : now)
+  } catch (error) {
+    if (values['state-file'] && existing === null) {
+      await writeFile(values['state-file'], JSON.stringify({ state: 'refused', verdict: error.message }))
+    }
+    throw error
+  }
+}
+
+export async function watchSegment(config, dependencies) {
+  const now = dependencies.now ?? Date.now
+  const sleep = dependencies.sleep ?? ((ms) => new Promise((resolve) => setTimeout(resolve, ms)))
+  const log = dependencies.log ?? console.log
+  const deps = { ...dependencies, now, sleep, fetchImpl: dependencies.fetchImpl ?? fetch }
+  const { request, directorService } = config
+  let state = await loadState(config.stateFile)
+  if (state === null) {
+    const refusal = await rolloutLeaseRefusal(dependencies.readRolloutLease, now())
+    if (refusal) {
+      state = { state: 'refused', verdict: refusal }
+      await writeFile(config.stateFile, JSON.stringify(state))
+      if (config.githubOutput) await writeOutcome(config.githubOutput, state)
+      return state
+    }
+    state = initialWatchState({
+      cellId: request.cellId,
+      startedAt: now(),
+      durationMinutes: config.durationMinutes,
+      background503PerMinute: config.background503PerMinute,
+      flippedAt: config.flippedAt
+    })
+  }
+  if (state.cellId !== request.cellId) throw new Error(`state file is for ${state.cellId}, not ${request.cellId}`)
+  const segmentEndsAt = now() + config.segmentMinutes * MINUTE_MS
+  while (state.state === 'watching' && now() < segmentEndsAt) {
+    const at = now()
+    const baselineEndsAt = state.flippedAt === null ? null : minuteFloor(state.flippedAt)
+    if (!state.background && baselineEndsAt !== null && at >= baselineEndsAt + METRIC_DELAY_MS) {
+      // The minutes just before the flip, once published; retried every sample until one reads.
+      const read = await readDirector503Minutes(createReader(deps, request.projectId), {
+        projectId: request.projectId,
+        directorService,
+        endedAt: baselineEndsAt,
+        minutes: BACKGROUND_MINUTES
+      })
+      state = { ...state, background: read.ok ? measuredBackground(read.perMinute, read.minutes, read.publishedThrough) : null }
+    }
+    const reader = createReader(deps, request.projectId)
+    // Serialised: concurrent Logging reads are what earn a 429.
+    const sample = {
+      at,
+      cell: await readCellSample(deps.fetchImpl, request, dependencies.idToken, sleep),
+      director503: await readDirector503Minutes(reader, {
+        projectId: request.projectId,
+        directorService,
+        // Complete minutes only, held back past the publishing delay.
+        endedAt: minuteFloor(at - METRIC_DELAY_MS),
+        minutes: RESERVE_WATCH_RULES.director503.readMinutes
+      }),
+      ledger: await readLedgerSample(reader, { directorService, at }),
+      seatBehindRow: await readSeatBehindRowSample(reader, { directorService, cellId: request.cellId, at })
+    }
+    const result = evaluateWatchSample(state, sample, { unverifiableLimit: config.unverifiableLimit })
+    state = result.state
+    log(result.line)
+    await writeFile(config.stateFile, JSON.stringify(state))
+    if (state.state !== 'watching') break
+    await sleep(Math.max(0, at + SAMPLE_INTERVAL_MS - now()))
+  }
+  if (config.githubOutput) await writeOutcome(config.githubOutput, state)
+  return state
+}
+
+function writeOutcome(githubOutput, { state, verdict }) {
+  return appendFile(githubOutput, `state=${state}\nverdict=${(verdict ?? '').replace(/[\r\n]+/g, ' ')}\n`)
+}
+
+// The job's outcome. A watch that ended mid-segment (a crash, a timeout) records failed, which the
+// workflow flips back in write mode: missing data is never healthy.
+export async function recordOutcome(stateFile, githubOutput) {
+  const state = await loadState(stateFile).catch(() => null)
+  const outcome =
+    state && state.state !== 'watching' ? state : { state: 'failed', verdict: 'the watch stopped before a verdict' }
+  await writeOutcome(githubOutput, outcome)
+  return outcome
+}
+
+function breakGlassInputs(request, generation) {
+  return {
+    mode: 'write',
+    'cell-id': request.cellId,
+    set: 'admitMode=default',
+    'break-glass': `${BREAK_GLASS_CONFIRMATION} ${request.cellId}`,
+    'expected-generation': generation,
+    confirmation: `${WRITE_CONFIRMATION} ${request.cellId}`
+  }
+}
+
+async function readRuntimeRetrying(fetchImpl, request, idToken, sleep, log) {
+  for (let attempt = 1; ; attempt += 1) {
+    try {
+      return await readRuntime(fetchImpl, request, idToken, sleep)
+    } catch (error) {
+      if (attempt >= FLIP_BACK_RUNTIME_ATTEMPTS) throw error
+      log(JSON.stringify({ event: 'orca_relay_reserve_watch_runtime_retry', cellId: request.cellId, attempt, message: error instanceof Error ? error.message : String(error) }))
+      await sleep(FLIP_BACK_BACKOFF_MS * 2 ** (attempt - 1))
+    }
+  }
+}
+
+/**
+ * The flag tool's admitMode=db write at the generation read just now. A generation conflict (some
+ * other write landed first) is re-read and retried while the switch file still says reserve; a
+ * file that already says db is done. A cell that still does not answer after the runtime retries
+ * needs break glass, which only an operator may type.
+ */
+export async function flipBack(values, dependencies) {
+  const log = dependencies.log ?? console.log
+  const fetchImpl = dependencies.fetchImpl ?? fetch
+  const sleep = dependencies.sleep ?? ((ms) => new Promise((resolve) => setTimeout(resolve, ms)))
+  const base = dbFlipRequest(values)
+  const record = (result) =>
+    log(
+      JSON.stringify({
+        event: 'orca_relay_reserve_watch_flip_back',
+        mode: base.mode,
+        cellId: base.cellId,
+        written: result.written,
+        generation: result.generation ?? null,
+        ...(result.alreadyDb ? { alreadyDb: true } : {})
+      })
+    )
+  for (let attempt = 1; ; attempt += 1) {
+    const current = await readCurrentObject(fetchImpl, base, dependencies.accessToken)
+    if ((current.object?.flags?.admitMode ?? 'db') !== 'reserve') {
+      const result = { written: false, alreadyDb: true, generation: current.generation }
+      record(result)
+      return result
+    }
+    try {
+      await readRuntimeRetrying(fetchImpl, base, dependencies.idToken, sleep, log)
+    } catch (error) {
+      const inputs = breakGlassInputs(base, current.generation)
+      log(JSON.stringify({ event: 'orca_relay_reserve_watch_break_glass_required', cellId: base.cellId, inputs }))
+      throw new Error(
+        `${base.cellId} does not answer runtime-status (${error instanceof Error ? error.message : error}), so ` +
+          'the flag tool cannot flip it back. Run Operate Relay Production Cell Flags with break glass: ' +
+          Object.entries(inputs).map(([key, value]) => `${key}=${value}`).join(', ')
+      )
+    }
+    try {
+      const result = await operateCellFlags(
+        { ...base, expectedGeneration: current.generation },
+        { ...dependencies, fetchImpl, log, sleep }
+      )
+      record(result)
+      return result
+    } catch (error) {
+      if (!error?.generationConflict || attempt >= FLIP_BACK_ATTEMPTS) throw error
+      log(JSON.stringify({ event: 'orca_relay_reserve_watch_flip_back_conflict', cellId: base.cellId, attempt, message: error.message }))
+      await sleep(FLIP_BACK_BACKOFF_MS * 2 ** (attempt - 1))
+    }
+  }
+}
+
+export const ALARM_EVENT = 'orca_relay_reserve_watch_alarm'
+export const ALARM_LOG_NAME = 'orca-relay-reserve-watch'
+export const ALARM_KINDS = ['refused', 'flip-back-failed']
+
+// GitHub's workflow-command escaping, so a verdict cannot end the annotation or start another.
+export const escapeWorkflowCommand = (text) => text.replace(/%/g, '%25').replace(/\r/g, '%0D').replace(/\n/g, '%0A')
+const escapeWorkflowProperty = (text) => escapeWorkflowCommand(text).replace(/:/g, '%3A').replace(/,/g, '%2C')
+
+/**
+ * Both kinds can leave the cell in reserve with nothing watching it. Only write mode pages: a
+ * dry-run watch never flips back, so its refusal or failure changes nothing for the cell.
+ */
+export async function raiseAlarm({ mode, cellId, projectId, kind, reason }, { runGcloud, log = console.log }) {
+  if (!ALARM_KINDS.includes(kind)) throw new Error(`--kind must be one of ${ALARM_KINDS.join(', ')}`)
+  const message =
+    kind === 'refused'
+      ? `${cellId} may be in reserve mode and unwatched: the watch was refused (${reason}). Rerun it, or flip the cell to db with Operate Relay Production Cell Flags.`
+      : `${cellId} may still be in reserve mode with no watch: the flip back did not land (${reason}). Flip it to db by hand with Operate Relay Production Cell Flags (set=admitMode=db), or with the break-glass inputs logged above if the cell does not answer.`
+  log(`::error title=${escapeWorkflowProperty(`Reserve watch ${kind} for ${cellId}`)}::${escapeWorkflowCommand(message)}`)
+  if (mode !== 'write') return { paged: false }
+  const entry = JSON.stringify({ event: ALARM_EVENT, cellId, kind, reason })
+  try {
+    // The flag tool's project check: a missing --project-id must not reach gcloud as "undefined".
+    if (!/^[a-z][a-z0-9-]{4,62}$/.test(projectId ?? '')) throw new Error('--project-id is not a GCP project id')
+    await runGcloud(
+      ['logging', 'write', ALARM_LOG_NAME, entry, '--payload-type=json', '--severity=ERROR', '--project', projectId],
+      { timeoutMs: 30_000 }
+    )
+  } catch (error) {
+    log(`::error title=Reserve watch page failed::${escapeWorkflowCommand(`the alarm log line was not written, so nothing pages: ${error instanceof Error ? error.message : error}`)}`)
+    return { paged: false }
+  }
+  return { paged: true }
+}
+
+if (import.meta.url === pathToFileURL(process.argv[1] ?? '').href) {
+  const [command, ...rest] = process.argv.slice(2)
+  const { values } = parseArgs({
+    args: rest,
+    options: Object.fromEntries(
+      [
+        'mode',
+        'cell-id',
+        'cell-origin',
+        'project-id',
+        'director-origin',
+        'director-service',
+        'confirmation',
+        'duration-minutes',
+        'segment-minutes',
+        'unverifiable-limit',
+        'background-503-per-minute',
+        'flipped-at',
+        'state-file',
+        'github-output',
+        'verdict',
+        'kind'
+      ].map((name) => [name, { type: 'string' }])
+    )
+  })
+  const idToken = process.env.ORCA_RELAY_ADMIN_ID_TOKEN
+  const runGcloud = (args, { timeoutMs }) =>
+    execFileAsync('gcloud', args, { maxBuffer: 64 * 1024 * 1024, timeout: timeoutMs, killSignal: 'SIGKILL' })
+  try {
+    if (command === 'alarm') {
+      await raiseAlarm(
+        { mode: values.mode, cellId: values['cell-id'] ?? '', projectId: values['project-id'], kind: values.kind, reason: values.verdict ?? '' },
+        { runGcloud }
+      )
+      // The alarm step is red, whatever the page did.
+      process.exitCode = 1
+    } else if (command === 'outcome') {
+      if (!values['state-file'] || !values['github-output']) throw new Error('--state-file and --github-output are required')
+      console.log(JSON.stringify({ event: 'orca_relay_reserve_watch_outcome', ...(await recordOutcome(values['state-file'], values['github-output'])) }))
+    } else if (!idToken) {
+      throw new Error('ORCA_RELAY_ADMIN_ID_TOKEN is required')
+    } else if (command === 'watch') {
+      const config = await parseWatchSegment(values)
+      const state = await watchSegment(config, {
+        idToken,
+        readRolloutLease: readProductionRolloutLease,
+        runGcloud
+      })
+      // A cell that was never reserve is the operator's mistake; one flipped back by hand is not.
+      if (state.state === 'refused') {
+        console.log(JSON.stringify({ event: 'orca_relay_reserve_watch_refused', verdict: state.verdict }))
+        process.exitCode = 1
+      }
+    } else if (command === 'flip-back') {
+      const accessToken = process.env.ORCA_RELAY_CONTROL_ACCESS_TOKEN
+      if (!accessToken) throw new Error('ORCA_RELAY_CONTROL_ACCESS_TOKEN is required')
+      if (!values.verdict) throw new Error('--verdict is required')
+      await flipBack(values, {
+        accessToken,
+        idToken,
+        audit: {
+          writtenBy: `reserve-watch/${process.env.GITHUB_TRIGGERING_ACTOR ?? process.env.GITHUB_ACTOR ?? 'unknown'}`,
+          runId: process.env.GITHUB_RUN_ID ?? 'local',
+          runAttempt: process.env.GITHUB_RUN_ATTEMPT ?? '0'
+        }
+      })
+      // A trip is a failed watch even once the cell is safely back on db.
+      console.log(JSON.stringify({ event: 'orca_relay_reserve_watch_verdict', mode: values.mode, verdict: values.verdict }))
+      process.exitCode = 1
+    } else {
+      throw new Error('command must be watch, outcome, flip-back or alarm')
+    }
+  } catch (error) {
+    console.log(
+      JSON.stringify({
+        event: 'orca_relay_reserve_watch_failed',
+        message: error instanceof Error ? error.message : String(error)
+      })
+    )
+    process.exitCode = 1
+  }
+}

cloud/dev/scripts/watch-relay-reserve-cell.test.mjs

@@ -0,0 +1,958 @@
+import assert from 'node:assert/strict'
+import { readFileSync } from 'node:fs'
+import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { test } from 'node:test'
+import {
+  evaluateWatchSample,
+  initialWatchState,
+  measuredBackground
+} from './relay-reserve-watch-rules.mjs'
+import {
+  ALARM_EVENT,
+  ALARM_LOG_NAME,
+  FLIP_BACK_ATTEMPTS,
+  FLIP_BACK_RUNTIME_ATTEMPTS,
+  flipBack,
+  parseWatchArguments,
+  parseWatchSegment,
+  raiseAlarm,
+  readDirector503Minutes,
+  readSeatBehindRowSample,
+  recordOutcome,
+  rolloutLeaseRefusal,
+  watchSegment
+} from './watch-relay-reserve-cell.mjs'
+import { readRelayWorkflow } from './relay-repository.mjs'
+
+const CELL = 'production-gce-c26'
+const START = Date.parse('2026-10-14T12:00:00Z')
+const MINUTE = 60_000
+const LIMIT = { unverifiableLimit: 3 }
+
+const healthyCell = (overrides = {}) => ({
+  ok: true,
+  admitModeConfigured: 'reserve',
+  admitModeEffective: 'reserve',
+  draining: false,
+  controls: 1_000,
+  enforcedUnits: 1_100,
+  hardCap: 2_000,
+  ...overrides
+})
+const minutesBefore = (at, count) =>
+  Array.from({ length: count }, (_, index) =>
+    new Date(Math.floor(at / MINUTE) * MINUTE - (count + 3 - index) * MINUTE).toISOString().slice(0, 16)
+  )
+const quiet503 = (at, perMinute = 10) => {
+  const minutes = minutesBefore(at, 3)
+  return { ok: true, minutes, perMinute: Object.fromEntries(minutes.map((minute) => [minute, perMinute])), publishedThrough: minutes.at(-1) }
+}
+const sampleAt = (at, overrides = {}) => ({
+  at,
+  cell: healthyCell(),
+  director503: quiet503(at),
+  ledger: { ok: true, unwrittenRows: 0, directors: 5 },
+  seatBehindRow: { ok: true, reports: [] },
+  ...overrides
+})
+const freshState = (overrides = {}) => ({
+  ...initialWatchState({ cellId: CELL, startedAt: START, durationMinutes: 60, background503PerMinute: 10 }),
+  ...overrides
+})
+
+// Runs samples one minute apart; returns every result.
+function run(samples, state = freshState()) {
+  const results = []
+  for (const sample of samples) {
+    const result = evaluateWatchSample(state, sample, LIMIT)
+    results.push(result)
+    state = result.state
+    if (state.state !== 'watching') break
+  }
+  return results
+}
+
+test('a healthy cell keeps watching and completes at the deadline', () => {
+  const results = run(Array.from({ length: 61 }, (_, minute) => sampleAt(START + minute * MINUTE)))
+  assert.equal(results.length, 61)
+  assert.ok(results.slice(0, 60).every((result) => result.state.state === 'watching'))
+  assert.equal(results.at(-1).state.state, 'completed')
+  const line = JSON.parse(results[0].line)
+  assert.equal(line.event, 'orca_relay_reserve_watch_sample')
+  assert.deepEqual(Object.keys(line.rules), ['deadMan', 'overrun', 'director503', 'duplicateSeat', 'hostDrop', 'ledgerDrift'])
+})
+
+test('503s trip only after three complete minutes over max(1.5x background, background + 20)', () => {
+  // Background 100: the line is 150. Two minutes over and one under is not a trip.
+  const state = freshState({ background: { medianPerMinute: 100, source: 'input' } })
+  const series = (counts) => {
+    const minutes = minutesBefore(START, 3)
+    return {
+      ok: true,
+      minutes,
+      perMinute: Object.fromEntries(minutes.map((minute, index) => [minute, counts[index]])),
+      publishedThrough: minutes.at(-1)
+    }
+  }
+  const twoOver = evaluateWatchSample(state, sampleAt(START, { director503: series([151, 140, 200]) }), LIMIT)
+  assert.equal(twoOver.state.state, 'watching')
+  const atLine = evaluateWatchSample(state, sampleAt(START, { director503: series([150, 150, 150]) }), LIMIT)
+  assert.equal(atLine.state.state, 'watching')
+  const threeOver = evaluateWatchSample(state, sampleAt(START, { director503: series([151, 160, 170]) }), LIMIT)
+  assert.equal(threeOver.state.state, 'tripped')
+  assert.match(threeOver.state.verdict, /^flip back production-gce-c26: director503: /)
+  // A zero background still needs 20 a minute, not one 503.
+  const zero = freshState({ background: { medianPerMinute: 0, source: 'input' } })
+  assert.equal(evaluateWatchSample(zero, sampleAt(START, { director503: series([5, 5, 5]) }), LIMIT).state.state, 'watching')
+  assert.equal(evaluateWatchSample(zero, sampleAt(START, { director503: series([21, 21, 21]) }), LIMIT).state.state, 'tripped')
+  // A published minute with no 503 point had none.
+  const gap = series([151, 160, 170])
+  delete gap.perMinute[gap.minutes[1]]
+  assert.equal(evaluateWatchSample(state, sampleAt(START, { director503: gap }), LIMIT).state.state, 'watching')
+})
+
+test('a quiet fleet with no 503 series at all reads zero, not unverifiable', () => {
+  const minutes = minutesBefore(START, 10)
+  const quiet = { ok: true, minutes, perMinute: {}, publishedThrough: minutes.at(-1) }
+  const [result] = run([sampleAt(START, { director503: quiet })], freshState({ background: { medianPerMinute: 0, source: 'measured' } }))
+  assert.equal(result.rules.director503.status, 'pass')
+  assert.deepEqual(Object.values(result.rules.director503.recent), [0, 0, 0])
+  assert.equal(result.state.unverifiableStreak, 0)
+  assert.deepEqual(measuredBackground({}, minutes, minutes.at(-1)), { medianPerMinute: 0, minutes: 10, source: 'measured' })
+})
+
+test('503 minutes past the publishing front are left out, and only no front at all is unverifiable', () => {
+  const state = freshState({ background: { medianPerMinute: 100, source: 'input' } })
+  const minutes = minutesBefore(START, 10)
+  const judge = (director503) => evaluateWatchSample(state, sampleAt(START, { director503 }), LIMIT)
+  // The front is two minutes behind: the three minutes before it are judged, the two after are not.
+  const over = Object.fromEntries(minutes.map((minute) => [minute, 500]))
+  const behind = judge({ ok: true, minutes, perMinute: over, publishedThrough: minutes[7] })
+  assert.equal(behind.rules.director503.status, 'trip')
+  assert.deepEqual(Object.keys(behind.rules.director503.recent), minutes.slice(5, 8))
+  // A sparse 503 series: minutes after its last point but within the front are zero, not late.
+  const sparse = judge({ ok: true, minutes, perMinute: { [minutes[2]]: 500 }, publishedThrough: minutes.at(-1) })
+  assert.equal(sparse.rules.director503.status, 'pass')
+  assert.equal(sparse.state.unverifiableStreak, 0)
+  // No request series at all.
+  const empty = judge({ ok: true, minutes, perMinute: {}, publishedThrough: null })
+  assert.equal(empty.rules.director503.status, 'unverifiable')
+  assert.match(empty.rules.director503.reason, /request series is empty/)
+  assert.equal(empty.state.unverifiableStreak, 1)
+  // A front too far behind to leave three published minutes in the read.
+  const stale = judge({ ok: true, minutes, perMinute: {}, publishedThrough: minutes[1] })
+  assert.equal(stale.rules.director503.status, 'unverifiable')
+  assert.match(stale.rules.director503.reason, new RegExp(`published only through ${minutes[1]}`))
+})
+
+const SEAT_EVENT = 'orca_relay_reserve_ledger_seat_behind_row'
+const SUMMARY_EVENT = 'orca_relay_reserve_ledger_summary'
+const summaryLine = (at, instanceId = 'a') => ({
+  timestamp: new Date(at).toISOString(),
+  labels: { instanceId },
+  jsonPayload: { event: SUMMARY_EVENT, pending: 0, dropped: 0 }
+})
+
+// A Cloud Monitoring or Cloud Logging reader over fakes, as the watch builds one.
+const fakeReader = ({ fetch: fetchImpl = async () => new Response('{}'), runGcloud = async () => ({ stdout: 'token' }) } = {}) => ({
+  fetch: fetchImpl,
+  runGcloud,
+  retryDelayMs: 0,
+  readTimeoutMs: 1_000,
+  now: () => START,
+  deadlineAt: START + 10_000,
+  projectId: 'onorca-cloud'
+})
+
+test('an unparseable timestamp makes the read unverifiable', async () => {
+  const badPoint = {
+    timeSeries: [{ points: [{ interval: { endTime: 'not a time' }, value: { int64Value: '3' } }] }]
+  }
+  const read503 = await readDirector503Minutes(
+    fakeReader({ fetch: async () => new Response(JSON.stringify(badPoint)) }),
+    { projectId: 'onorca-cloud', directorService: 'orca-cloud-relay', endedAt: START, minutes: 3 }
+  )
+  assert.equal(read503.ok, false)
+  assert.match(read503.error, /unreadable/)
+  const seats = await readSeatBehindRowSample(
+    fakeReader({ runGcloud: async () => ({ stdout: JSON.stringify([summaryLine(START), { timestamp: 'garbage', jsonPayload: { epoch: 1, rowEpoch: 2 } }]) }) }),
+    { directorService: 'orca-cloud-relay', cellId: CELL, at: START }
+  )
+  assert.equal(seats.ok, false)
+  assert.match(seats.error, /unparseable timestamp/)
+})
+
+test('a seat-behind-row key carries the director instance and cell beside the epochs', async () => {
+  const entry = (instanceId) => ({ timestamp: new Date(START).toISOString(), labels: { instanceId }, jsonPayload: { event: SEAT_EVENT, epoch: 7, rowEpoch: 8 } })
+  const read = await readSeatBehindRowSample(
+    fakeReader({ runGcloud: async () => ({ stdout: JSON.stringify([entry('a'), summaryLine(START), entry('b')]) }) }),
+    { directorService: 'orca-cloud-relay', cellId: CELL, at: START }
+  )
+  assert.deepEqual(read.reports.map((report) => report.key), [`a/${CELL}/7:8`, `b/${CELL}/7:8`])
+})
+
+test('the duplicate read carries a positive control: no fresh ledger summary in it is unverifiable, not zero', async () => {
+  const filters = []
+  const readWith = (entries) =>
+    readSeatBehindRowSample(
+      fakeReader({ runGcloud: async (args) => (filters.push(args[2]), { stdout: JSON.stringify(entries) }) }),
+      { directorService: 'orca-cloud-relay', cellId: CELL, at: START }
+    )
+  // One read, one window: the control rides the same filter as the seat lines.
+  assert.deepEqual(await readWith([summaryLine(START - 30_000)]), { ok: true, reports: [] })
+  assert.match(filters[0], new RegExp(`\\(\\(jsonPayload\\.event="${SEAT_EVENT}" AND jsonPayload\\.cellId="${CELL}"\\) OR jsonPayload\\.event="${SUMMARY_EVENT}"\\)`))
+  assert.match(filters[0], /resource\.labels\.service_name="orca-cloud-relay"/)
+  for (const entries of [[], [summaryLine(START - 10 * MINUTE)]]) {
+    const blind = await readWith(entries)
+    assert.equal(blind.ok, false)
+    assert.match(blind.error, /no fresh ledger summary \(positive control\)/)
+  }
+  // Blind is a gap in the sample, which counts toward the unverifiable streak.
+  const [gap] = run([sampleAt(START, { seatBehindRow: await readWith([]) })])
+  assert.equal(gap.rules.duplicateSeat.status, 'unverifiable')
+  assert.equal(gap.state.unverifiableStreak, 1)
+})
+
+test('the measured background is the median of the minutes before the flip, published gaps as zero', () => {
+  const minutes = ['2026-10-14T11:47', '2026-10-14T11:48', '2026-10-14T11:49']
+  const front = minutes[2]
+  assert.equal(measuredBackground({ '2026-10-14T11:48': 30 }, minutes, front).medianPerMinute, 0)
+  assert.equal(measuredBackground({ [minutes[0]]: 10, [minutes[1]]: 30, [minutes[2]]: 20 }, minutes, front).medianPerMinute, 20)
+  assert.equal(measuredBackground({}, minutes, '2026-10-14T11:52').medianPerMinute, 0)
+  // Not published through the last minute yet, or no request series: no background yet.
+  assert.equal(measuredBackground({ [minutes[0]]: 10 }, minutes, minutes[1]), null)
+  assert.equal(measuredBackground({}, minutes, null), null)
+  assert.equal(measuredBackground({}, [], front), null)
+})
+
+test('any overrun trips at once', () => {
+  const [result] = run([sampleAt(START, { cell: healthyCell({ enforcedUnits: 2_001 }) })])
+  assert.equal(result.state.state, 'tripped')
+  assert.match(result.state.verdict, /overrun: ledger units 2001 over hard cap 2000/)
+  const [atCap] = run([sampleAt(START, { cell: healthyCell({ enforcedUnits: 2_000 }) })])
+  assert.equal(atCap.state.state, 'watching')
+})
+
+test('a seat reported behind its row again past the grace trips; inside the grace it does not', () => {
+  const report = (at) => ({ at, key: '7:8' })
+  const within = run([
+    sampleAt(START, { seatBehindRow: { ok: true, reports: [report(START - 30_000)] } }),
+    sampleAt(START + MINUTE, { seatBehindRow: { ok: true, reports: [report(START - 30_000), report(START + 50_000)] } })
+  ])
+  assert.equal(within.at(-1).state.state, 'watching')
+  // Carried across samples: the second read no longer holds the first report.
+  const past = run([
+    sampleAt(START, { seatBehindRow: { ok: true, reports: [report(START - 30_000)] } }),
+    sampleAt(START + 2 * MINUTE, { seatBehindRow: { ok: true, reports: [report(START + 70_000)] } })
+  ])
+  assert.equal(past.at(-1).state.state, 'tripped')
+  assert.match(past.at(-1).state.verdict, /duplicateSeat: a seat behind its ledger row \(7:8\)/)
+  // Another seat's report is not the same seat.
+  const other = run([
+    sampleAt(START, { seatBehindRow: { ok: true, reports: [report(START - 30_000)] } }),
+    sampleAt(START + 2 * MINUTE, { seatBehindRow: { ok: true, reports: [{ at: START + 70_000, key: '3:4' }] } })
+  ])
+  assert.equal(other.at(-1).state.state, 'watching')
+})
+
+test('a host drop over 5% in 5 minutes trips outside a drain only', () => {
+  const controls = [1_000, 990, 980, 960, 949]
+  const samples = (draining) =>
+    controls.map((count, minute) =>
+      sampleAt(START + minute * MINUTE, { cell: healthyCell({ controls: count, draining: draining && minute === 2 }) })
+    )
+  assert.equal(run(samples(false)).at(-1).state.state, 'tripped')
+  assert.match(run(samples(false)).at(-1).state.verdict, /hostDrop: connected hosts fell 1000 -> 949/)
+  assert.equal(run(samples(true)).at(-1).state.state, 'watching')
+  // A drop spread over more than 5 minutes is not one.
+  const slow = [1_000, 990, 980, 970, 960, 955, 950].map((count, minute) =>
+    sampleAt(START + minute * MINUTE, { cell: healthyCell({ controls: count }) })
+  )
+  assert.equal(run(slow).at(-1).state.state, 'watching')
+  // Under the floor on a nearly empty cell: 100 -> 92 is 8%, but only 8 hosts.
+  const small = [100, 92].map((count, minute) => sampleAt(START + minute * MINUTE, { cell: healthyCell({ controls: count }) }))
+  assert.equal(run(small).at(-1).state.state, 'watching')
+})
+
+test('ledger drift over 0.5% trips once it has lasted 2 minutes', () => {
+  const drift = (rows) => ({ ok: true, unwrittenRows: rows, directors: 5 })
+  const brief = run([6, 6, 0, 6].map((rows, minute) => sampleAt(START + minute * MINUTE, { ledger: drift(rows) })))
+  assert.equal(brief.at(-1).state.state, 'watching')
+  const sustained = run([6, 6, 6].map((rows, minute) => sampleAt(START + minute * MINUTE, { ledger: drift(rows) })))
+  assert.equal(sustained.length, 3)
+  assert.equal(sustained.at(-1).state.state, 'tripped')
+  assert.match(sustained.at(-1).state.verdict, /ledgerDrift: ledger drift 0\.60% for 120 s/)
+  const atLine = run([5, 5, 5, 5].map((rows, minute) => sampleAt(START + minute * MINUTE, { ledger: drift(rows) })))
+  assert.equal(atLine.at(-1).state.state, 'watching')
+})
+
+test('a tripped dead-man is a trip: Postgres still records reserve until a db write', () => {
+  const [result] = run([sampleAt(START, { cell: healthyCell({ admitModeEffective: 'db' }) })])
+  assert.equal(result.state.state, 'tripped')
+  assert.match(result.state.verdict, /deadMan: the cell tripped its dead-man/)
+})
+
+test('unreadable data is never healthy: N samples in a row with any gap trip', () => {
+  const unreadable = { ok: false, error: 'cell runtime status returned 502' }
+  const gaps = [
+    { cell: unreadable },
+    { ledger: { ok: false, error: 'no placing director logged a ledger summary' } },
+    { director503: { ok: false, error: 'director 503 metric read failed', perMinute: {}, minutes: [], publishedThrough: null } }
+  ]
+  const results = run(gaps.map((gap, minute) => sampleAt(START + minute * MINUTE, gap)))
+  assert.deepEqual(results.map((result) => result.state.unverifiableStreak), [1, 2, 3])
+  assert.equal(results.at(-1).state.state, 'tripped')
+  assert.match(results.at(-1).state.verdict, /unverifiable: 3 samples in a row could not read director503/)
+  // One whole sample resets the streak.
+  const reset = run(
+    [gaps[0], {}, gaps[0], gaps[0]].map((gap, minute) => sampleAt(START + minute * MINUTE, gap))
+  )
+  assert.deepEqual(reset.map((result) => result.state.unverifiableStreak), [1, 0, 1, 2])
+  assert.equal(reset.at(-1).state.state, 'watching')
+  // No background yet is a gap too.
+  const noBackground = freshState({ background: null })
+  assert.equal(evaluateWatchSample(noBackground, sampleAt(START), LIMIT).state.unverifiableStreak, 1)
+  // A measured trip still wins over a gap elsewhere.
+  const [both] = run([sampleAt(START, { cell: healthyCell({ enforcedUnits: 2_500 }), ledger: gaps[1].ledger })])
+  assert.equal(both.state.state, 'tripped')
+  assert.match(both.state.verdict, /overrun/)
+})
+
+test('a cell no longer in reserve stops the watch without a write; one never in reserve is refused', () => {
+  const dbCell = healthyCell({ admitModeConfigured: 'db', admitModeEffective: 'reserve' })
+  const later = run([sampleAt(START), sampleAt(START + MINUTE, { cell: dbCell })])
+  assert.equal(later.at(-1).state.state, 'stopped')
+  assert.match(later.at(-1).state.verdict, /is not in reserve mode \(switch file says db\)/)
+  // Never reserve within the wait: refused, which the workflow does not flip back.
+  const never = run(Array.from({ length: 12 }, (_, minute) => sampleAt(START + minute * MINUTE, { cell: dbCell })))
+  assert.equal(never.at(-1).state.state, 'refused')
+  assert.equal(never.length, 11)
+  assert.match(never.at(-1).state.verdict, /did not report reserve mode within 10 min \(switch file says db\)/)
+})
+
+test('a watch dispatched before the reserve write applied waits for it, then judges as usual', () => {
+  const dbCell = healthyCell({ admitModeConfigured: 'db', admitModeEffective: 'db' })
+  const results = run([
+    sampleAt(START, { cell: dbCell, ledger: { ok: false, error: 'not yet' } }),
+    sampleAt(START + MINUTE, { cell: dbCell }),
+    sampleAt(START + 2 * MINUTE),
+    sampleAt(START + 3 * MINUTE, { cell: healthyCell({ enforcedUnits: 2_500 }) })
+  ])
+  // Waiting samples judge nothing: not a dead-man trip on a db cell, not a gap.
+  assert.deepEqual(results.slice(0, 2).map((result) => result.state.state), ['watching', 'watching'])
+  assert.equal(results[0].rules.reserve.status, 'pending')
+  assert.deepEqual(Object.keys(results[0].rules), ['reserve'])
+  assert.deepEqual(results.map((result) => result.state.unverifiableStreak), [0, 0, 0, 0])
+  assert.equal(results[2].state.reserveSeenAt, START + 2 * MINUTE)
+  assert.equal(results[2].rules.overrun.status, 'pass')
+  assert.equal(results[3].state.state, 'tripped')
+  // Once seen in reserve, a later db reads as a hand flip back, not a refusal.
+  const after = run([sampleAt(START + 2 * MINUTE), sampleAt(START + 3 * MINUTE, { cell: dbCell })], results[0].state)
+  assert.equal(after.at(-1).state.state, 'stopped')
+})
+
+const watchValues = (overrides = {}) => ({
+  mode: 'dry-run',
+  'cell-id': CELL,
+  'cell-origin': 'https://c26.relay.onorca.dev',
+  'project-id': 'onorca-cloud',
+  'director-origin': 'https://relay.onorca.dev',
+  'director-service': 'orca-cloud-relay',
+  'duration-minutes': '120',
+  'segment-minutes': '55',
+  'unverifiable-limit': '3',
+  'background-503-per-minute': '',
+  'flipped-at': new Date(START - 5 * MINUTE).toISOString(),
+  'state-file': '/unused',
+  ...overrides
+})
+const parseAt = (values) => parseWatchArguments(values, START)
+
+test('watch arguments reuse the flag tool checks, so a write watch needs its confirmation up front', () => {
+  assert.equal(parseAt(watchValues()).background503PerMinute, null)
+  assert.equal(parseAt(watchValues()).flippedAt, START - 5 * MINUTE)
+  assert.equal(parseAt(watchValues({ 'background-503-per-minute': '12.5', 'flipped-at': '' })).background503PerMinute, 12.5)
+  assert.throws(() => parseAt(watchValues({ mode: 'write' })), /WRITE_RELAY_CELL_FLAGS production-gce-c26/)
+  assert.doesNotThrow(() => parseAt(watchValues({ mode: 'write', confirmation: `WRITE_RELAY_CELL_FLAGS ${CELL}` })))
+  assert.throws(() => parseAt(watchValues({ 'cell-origin': 'https://c27.relay.onorca.dev' })), /names another cell/)
+  assert.throws(() => parseAt(watchValues({ 'duration-minutes': '301' })), /duration-minutes/)
+  assert.throws(() => parseAt(watchValues({ 'segment-minutes': '56' })), /segment-minutes/)
+  assert.throws(() => parseAt(watchValues({ 'background-503-per-minute': '-1' })), /background/)
+})
+
+test('the 503 background needs flipped-at or an explicit value; flipped-at must be a recent zoned time', () => {
+  assert.throws(() => parseAt(watchValues({ 'flipped-at': '' })), /--flipped-at .* or --background-503-per-minute is required/)
+  assert.throws(() => parseAt(watchValues({ 'flipped-at': '2026-10-14T11:55:00' })), /ISO time with a zone/)
+  assert.throws(() => parseAt(watchValues({ 'flipped-at': '2026-10-14T12:01:00Z' })), /in the future/)
+  assert.throws(() => parseAt(watchValues({ 'flipped-at': '2026-10-13T11:00:00Z' })), /more than 24 hours/)
+  assert.equal(parseAt(watchValues({ 'flipped-at': '2026-10-14T07:55:00-04:00' })).flippedAt, START - 5 * MINUTE)
+})
+
+// A cell, Cloud Monitoring and Cloud Logging, answering from `script` per sample.
+function fakeSources({
+  cell = () => ({}),
+  // Logged at each sample's minute, so it stays fresh for the positive control.
+  ledger = (sample) => [{ timestamp: new Date(START + (sample - 1) * MINUTE).toISOString(), labels: { instanceId: 'a' }, jsonPayload: { pending: 0, dropped: 0 } }],
+  per503 = () => 10,
+  // Requests of every other code per minute; zero is a minute with no traffic at all.
+  perOther = () => 1_000,
+  seatControl = true,
+  rolloutLease = null
+} = {}) {
+  let samples = 0
+  const monitoringSpans = []
+  const fetchImpl = async (url) => {
+    const text = String(url)
+    if (text.endsWith('/v1/admin/runtime-status')) {
+      samples += 1
+      const overrides = cell(samples)
+      return new Response(
+        JSON.stringify({
+          role: 'cell',
+          cellId: CELL,
+          draining: false,
+          flagsApplied: { generation: 4, flags: { admitMode: 'reserve', readinessLocal: true } },
+          admitModeEffective: 'reserve',
+          connectionCapacity: { hardCap: 2_000 },
+          runtime: { controls: 1_000, enforcedConnectionUnits: 1_100 },
+          ...overrides
+        })
+      )
+    }
+    if (text.startsWith('https://monitoring.googleapis.com/')) {
+      const params = new URL(text).searchParams
+      const span = { startedAt: Date.parse(params.get('interval.startTime')), endedAt: Date.parse(params.get('interval.endTime')) }
+      monitoringSpans.push(span)
+      assert.equal(params.get('aggregation.groupByFields'), 'metric.label."response_code"')
+      assert.doesNotMatch(params.get('filter'), /response_code/)
+      // Cloud Monitoring returns no point for a minute with no requests of that code.
+      const seriesOf = (code, count) => {
+        const points = []
+        for (let minute = span.startedAt; minute < span.endedAt; minute += MINUTE) {
+          if (count(minute) > 0) {
+            points.push({ interval: { endTime: new Date(minute + MINUTE).toISOString() }, value: { int64Value: String(count(minute)) } })
+          }
+        }
+        return points.length === 0 ? [] : [{ metric: { labels: { response_code: code } }, points }]
+      }
+      return new Response(JSON.stringify({ timeSeries: [...seriesOf('200', perOther), ...seriesOf('503', per503)] }))
+    }
+    throw new Error(`unexpected fetch ${text}`)
+  }
+  const runGcloud = async (args) => {
+    if (args[0] === 'auth') return { stdout: 'token' }
+    const filter = args[2]
+    const summaries = ledger(samples).map((entry) => ({ ...entry, jsonPayload: { ...entry.jsonPayload, event: SUMMARY_EVENT } }))
+    if (filter.includes(SEAT_EVENT)) return { stdout: JSON.stringify(seatControl ? summaries : []) }
+    if (filter.includes(SUMMARY_EVENT)) return { stdout: JSON.stringify(ledger(samples)) }
+    return { stdout: '[]' }
+  }
+  return { fetchImpl, runGcloud, monitoringSpans, readRolloutLease: async () => rolloutLease }
+}
+
+test('a segment samples once a minute, carries state to the next segment, and stops on a trip', async () => {
+  const directory = await mkdtemp(join(tmpdir(), 'relay-reserve-watch-'))
+  try {
+    let clock = START
+    const lines = []
+    const config = parseAt(
+      watchValues({ 'state-file': join(directory, 'state.json'), 'github-output': join(directory, 'out'), 'segment-minutes': '3' })
+    )
+    const sources = fakeSources({ cell: (sample) => (sample === 5 ? { runtime: { controls: 1_000, enforcedConnectionUnits: 2_001 } } : {}) })
+    const dependencies = {
+      ...sources,
+      idToken: 'id',
+      retryDelayMs: 0,
+      now: () => clock,
+      sleep: async (ms) => { clock += ms },
+      log: (line) => lines.push(JSON.parse(line))
+    }
+    const first = await watchSegment(config, dependencies)
+    assert.equal(first.state, 'watching')
+    assert.equal(lines.length, 3)
+    assert.equal(first.background.source, 'measured')
+    assert.equal(await readFile(join(directory, 'out'), 'utf8'), 'state=watching\nverdict=\n')
+    const second = await watchSegment(config, dependencies)
+    assert.equal(second.state, 'tripped')
+    assert.equal(second.samples, 5)
+    assert.deepEqual(lines.map((line) => line.sample), [1, 2, 3, 4, 5])
+    assert.equal(lines.at(-1).rules.overrun.status, 'trip')
+    assert.match(await readFile(join(directory, 'out'), 'utf8'), /state=tripped\nverdict=flip back production-gce-c26: overrun: /)
+  } finally {
+    await rm(directory, { recursive: true, force: true })
+  }
+})
+
+test('a sample that cannot read its sources counts as unverifiable rather than crashing the watch', async () => {
+  const directory = await mkdtemp(join(tmpdir(), 'relay-reserve-watch-'))
+  try {
+    let clock = START
+    const config = parseAt(watchValues({ 'state-file': join(directory, 'state.json'), 'segment-minutes': '10' }))
+    const sources = fakeSources({ ledger: () => [] })
+    const state = await watchSegment(config, {
+      ...sources,
+      idToken: 'id',
+      retryDelayMs: 0,
+      now: () => clock,
+      sleep: async (ms) => { clock += ms },
+      log: () => {}
+    })
+    assert.equal(state.state, 'tripped')
+    assert.equal(state.samples, 3)
+    // No summary also blinds the duplicate rule, whose positive control it is.
+    assert.match(state.verdict, /unverifiable: 3 samples in a row could not read duplicateSeat, ledgerDrift/)
+  } finally {
+    await rm(directory, { recursive: true, force: true })
+  }
+})
+
+test('a watch that ended without a verdict records failed, which the workflow flips back', async () => {
+  const directory = await mkdtemp(join(tmpdir(), 'relay-reserve-watch-'))
+  try {
+    const output = join(directory, 'out')
+    assert.equal((await recordOutcome(join(directory, 'missing.json'), output)).state, 'failed')
+    await writeFile(join(directory, 'watching.json'), JSON.stringify(freshState()))
+    assert.equal((await recordOutcome(join(directory, 'watching.json'), output)).state, 'failed')
+    await writeFile(join(directory, 'tripped.json'), JSON.stringify(freshState({ state: 'tripped', verdict: 'flip back x:\ny' })))
+    assert.equal((await recordOutcome(join(directory, 'tripped.json'), output)).state, 'tripped')
+    assert.match(await readFile(output, 'utf8'), /state=tripped\nverdict=flip back x: y\n$/)
+  } finally {
+    await rm(directory, { recursive: true, force: true })
+  }
+})
+
+// Google Storage and the director, as far as the flag tool's flip back reaches them. `onUpload`
+// returns 'conflict' to play another writer landing first, optionally with its own admitMode.
+function fakeFlipBack({ generation = '7', onUpload = () => 'ok', cellAnswers = true } = {}) {
+  const calls = []
+  let uploads = 0
+  let object = { generation: Number(generation), flags: { admitMode: 'reserve', readinessLocal: true } }
+  let applied = object
+  const fetchImpl = async (url, init = {}) => {
+    const text = String(url)
+    calls.push({ url: text, method: init.method ?? 'GET', body: init.body })
+    if (text.includes('/upload/storage/v1/')) {
+      uploads += 1
+      assert.match(text, new RegExp(`ifGenerationMatch=${object.generation}$`))
+      const outcome = onUpload(uploads)
+      if (outcome !== 'ok') {
+        object = { generation: object.generation + 1, flags: { ...object.flags, admitMode: outcome.admitMode ?? 'reserve' } }
+        applied = object
+        return new Response('', { status: 412 })
+      }
+      object = { generation: object.generation + 1, flags: JSON.parse(init.body.split('\r\n')[7]).flags }
+      applied = object
+      return new Response(JSON.stringify({ generation: String(object.generation) }))
+    }
+    if (text.includes('?alt=media')) {
+      return new Response(JSON.stringify({ v: 1, cellId: CELL, flags: object.flags }))
+    }
+    if (text.includes('/storage/v1/b/')) return new Response(JSON.stringify({ generation: String(object.generation) }))
+    if (text.endsWith('/v1/admin/runtime-status')) {
+      if (!cellAnswers) return new Response('', { status: 502 })
+      return new Response(
+        JSON.stringify({
+          role: 'cell',
+          cellId: CELL,
+          flagsApplied: applied,
+          supportedFlags: {
+            admitMode: { type: 'enum', values: ['db', 'reserve'] },
+            readinessLocal: { type: 'boolean' }
+          },
+          admitModeEffective: applied.flags.admitMode,
+          runtime: { controls: 10 }
+        })
+      )
+    }
+    if (text.endsWith('/v1/admin/cell-admit-mode')) {
+      const admitMode = JSON.parse(init.body).admitMode
+      return new Response(JSON.stringify({ v: 1, cellId: CELL, admitMode: admitMode ?? 'reserve', heartbeatFresh: true }))
+    }
+    throw new Error(`unexpected fetch ${text}`)
+  }
+  return { fetchImpl, calls, uploads: () => uploads }
+}
+
+const flipDependencies = (fetchImpl, lines, sleeps = []) => {
+  let clock = START
+  return {
+    fetchImpl,
+    accessToken: 'access',
+    idToken: 'id',
+    audit: { writtenBy: 'reserve-watch/test', runId: '1', runAttempt: '1' },
+    now: () => clock,
+    sleep: async (ms) => { sleeps.push(ms); clock += ms },
+    log: (line) => lines.push(JSON.parse(line))
+  }
+}
+
+const flipValues = (overrides = {}) => ({
+  mode: 'write',
+  'cell-id': CELL,
+  'cell-origin': 'https://c26.relay.onorca.dev',
+  'project-id': 'onorca-cloud',
+  'director-origin': 'https://relay.onorca.dev',
+  confirmation: `WRITE_RELAY_CELL_FLAGS ${CELL}`,
+  ...overrides
+})
+
+test('the flip back is the flag tool write: admitMode=db at the generation just read, then Postgres db', async () => {
+  const { fetchImpl, calls } = fakeFlipBack()
+  const lines = []
+  const result = await flipBack(flipValues(), flipDependencies(fetchImpl, lines))
+  assert.equal(result.written, true)
+  const upload = calls.find((call) => call.url.includes('/upload/storage/v1/'))
+  assert.match(upload.body, /"flags":\{"admitMode":"db","readinessLocal":true\}/)
+  assert.ok(calls.some((call) => call.url.endsWith('/v1/admin/cell-admit-mode') && JSON.parse(call.body).admitMode === 'db'))
+  assert.deepEqual(lines.at(-1), {
+    event: 'orca_relay_reserve_watch_flip_back',
+    mode: 'write',
+    cellId: CELL,
+    written: true,
+    generation: '8'
+  })
+})
+
+test('a dry-run flip back plans the write and writes nothing', async () => {
+  const { fetchImpl, calls } = fakeFlipBack()
+  const lines = []
+  const result = await flipBack(flipValues({ mode: 'dry-run', confirmation: '' }), {
+    fetchImpl,
+    accessToken: 'access',
+    idToken: 'id',
+    audit: {},
+    sleep: async () => {},
+    log: (line) => lines.push(JSON.parse(line))
+  })
+  assert.equal(result.written, false)
+  assert.ok(!calls.some((call) => call.url.includes('/upload/')))
+  assert.ok(!calls.some((call) => call.url.endsWith('/v1/admin/cell-admit-mode') && JSON.parse(call.body).admitMode))
+  assert.equal(lines.find((line) => line.event === 'orca_relay_cell_flags_plan').desired.flags.admitMode, 'db')
+})
+
+test('a write flip back without the typed confirmation is refused before any read', async () => {
+  const { fetchImpl, calls } = fakeFlipBack()
+  await assert.rejects(
+    flipBack(flipValues({ confirmation: '' }), { fetchImpl, accessToken: 'a', idToken: 'i', audit: {}, log: () => {} }),
+    /WRITE_RELAY_CELL_FLAGS production-gce-c26/
+  )
+  assert.equal(calls.length, 0)
+})
+
+test('a generation conflict is re-read and retried while the switch file still says reserve', async () => {
+  const { fetchImpl, uploads } = fakeFlipBack({ onUpload: (n) => (n < 3 ? { admitMode: 'reserve' } : 'ok') })
+  const lines = []
+  const sleeps = []
+  const result = await flipBack(flipValues(), flipDependencies(fetchImpl, lines, sleeps))
+  assert.equal(result.written, true)
+  assert.equal(uploads(), 3)
+  assert.deepEqual(lines.filter((line) => line.event === 'orca_relay_reserve_watch_flip_back_conflict').map((line) => line.attempt), [1, 2])
+  assert.deepEqual(sleeps.slice(0, 1), [2_000])
+  assert.equal(lines.at(-1).generation, '10')
+})
+
+test('a conflict whose writer already flipped the cell to db ends the flip back without a write', async () => {
+  const { fetchImpl, uploads } = fakeFlipBack({ onUpload: () => ({ admitMode: 'db' }) })
+  const lines = []
+  const result = await flipBack(flipValues(), flipDependencies(fetchImpl, lines))
+  assert.equal(uploads(), 1)
+  assert.equal(result.written, false)
+  assert.equal(result.alreadyDb, true)
+  assert.deepEqual(lines.at(-1), { event: 'orca_relay_reserve_watch_flip_back', mode: 'write', cellId: CELL, written: false, generation: '8', alreadyDb: true })
+})
+
+test('conflicts past the retry bound fail the flip back', async () => {
+  const { fetchImpl, uploads } = fakeFlipBack({ onUpload: () => ({ admitMode: 'reserve' }) })
+  await assert.rejects(flipBack(flipValues(), flipDependencies(fetchImpl, [])), /object changed since expected-generation/)
+  assert.equal(uploads(), FLIP_BACK_ATTEMPTS)
+})
+
+test('an unreachable cell fails with the exact break-glass inputs and writes nothing', async () => {
+  const { fetchImpl, uploads } = fakeFlipBack({ cellAnswers: false })
+  const lines = []
+  await assert.rejects(
+    flipBack(flipValues(), flipDependencies(fetchImpl, lines)),
+    (error) => {
+      assert.match(error.message, /production-gce-c26 does not answer runtime-status \(cell runtime status returned 502\)/)
+      assert.match(error.message, /set=admitMode=default, break-glass=BREAK GLASS production-gce-c26, expected-generation=7/)
+      return true
+    }
+  )
+  assert.equal(uploads(), 0)
+  assert.deepEqual(lines.at(-1).inputs, {
+    mode: 'write',
+    'cell-id': CELL,
+    set: 'admitMode=default',
+    'break-glass': `BREAK GLASS ${CELL}`,
+    'expected-generation': '7',
+    confirmation: `WRITE_RELAY_CELL_FLAGS ${CELL}`
+  })
+})
+
+test('the 503 background is the 10 minutes before flipped-at, read once they are published', async () => {
+  const directory = await mkdtemp(join(tmpdir(), 'relay-reserve-watch-'))
+  try {
+    let clock = START
+    // Flipped one minute ago: the minutes before it are not published until 3 minutes after it.
+    const flippedAt = START - MINUTE + 20_000
+    const config = parseAt(
+      watchValues({ 'state-file': join(directory, 'state.json'), 'segment-minutes': '3', 'flipped-at': new Date(flippedAt).toISOString() })
+    )
+    const sources = fakeSources({ per503: (minute) => (minute < START - MINUTE ? 4 : 40) })
+    const lines = []
+    const state = await watchSegment(config, {
+      ...sources,
+      idToken: 'id',
+      retryDelayMs: 0,
+      now: () => clock,
+      sleep: async (ms) => { clock += ms },
+      log: (line) => lines.push(JSON.parse(line))
+    })
+    // Pending until then, and pending is not a gap: the streak stays at zero.
+    assert.deepEqual(lines.map((line) => line.rules.director503.status), ['pending', 'pending', 'pass'])
+    assert.match(lines[0].rules.director503.reason, /background before the flip is not published yet/)
+    assert.deepEqual(lines.map((line) => line.unverifiableStreak), [0, 0, 0])
+    const baseline = sources.monitoringSpans.find((span) => span.endedAt === START - MINUTE)
+    assert.equal(baseline.startedAt, START - 11 * MINUTE)
+    assert.deepEqual(state.background, { medianPerMinute: 4, minutes: 10, source: 'measured' })
+    // An explicit value overrides the measurement.
+    const explicit = initialWatchState({ cellId: CELL, startedAt: START, durationMinutes: 60, background503PerMinute: 9, flippedAt })
+    assert.deepEqual(explicit.background, { medianPerMinute: 9, source: 'input' })
+  } finally {
+    await rm(directory, { recursive: true, force: true })
+  }
+})
+
+test('a quiet fleet with zero director 503s watches on and never trips', async () => {
+  const directory = await mkdtemp(join(tmpdir(), 'relay-reserve-watch-'))
+  try {
+    let clock = START
+    const config = parseAt(
+      watchValues({
+        'state-file': join(directory, 'state.json'),
+        'segment-minutes': '20',
+        'flipped-at': new Date(START - 2 * MINUTE).toISOString()
+      })
+    )
+    const lines = []
+    const state = await watchSegment(config, {
+      ...fakeSources({ per503: () => 0 }),
+      idToken: 'id',
+      retryDelayMs: 0,
+      now: () => clock,
+      sleep: async (ms) => { clock += ms },
+      log: (line) => lines.push(JSON.parse(line))
+    })
+    assert.equal(state.state, 'watching')
+    assert.equal(lines.length, 20)
+    assert.deepEqual(state.background, { medianPerMinute: 0, minutes: 10, source: 'measured' })
+    assert.ok(lines.slice(1).every((line) => line.rules.director503.status === 'pass'))
+    assert.ok(lines.every((line) => line.unverifiableStreak === 0))
+  } finally {
+    await rm(directory, { recursive: true, force: true })
+  }
+})
+
+test('no request series at all is unverifiable and trips at the limit', async () => {
+  const directory = await mkdtemp(join(tmpdir(), 'relay-reserve-watch-'))
+  try {
+    let clock = START
+    const config = parseAt(watchValues({ 'state-file': join(directory, 'state.json'), 'segment-minutes': '10', 'background-503-per-minute': '0', 'flipped-at': '' }))
+    const state = await watchSegment(config, {
+      ...fakeSources({ per503: () => 0, perOther: () => 0 }),
+      idToken: 'id',
+      retryDelayMs: 0,
+      now: () => clock,
+      sleep: async (ms) => { clock += ms },
+      log: () => {}
+    })
+    assert.equal(state.state, 'tripped')
+    assert.match(state.verdict, /unverifiable: 3 samples in a row could not read director503/)
+  } finally {
+    await rm(directory, { recursive: true, force: true })
+  }
+})
+
+test('a later segment judges flipped-at against the watch start, so a long watch does not age it out', async () => {
+  const directory = await mkdtemp(join(tmpdir(), 'relay-reserve-watch-'))
+  try {
+    const stateFile = join(directory, 'state.json')
+    const values = watchValues({ 'state-file': stateFile, 'flipped-at': new Date(START - 23 * 60 * MINUTE).toISOString() })
+    const later = START + 4 * 60 * MINUTE
+    await writeFile(stateFile, JSON.stringify(freshState()))
+    assert.equal((await parseWatchSegment(values, later)).flippedAt, START - 23 * 60 * MINUTE)
+    // The first segment (no state file yet) still refuses it, and records refused.
+    await rm(stateFile)
+    await assert.rejects(parseWatchSegment(values, later), /more than 24 hours/)
+    assert.equal(JSON.parse(await readFile(stateFile, 'utf8')).state, 'refused')
+  } finally {
+    await rm(directory, { recursive: true, force: true })
+  }
+})
+
+test('a corrupt state file fails the segment, never refuses or restarts it, and the outcome reads failed', async () => {
+  const directory = await mkdtemp(join(tmpdir(), 'relay-reserve-watch-'))
+  try {
+    const stateFile = join(directory, 'state.json')
+    const output = join(directory, 'out')
+    const values = watchValues({ 'state-file': stateFile })
+    for (const corrupt of ['null', '[]', '7', JSON.stringify({ state: 'watching', cellId: 'c7' })]) {
+      await writeFile(stateFile, corrupt)
+      await assert.rejects(parseWatchSegment(values, START), /state file/)
+      assert.equal(await readFile(stateFile, 'utf8'), corrupt)
+      assert.equal((await recordOutcome(stateFile, output)).state, 'failed')
+    }
+  } finally {
+    await rm(directory, { recursive: true, force: true })
+  }
+})
+
+test('a slow runtime-status reply is retried before the flip back calls for break glass', async () => {
+  const { fetchImpl: answering, uploads } = fakeFlipBack()
+  let runtimeCalls = 0
+  // fetchAdminOnceMore makes two tries per read; the first read's two both fail.
+  const fetchImpl = async (url, init) => {
+    if (String(url).endsWith('/v1/admin/runtime-status') && (runtimeCalls += 1) <= 2) return new Response('', { status: 503 })
+    return answering(url, init)
+  }
+  const lines = []
+  const sleeps = []
+  const result = await flipBack(flipValues(), flipDependencies(fetchImpl, lines, sleeps))
+  assert.equal(result.written, true)
+  assert.equal(uploads(), 1)
+  assert.deepEqual(lines.filter((line) => line.event === 'orca_relay_reserve_watch_runtime_retry').map((line) => line.attempt), [1])
+  assert.ok(!lines.some((line) => line.event === 'orca_relay_reserve_watch_break_glass_required'))
+  assert.ok(sleeps.includes(2_000))
+  assert.ok(FLIP_BACK_RUNTIME_ATTEMPTS > 1)
+})
+
+const watchWorkflow = readRelayWorkflow('operate-relay-production-reserve-watch.yml')
+const jobOf = (name) => {
+  const start = watchWorkflow.indexOf(`\n  ${name}:\n`)
+  assert.notEqual(start, -1, `job ${name} is missing`)
+  const next = watchWorkflow.slice(start + 1).search(/\n  [a-z-]+:\n/)
+  return watchWorkflow.slice(start, next === -1 ? undefined : start + 1 + next)
+}
+
+test('workflow: neither job takes the flag workflow group, which would let a newer run cancel a pending flip back', () => {
+  assert.doesNotMatch(watchWorkflow, /group: relay-production-cell-flags/)
+  assert.doesNotMatch(jobOf('flip-back'), /^\s+concurrency:/m)
+  assert.doesNotMatch(jobOf('watch'), /^\s+concurrency:/m)
+})
+
+test('workflow: a watch that failed or was cancelled without a verdict flips back too', () => {
+  const flip = jobOf('flip-back')
+  // A cancel re-evaluates each step: every flip-back step must run on always() too.
+  const steps = flip.slice(flip.indexOf('\n    steps:\n')).split(/\n      - /).slice(1)
+  assert.equal(steps.length, 7)
+  for (const step of steps) assert.match(step, /\n\s+if: \$\{\{ always\(\) \}\}/, step.split('\n')[0])
+  assert.match(flip, /if: \$\{\{ always\(\) && vars\.ORCA_CLOUD_OPERATIONS_ENABLED == 'true' && needs\.watch\.result != 'skipped'/)
+  assert.match(flip, /!contains\(fromJSON\('\["completed","stopped","refused"\]'\), needs\.watch\.outputs\.state\)/)
+  // Every segment checks its ID token's shape, and runs the one shared command.
+  const segments = watchWorkflow.match(/run: bash -c "\$\{WATCH_SEGMENT\}"/g) ?? []
+  assert.equal(segments.length, 6)
+  assert.match(jobOf('watch'), /process\.env\.ORCA_RELAY_ADMIN_ID_TOKEN/)
+})
+
+test('a live rollout lease refuses the watch before its first sample; an expired one does not', async () => {
+  const directory = await mkdtemp(join(tmpdir(), 'relay-reserve-watch-'))
+  try {
+    const stateFile = join(directory, 'state.json')
+    const output = join(directory, 'out')
+    const config = parseAt(watchValues({ 'state-file': stateFile, 'github-output': output, 'segment-minutes': '1' }))
+    const lease = (expiresAt) => ({ record: { expires_at: expiresAt }, holder: 'Deploy Relay Production Same-Cap (run 9)' })
+    let clock = START
+    const dependencies = (rolloutLease) => ({
+      ...fakeSources({ rolloutLease }),
+      idToken: 'id',
+      retryDelayMs: 0,
+      now: () => clock,
+      sleep: async (ms) => { clock += ms },
+      log: () => {}
+    })
+    const held = await watchSegment(config, dependencies(lease(START + MINUTE)))
+    assert.equal(held.state, 'refused')
+    assert.match(held.verdict, /a rollout holds the Cloud SQL rollout lease until .*Same-Cap \(run 9\)/)
+    assert.equal(JSON.parse(await readFile(stateFile, 'utf8')).state, 'refused')
+    assert.match(await readFile(output, 'utf8'), /^state=refused\n/)
+    await rm(stateFile)
+    assert.equal((await watchSegment(config, dependencies(lease(START - 1)))).state, 'watching')
+    // Only the first segment checks: a later one carries on from its state file.
+    assert.equal((await watchSegment(config, dependencies(lease(clock + MINUTE)))).state, 'watching')
+    // A lease it cannot read refuses too, as the lease action fails closed.
+    assert.match(await rolloutLeaseRefusal(async () => { throw new Error('403') }, START), /lease is unreadable \(403\)/)
+    assert.equal(await rolloutLeaseRefusal(async () => null, START), null)
+  } finally {
+    await rm(directory, { recursive: true, force: true })
+  }
+})
+
+const stepOf = (job, name) => {
+  const text = jobOf(job)
+  const start = text.indexOf(`- name: ${name}\n`)
+  assert.notEqual(start, -1, `step ${name} is missing`)
+  const end = text.indexOf('\n      - ', start + 1)
+  return text.slice(start, end === -1 ? undefined : end)
+}
+const ALARM_COMMAND = /node dev\/scripts\/watch-relay-reserve-cell\.mjs alarm --kind KIND \\\n\s+--mode "\$\{MODE\}" --cell-id "\$\{CELL_ID\}" --project-id "\$\{GCP_PROJECT_ID\}" --verdict "\$\{VERDICT_VAR\}"/
+const alarmCommand = (kind, verdict) => new RegExp(ALARM_COMMAND.source.replace('KIND', kind).replace('VERDICT_VAR', verdict))
+
+test('workflow: a flip back that did not land and a refused watch both raise the alarm', () => {
+  const alarm = stepOf('flip-back', 'Alarm when the flip back did not land')
+  assert.match(alarm, /if: \$\{\{ always\(\) \}\}/)
+  assert.match(alarm, /grep -qs '"event":"orca_relay_reserve_watch_flip_back"' "\$\{RUNNER_TEMP\}\/relay-reserve-flip-back\.jsonl"/)
+  assert.match(alarm, alarmCommand('flip-back-failed', 'VERDICT'))
+  const refused = stepOf('watch', 'Alarm on a refused watch')
+  assert.match(refused, /if: \$\{\{ always\(\) && steps\.outcome\.outputs\.state == 'refused' \}\}/)
+  assert.match(refused, /WATCH_VERDICT: \$\{\{ steps\.outcome\.outputs\.verdict \}\}/)
+  assert.match(refused, alarmCommand('refused', 'WATCH_VERDICT'))
+  // After the outcome step, so it reads the recorded state.
+  assert.ok(jobOf('watch').indexOf('- name: Alarm on a refused watch') > jobOf('watch').indexOf('- id: outcome'))
+  // The flip-back line names only the write's own event, never its conflict retries.
+  assert.match(readFileSync(new URL('./watch-relay-reserve-cell.mjs', import.meta.url), 'utf8'), /event: 'orca_relay_reserve_watch_flip_back',\n/)
+})
+
+test('an alarm is an escaped error annotation everywhere, and in write mode a paging log line', async () => {
+  const calls = []
+  const lines = []
+  const runGcloud = async (args, options) => (calls.push({ args, options }), { stdout: '' })
+  const alarm = (overrides) =>
+    raiseAlarm(
+      { mode: 'write', cellId: CELL, projectId: 'onorca-cloud', kind: 'refused', reason: '50%\r\nlease held', ...overrides },
+      { runGcloud, log: (line) => lines.push(line) }
+    )
+  assert.deepEqual(await alarm(), { paged: true })
+  assert.equal(
+    lines[0],
+    `::error title=Reserve watch refused for ${CELL}::${CELL} may be in reserve mode and unwatched: the watch was refused (50%25%0D%0Alease held). ` +
+      'Rerun it, or flip the cell to db with Operate Relay Production Cell Flags.'
+  )
+  assert.deepEqual(calls[0].args, [
+    'logging', 'write', ALARM_LOG_NAME,
+    JSON.stringify({ event: ALARM_EVENT, cellId: CELL, kind: 'refused', reason: '50%\r\nlease held' }),
+    '--payload-type=json', '--severity=ERROR', '--project', 'onorca-cloud'
+  ])
+  // A dry run never flips back, so its alarm annotates but does not page.
+  assert.deepEqual(await alarm({ mode: 'dry-run', kind: 'flip-back-failed' }), { paged: false })
+  assert.equal(calls.length, 1)
+  assert.match(lines[1], /^::error title=Reserve watch flip-back-failed for production-gce-c26::.*did not land/)
+  // A title property escapes its own separators too.
+  await alarm({ mode: 'dry-run', cellId: 'a:b,c' })
+  assert.match(lines[2], /^::error title=Reserve watch refused for a%3Ab%2Cc::/)
+  // A failed log write is named, never swallowed silently.
+  const failed = await raiseAlarm(
+    { mode: 'write', cellId: CELL, projectId: 'onorca-cloud', kind: 'flip-back-failed', reason: 'x' },
+    { runGcloud: async () => { throw new Error('PERMISSION_DENIED') }, log: (line) => lines.push(line) }
+  )
+  assert.deepEqual(failed, { paged: false })
+  assert.match(lines.at(-1), /^::error title=Reserve watch page failed::.*nothing pages: PERMISSION_DENIED/)
+  await assert.rejects(alarm({ kind: 'other' }), /--kind must be one of/)
+  // A missing project still annotates, then names the page it could not send.
+  const before = calls.length
+  assert.deepEqual(await alarm({ projectId: undefined }), { paged: false })
+  assert.equal(calls.length, before)
+  assert.match(lines.at(-2), /^::error title=Reserve watch refused for /)
+  assert.match(lines.at(-1), /nothing pages: --project-id is not a GCP project id/)
+})

cloud/dev/scripts/workload-identity-attribute-conditions.test.mjs

@@ -34,7 +34,7 @@ const EXPECTED_CONDITIONS = {
         "assertion.repository == 'stablyai/orca' && assertion.repository_id == '1183888342' && assertion.repository_owner_id == '127256420' && assertion.ref == 'refs/heads/main' && assertion.environment == 'production' && assertion.event_name == 'workflow_dispatch' && assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-push-deploy.yml@refs/heads/main' && assertion.job_workflow_ref == 'stablyai/orca/.github/workflows/cloud-push-deploy.yml@refs/heads/main'",
 
       github:
-        "assertion.repository == 'stablyai/orca' && assertion.repository_id == '1183888342' && assertion.repository_owner_id == '127256420' && assertion.ref == 'refs/heads/main' && assertion.environment == 'production' && ((assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-fence-broker.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production-capacity.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production-director.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production-multi-target.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-operate-relay-asia-admission.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-operate-relay-production-cell-flags.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-publish-relay-production.yml@refs/heads/main') || (assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-operate-relay-production-rehome.yml@refs/heads/main' && assertion.job_workflow_ref == 'stablyai/orca/.github/workflows/cloud-operate-relay-production-rehome-job.yml@refs/heads/main') || (assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production-same-cap.yml@refs/heads/main' && (assertion.job_workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production-same-cap-job.yml@refs/heads/main' || assertion.job_workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production-same-cap.yml@refs/heads/main')))",
+        "assertion.repository == 'stablyai/orca' && assertion.repository_id == '1183888342' && assertion.repository_owner_id == '127256420' && assertion.ref == 'refs/heads/main' && assertion.environment == 'production' && ((assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-fence-broker.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production-capacity.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production-director.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production-multi-target.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-operate-relay-asia-admission.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-operate-relay-production-cell-flags.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-operate-relay-production-reserve-watch.yml@refs/heads/main' || assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-publish-relay-production.yml@refs/heads/main') || (assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-operate-relay-production-rehome.yml@refs/heads/main' && assertion.job_workflow_ref == 'stablyai/orca/.github/workflows/cloud-operate-relay-production-rehome-job.yml@refs/heads/main') || (assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production-same-cap.yml@refs/heads/main' && (assertion.job_workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production-same-cap-job.yml@refs/heads/main' || assertion.job_workflow_ref == 'stablyai/orca/.github/workflows/cloud-deploy-relay-production-same-cap.yml@refs/heads/main')))",
       github_monitor:
         "assertion.repository == 'stablyai/orca' && assertion.repository_id == '1183888342' && assertion.repository_owner_id == '127256420' && assertion.ref == 'refs/heads/main' && assertion.environment == 'production' && assertion.workflow_ref == 'stablyai/orca/.github/workflows/cloud-monitor-relay-production.yml@refs/heads/main' && assertion.job_workflow_ref == 'stablyai/orca/.github/workflows/cloud-monitor-relay-production-job.yml@refs/heads/main'",
       github_fence:

cloud/docs/relay-workflows.md

@@ -109,7 +109,7 @@ identity bootstrap apply. The providers require their exact workflows on
 `refs/heads/main` with the `production` environment. Missing values fail
 closed; no dedicated operations identity falls back to the shared deploy identity.
 
-The shared production identity is restricted to seven named direct Relay callers plus the exact
+The shared production identity is restricted to nine named direct Relay callers plus the exact
 regional-rehome and same-cap reusable wrapper/job pairs on `main` in the `production` environment.
 Its Artifact Registry and Cloud Run mutation permissions are scoped to the Orca repository, Relay
 director, and Relay fence broker; it cannot mutate the API or auth services.
@@ -801,6 +801,76 @@ It ends with the single command that finishes from where it stopped. That comman
 state. Each run writes a timestamped log under `~/.orca/relay-director-deploy/`
 (`--log-directory` overrides it).
 
+## Reserve watch
+
+`Operate Relay Production Reserve Watch` watches one cell already switched to reserve mode and, on
+the first step 5 stop rule it trips, flips it back to db through the flag tool's own write
+(`operate-relay-cell-flags.mjs`: CAS, read-back, re-registration wait, Postgres record). It is
+manual only and inert until dispatched; `dry-run` logs the flag tool's plan instead of writing.
+`write` takes the flag workflow's confirmation, `WRITE_RELAY_CELL_FLAGS <cell-id>`.
+
+It samples once a minute for up to 300 minutes, re-authenticating every 55-minute segment because
+an ID token lasts an hour, and logs one JSON line per sample. The 503 background must predate the
+flip: give `flipped-at` (when the reserve write applied; the background is the median of the 10
+minutes before it) or `background-503-per-minute`, which overrides it. With neither, the watch is
+refused. A watch dispatched before the reserve write applied waits up to 10 minutes for the cell to
+report reserve, judging nothing meanwhile; a cell that never does ends the watch refused, without a
+write. It also refuses to start while any run holds the Cloud SQL rollout lease (same-cap and
+capacity rolls hold it for their whole run, and it names no cell), or when that lease cannot be
+read. It only reads the lease: holding it for hours would block every rollout. Pausing on a roll
+that starts mid-watch was not built, since a pause blinds the stop rules for the roll's length and
+a flip back mid-roll races its drain, so do not start a roll on a watched cell. It trips on, in
+this order:
+
+| Rule | Signal | Exact or proxy |
+|---|---|---|
+| Dead-man tripped | runtime-status `admitModeEffective=db` under a reserve switch file | Exact |
+| Overrun | runtime-status `enforcedConnectionUnits` over `connectionCapacity.hardCap` | Invariant check: the ledger already refuses past its cap, so this trips only if that refusal broke. It counts seats, bookings, in-flight and phone units |
+| 503s | Director `request_count` 503s above max(1.5x background, background + 20) for 3 complete minutes | Proxy that can miss: fleet-wide, since no signal attributes a 503 to a cell, so a problem on one cell can stay under the fleet's line |
+| Duplicate past grace | The ledger reconcile's `orca_relay_reserve_ledger_seat_behind_row` for the cell, reported again more than 90 s (60 s demotion close + 30 s map lag) after it was first seen, keyed by director instance, cell, epoch and row epoch | Proxy that can only over-count: the line names no host, so two hosts sharing that key read as one seat |
+| Host drop | runtime-status `controls` fell more than 5% (and at least 10 hosts) within 5 minutes with no `draining` sample | Exact |
+| Ledger drift | Queued plus dropped rows in each director's newest `orca_relay_reserve_ledger_summary`, over the cell's controls, above 0.5% for 2 minutes | Proxy that can only over-count: the summaries are fleet-wide, so exact only while this is the only reserve cell |
+
+Any rule it cannot read in a sample makes that sample unverifiable; `unverifiable-limit` such
+samples in a row is a trip of its own. The 503 read takes every response code in one query, and the
+newest minute any code has a point for is how far the series is published: a 503 minute at or
+before it with no point had no 503s, so a quiet fleet reads zero. Minutes past it are left out, not
+counted; fewer than 3 published minutes in the read, no request series at all, or an unparseable
+timestamp is unverifiable. This assumes the codes publish together; one code lagging the others
+delays a trip by its lag but cannot hide one, because a later read sees its points. Until the
+minutes before `flipped-at` are published (3 minutes past it) the 503 rule is pending, which does
+not add to the streak. The duplicate read also takes `orca_relay_reserve_ledger_summary`, which
+every placing director logs each minute, in the same query and window: with no summary from the
+last 150 s in it, the read is blind and the sample unverifiable, not zero duplicates. The duplicate
+rule keys a seat by director instance, so a director restart (a new instance id) starts its
+first-seen clock over. A cell whose switch file no longer says reserve ends the watch without a
+write.
+
+In `write` mode a watch that fails, times out or is cancelled before a verdict flips the cell back
+as a trip would; the flip-back job and each of its steps run on `always()`, so a cancel during the
+lease wait does not skip the switch-file write. That cancel can still kill the process before
+Postgres records db, leaving the file at db and Postgres at reserve; a force-cancel can stop it
+outright. The workflow's only group is
+its own per-cell one, which a running run never yields. It stays out of the flag workflow's group
+because GitHub keeps one pending run per group and a newer one cancels it, so a flip back queued
+there could be dropped by any later flag dispatch, and a watch held there would block manual writes
+for hours. The flag tool's generation check (CAS) orders it against other writes instead; on a
+conflict it re-reads and retries, up to 5 times, while the switch file still says reserve, and stops
+if it already says db. A cell that still does not answer runtime-status after 4 tries with backoff
+cannot take the ordinary flip back: the run fails and prints the exact break-glass inputs for
+`Operate Relay Production Cell Flags` (`set=admitMode=default`, `break-glass=BREAK GLASS <cell-id>`).
+
+An operator watches the run. A refused watch and a flip back that did not land (no ID token, a
+failed login, a cell that does not answer) can each leave the cell in reserve with nothing watching
+it, so both raise an `::error::` annotation naming the cell, as the clock-skew monitor does. In
+`write` mode both also log `orca_relay_reserve_watch_alarm` (log `orca-relay-reserve-watch`,
+resource `global`) with the deploy identity, and the `Orca Relay: reserve cell may be unwatched`
+alert pages on any such line. A failed login cannot page: the runner then has no credentials for
+`gcloud logging write`, so only the `::error::` and the operator watching the run report it. Apply
+the root Terraform that adds the workflow to the shared deploy identity's provider condition and
+grants that identity `roles/logging.logWriter` before the first `write` run; until then the
+workflow cannot log in and no alarm line can be written.
+
 ## Mobile push gateway
 
 `Deploy Push Gateway Production` (`.github/workflows/cloud-push-deploy.yml`) is the deploy path

cloud/infra/terraform/relay-github-actions.tf

@@ -20,6 +20,7 @@ locals {
     "deploy-relay-production.yml",
     "operate-relay-asia-admission.yml",
     "operate-relay-production-cell-flags.yml",
+    "operate-relay-production-reserve-watch.yml",
     "publish-relay-production.yml"
 
   ]
@@ -380,6 +381,16 @@ resource "google_project_iam_member" "github_logging_viewer" {
   member  = google_service_account.github_deploy[0].member
 }
 
+# The reserve watch logs its paging alarm line (orca_relay_reserve_watch_alarm) from the runner.
+# Accepted risk: Cloud Logging cannot narrow writes to one log, so this identity can write any line.
+resource "google_project_iam_member" "github_log_writer" {
+  count = local.relay_create_production_ops_identity ? 1 : 0
+
+  project = var.project_id
+  role    = "roles/logging.logWriter"
+  member  = google_service_account.github_deploy[0].member
+}
+
 resource "google_project_iam_member" "github_cloudsql_viewer" {
   count = local.relay_create_production_ops_identity ? 1 : 0
 

cloud/infra/terraform/relay-observability.tf

@@ -1748,6 +1748,74 @@ resource "google_monitoring_alert_policy" "relay_cell_reserve_dead_man_tripped"
   depends_on = [google_logging_metric.relay_cell_reserve_dead_man_tripped]
 }
 
+# The reserve watch workflow logs this from its runner (resource global) when a watch is refused or
+# a flip back did not land in write mode: either can leave a cell in reserve unwatched.
+resource "google_logging_metric" "relay_reserve_watch_alarm" {
+  project     = var.project_id
+  name        = "orca_relay_reserve_watch_alarm"
+  description = "Reserve watch runs that were refused or whose flip back to db did not land, so the cell may be in reserve mode with nothing watching it."
+  filter      = "resource.type=\"global\" AND logName=\"projects/${var.project_id}/logs/orca-relay-reserve-watch\" AND jsonPayload.event=\"orca_relay_reserve_watch_alarm\""
+  label_extractors = {
+    cell_id = "EXTRACT(jsonPayload.cellId)"
+    kind    = "EXTRACT(jsonPayload.kind)"
+  }
+
+  metric_descriptor {
+    metric_kind = "DELTA"
+    value_type  = "INT64"
+    unit        = "1"
+
+    labels {
+      key         = "cell_id"
+      value_type  = "STRING"
+      description = "Durable relay cell identifier."
+    }
+
+    labels {
+      key         = "kind"
+      value_type  = "STRING"
+      description = "refused or flip-back-failed."
+    }
+  }
+}
+
+resource "google_monitoring_alert_policy" "relay_reserve_watch_alarm" {
+  project               = var.project_id
+  display_name          = "Orca Relay: reserve cell may be unwatched"
+  combiner              = "OR"
+  enabled               = true
+  notification_channels = var.relay_alert_notification_channels
+
+  conditions {
+    display_name = "A reserve watch was refused or its flip back did not land"
+
+    condition_threshold {
+      filter          = "resource.type=\"global\" AND metric.type=\"logging.googleapis.com/user/orca_relay_reserve_watch_alarm\""
+      comparison      = "COMPARISON_GT"
+      threshold_value = 0
+      duration        = "0s"
+
+      aggregations {
+        alignment_period     = "300s"
+        per_series_aligner   = "ALIGN_SUM"
+        cross_series_reducer = "REDUCE_SUM"
+        group_by_fields      = ["metric.label.\"cell_id\""]
+      }
+
+      trigger {
+        count = 1
+      }
+    }
+  }
+
+  documentation {
+    content   = "A write-mode `Operate Relay Production Reserve Watch` run was refused (a rollout held the Cloud SQL rollout lease, the lease was unreadable, bad inputs, or the cell never reported reserve within 10 minutes) or its flip back to db did not land (no ID token, a failed login, or a cell that does not answer). The cell may be in reserve mode with nothing watching it. Read the cell's `runtime-status`; if its switch file says reserve, either rerun the watch or run `cloud-operate-relay-production-cell-flags.yml` with `admitMode=db` for that cell, using the break-glass inputs the failed run printed if the cell does not answer."
+    mime_type = "text/markdown"
+  }
+
+  depends_on = [google_logging_metric.relay_reserve_watch_alarm]
+}
+
 # host-session-registry.ts: a flip back closed a control it could not re-register (4409), after 20 refusals.
 resource "google_logging_metric" "relay_cell_reregistration_gave_up" {
   project     = var.project_id

cloud/package.json

@@ -17,7 +17,7 @@
     "load:relay:recovery-gate": "node dev/scripts/run-relay-recovery-wave-gate.mjs",
     "ops:relay": "pnpm --filter @orca-cloud/relay-ops dev",
     "pretest": "node --test dev/scripts/capture-terraform-plan-baseline.test.mjs dev/scripts/operate-relay-asia-admission.test.mjs dev/scripts/prepare-relay-asia-director-cells.test.mjs dev/scripts/prepare-relay-asia-topology-input.test.mjs dev/scripts/production-cloud-sql-rollout-lock.test.mjs dev/scripts/read-relay-serving-regional-placement-version.test.mjs dev/scripts/relay-asia-rollout-evidence.test.mjs dev/scripts/relay-asia-topology-workflow.test.mjs dev/scripts/relay-cloud-sql-connection-budget.test.mjs dev/scripts/relay-load-reader-evidence.test.mjs dev/scripts/relay-control-flag-alerts.test.mjs dev/scripts/relay-db-fault-alerts.test.mjs dev/scripts/relay-lock-contention-alerts.test.mjs dev/scripts/relay-reconnect-herd-alerts.test.mjs dev/scripts/relay-reserve-dead-man-alert.test.mjs dev/scripts/relay-region-hint-metrics.test.mjs dev/scripts/relay-staging-deploy-identity.test.mjs dev/scripts/sanitize-relay-asia-admission-result.test.mjs dev/scripts/terraform-root-partition.test.mjs dev/scripts/validate-relay-asia-topology-plan.test.mjs ../.github/actions/cloud-sql-rollout-lease/action-contract.test.mjs ../.github/actions/cloud-sql-rollout-lease/storage-lease.test.mjs && pnpm test:runner",
-    "test": "pnpm -r --filter \"!@orca-cloud/relay\" test && pnpm --filter @orca-cloud/relay test && node --test dev/scripts/check-relay-same-cap-headroom.test.mjs dev/scripts/classify-relay-production-capacity-director.test.mjs dev/scripts/classify-relay-staging-bootstrap.test.mjs dev/scripts/deploy-relay-blue-green.test.mjs dev/scripts/deploy-relay-gce-candidate.test.mjs dev/scripts/deploy-relay-gce-multi-target.test.mjs dev/scripts/drive-relay-director-deploy.test.mjs dev/scripts/github-smoke-token.test.mjs dev/scripts/infra.test.mjs dev/scripts/measure-relay-drain-disconnect-gap.test.mjs dev/scripts/operate-relay-cell-flags.test.mjs dev/scripts/operate-relay-regional-rehome.test.mjs dev/scripts/power-staging-relay.test.mjs dev/scripts/prepare-relay-capacity-canary.test.mjs dev/scripts/prepare-relay-production-capacity-canary.test.mjs dev/scripts/probe-relay-legacy-admission.test.mjs dev/scripts/probe-relay-rehome-trust.test.mjs dev/scripts/production-cell-image-digest-consistency.test.mjs dev/scripts/push-gateway-workflow.test.mjs dev/scripts/push-gateway-recovery.test.mjs dev/scripts/read-relay-production-capacity-identity.test.mjs dev/scripts/relay-admin-endpoint-retry-workflow.test.mjs dev/scripts/relay-admin-transient-retry.test.mjs dev/scripts/relay-admission-selector.test.mjs dev/scripts/relay-gce-terraform-fence.test.mjs dev/scripts/relay-load-connection-failure.test.mjs dev/scripts/relay-load-control-peer.test.mjs dev/scripts/relay-load-director-capacity-gate.test.mjs dev/scripts/relay-load-duplicate-assign.test.mjs dev/scripts/relay-load-model.test.mjs dev/scripts/relay-load-phase-barrier.test.mjs dev/scripts/relay-load-placement-boundary.test.mjs dev/scripts/relay-load-profile.test.mjs dev/scripts/relay-load-rebind-boundary.test.mjs dev/scripts/relay-load-region-behavior.test.mjs dev/scripts/relay-load-request-unit-boundary.test.mjs dev/scripts/relay-load-run-lifecycle.test.mjs dev/scripts/relay-load-splice-retry.test.mjs dev/scripts/relay-monitor-evidence.test.mjs dev/scripts/relay-production-capacity-wave.test.mjs dev/scripts/relay-production-capacity-workflow.test.mjs dev/scripts/relay-production-identity-boundaries.test.mjs dev/scripts/relay-production-same-cap-wave.test.mjs dev/scripts/relay-public-workflow-contract.test.mjs dev/scripts/relay-recovery-wave-gate.test.mjs dev/scripts/relay-region-observation-evidence.test.mjs dev/scripts/relay-rehome-aggregate-evidence.test.mjs dev/scripts/relay-repository.test.mjs dev/scripts/relay-same-cap-job-mode-conditions.test.mjs dev/scripts/relay-same-cap-script-census.test.mjs dev/scripts/relay-same-cap-shadow-gate.test.mjs dev/scripts/relay-staging-c4-refresh-workflow.test.mjs dev/scripts/relay-staging-capacity-identity.test.mjs dev/scripts/staging-relay-apply-guard.test.mjs dev/scripts/validate-relay-capacity-plan.test.mjs dev/scripts/verify-relay-capacity-transition.test.mjs dev/scripts/verify-relay-legacy-bootstrap.test.mjs dev/scripts/workload-identity-attribute-conditions.test.mjs",
+    "test": "pnpm -r --filter \"!@orca-cloud/relay\" test && pnpm --filter @orca-cloud/relay test && node --test dev/scripts/check-relay-same-cap-headroom.test.mjs dev/scripts/classify-relay-production-capacity-director.test.mjs dev/scripts/classify-relay-staging-bootstrap.test.mjs dev/scripts/deploy-relay-blue-green.test.mjs dev/scripts/deploy-relay-gce-candidate.test.mjs dev/scripts/deploy-relay-gce-multi-target.test.mjs dev/scripts/drive-relay-director-deploy.test.mjs dev/scripts/github-smoke-token.test.mjs dev/scripts/infra.test.mjs dev/scripts/measure-relay-drain-disconnect-gap.test.mjs dev/scripts/operate-relay-cell-flags.test.mjs dev/scripts/operate-relay-regional-rehome.test.mjs dev/scripts/power-staging-relay.test.mjs dev/scripts/prepare-relay-capacity-canary.test.mjs dev/scripts/prepare-relay-production-capacity-canary.test.mjs dev/scripts/probe-relay-legacy-admission.test.mjs dev/scripts/probe-relay-rehome-trust.test.mjs dev/scripts/production-cell-image-digest-consistency.test.mjs dev/scripts/push-gateway-workflow.test.mjs dev/scripts/push-gateway-recovery.test.mjs dev/scripts/read-relay-production-capacity-identity.test.mjs dev/scripts/relay-admin-endpoint-retry-workflow.test.mjs dev/scripts/relay-admin-transient-retry.test.mjs dev/scripts/relay-admission-selector.test.mjs dev/scripts/relay-gce-terraform-fence.test.mjs dev/scripts/relay-load-connection-failure.test.mjs dev/scripts/relay-load-control-peer.test.mjs dev/scripts/relay-load-director-capacity-gate.test.mjs dev/scripts/relay-load-duplicate-assign.test.mjs dev/scripts/relay-load-model.test.mjs dev/scripts/relay-load-phase-barrier.test.mjs dev/scripts/relay-load-placement-boundary.test.mjs dev/scripts/relay-load-profile.test.mjs dev/scripts/relay-load-rebind-boundary.test.mjs dev/scripts/relay-load-region-behavior.test.mjs dev/scripts/relay-load-request-unit-boundary.test.mjs dev/scripts/relay-load-run-lifecycle.test.mjs dev/scripts/relay-load-splice-retry.test.mjs dev/scripts/relay-monitor-evidence.test.mjs dev/scripts/relay-production-capacity-wave.test.mjs dev/scripts/relay-production-capacity-workflow.test.mjs dev/scripts/relay-production-identity-boundaries.test.mjs dev/scripts/relay-production-same-cap-wave.test.mjs dev/scripts/relay-public-workflow-contract.test.mjs dev/scripts/relay-recovery-wave-gate.test.mjs dev/scripts/relay-region-observation-evidence.test.mjs dev/scripts/relay-rehome-aggregate-evidence.test.mjs dev/scripts/relay-repository.test.mjs dev/scripts/relay-same-cap-job-mode-conditions.test.mjs dev/scripts/relay-same-cap-script-census.test.mjs dev/scripts/relay-same-cap-shadow-gate.test.mjs dev/scripts/relay-staging-c4-refresh-workflow.test.mjs dev/scripts/relay-staging-capacity-identity.test.mjs dev/scripts/staging-relay-apply-guard.test.mjs dev/scripts/validate-relay-capacity-plan.test.mjs dev/scripts/verify-relay-capacity-transition.test.mjs dev/scripts/verify-relay-legacy-bootstrap.test.mjs dev/scripts/watch-relay-reserve-cell.test.mjs dev/scripts/workload-identity-attribute-conditions.test.mjs",
     "typecheck": "pnpm -r typecheck",
     "test:runner": "node --import tsx --conditions=orca-source --test dev/scripts/run-relay-tests.test.mjs"
   },