Reduce alert checkpoint churn and clarify ZFS monitoring (#2356)

## Changes and why
- Bound asynchronous active-alert checkpoints to one worker and one
requested follow-up during a burst. Current-state snapshots, immediate
lifecycle durability, failure recovery and the final shutdown save
remain intact. A blocked-writer fixture reproduced 512 workers for 512
requests before the repair; this removes that amplification without
claiming to resolve the installed CPU/write symptoms in #1966, #2112 or
#2113.
- Correct both ZFS help copies after discussion #2354 exposed a promise
of discovering every local pool. Explain the configured-Proxmox-storage
boundary, the host agent's mounted capacity view and the separate
unregistered-pool health-alert gap. The read-only guidance does not ask
users to alter their storage configuration.
- Permit a bounded, exact-source release-path watchdog check on demand,
retaining the scheduled no-publication/no-mutation envelope, rejecting
candidate/exception overrides and skipping candidate builds. This
addresses missing timely coverage after a prior watchdog ordering
failure; it does not publish a release.

## Included source
Complete range
`cfaad6b46d6c47ef9aed400cb57e1c5b54bd02f2..6fa78a01f813f373fda64922533f060963ff32fc`:
- Alert checkpoint candidate `5dea312f2b17323a25e616b81324855b4ca939da`.
- ZFS guidance candidate `cf8a3f9a6cca47509efc10df62ba24bb00f9aa09`.
- Watchdog candidate `40037f003581f8f553bb41de313c927256a0d3c2` through
`2043f3dd767d3f0970e4008a4d9067c5f7327c32`.
The additive integrations preserve every reviewed commit and include
upstream History publication `cfaad6b46d6c47ef9aed400cb57e1c5b54bd02f2`
without rewriting it.

## Validation and limits
- `source-gdu65a_x`: 25 focused race repetitions, complete
alerts/config/eventlog/reducer/specs race suites and vet pass.
`source-ezs4tqqg` retains the original blocked-worker reproduction.
- `source-_7_yx4oh`: 112 release-control Python tests and selected Go
install checks pass, including actual workflow
admission/source-selection checks.
- `source-y0mu1jos` observes the existing ZFS limits and configured
controls; `source-s2wg_5ix` passes 26 docs/card tests.
- Publication provenance and complete-range whitespace checks pass.
Integrated executable/test files match the validated candidates; the
only later watchdog contract-doc change is an already-published updater
section. The two help copies match.

Sufficient candidate evidence is reused, not rerun merely for batching.
Exact-head CI and independent review still govern landing. These checks
do not establish installed alert/write recovery, complete PBS
acceptance, restored operational watchdog coverage or release
qualification.


[How Pulse handles
triage](https://github.com/rcourtman/Pulse/blob/main/docs/AI_TRANSPARENCY.md)

.github/workflows/release-dry-run.yml

@@ -1,5 +1,5 @@
 name: Release Dry Run
-run-name: Release Dry Run v${{ inputs.version || 'scheduled' }}
+run-name: ${{ inputs.watchdog && format('Release Watchdog at {0}', inputs.expected_workflow_sha) || format('Release Dry Run v{0}', inputs.version || 'scheduled') }}
 
 on:
   # Weekly drift watchdog: starts on main every Tuesday 07:00 UTC and selects the governed
@@ -7,12 +7,21 @@ on:
   # week at a time instead of piling up until an RC publish. Scheduled runs
   # carry no workflow_dispatch inputs (GitHub does not apply input defaults to
   # schedule events), so the rehearsal step derives the rollback target as the
-  # latest stable tag preceding VERSION. Manual dispatches must still supply
-  # rollback_version explicitly.
+  # latest stable tag preceding VERSION. The fixed on-demand watchdog uses the
+  # same envelope; ordinary manual rehearsals still supply rollback explicitly.
   schedule:
     - cron: '0 7 * * 2'
   workflow_dispatch:
     inputs:
+      watchdog:
+        description: 'Run the scheduled no-publication watchdog now, from exact reviewed main'
+        required: false
+        type: boolean
+        default: false
+      expected_workflow_sha:
+        description: 'Required exact main workflow SHA for watchdog dispatch; forbidden for candidate rehearsals'
+        required: false
+        type: string
       version:
         description: 'Optional version under rehearsal (e.g. 6.0.0-rc.2 or 6.0.0)'
         required: false
@@ -23,7 +32,9 @@ on:
         type: string
       rollback_version:
         description: 'Required rollback stable version to rehearse (for example 6.0.4 or v6.0.4)'
-        required: true
+        # Candidate rehearsals still enforce this before checkout. Only the
+        # fixed watchdog envelope derives the preceding stable rollback.
+        required: false
         type: string
       ga_date:
         description: 'Stable v6.0.0 rehearsal only: planned GA publish date (YYYY-MM-DD)'
@@ -70,7 +81,7 @@ permissions:
 jobs:
   build_release_candidate:
     name: Build Immutable Release Candidate
-    if: ${{ inputs.version != '' }}
+    if: ${{ inputs.watchdog != true && inputs.version != '' }}
     permissions:
       actions: write
       attestations: write
@@ -115,11 +126,67 @@ jobs:
 
     steps:
       - name: Validate release ref
+        id: mode
+        env:
+          EVENT_NAME: ${{ github.event_name }}
+          WATCHDOG_INPUT: ${{ inputs.watchdog }}
+          EXPECTED_WORKFLOW_SHA_INPUT: ${{ inputs.expected_workflow_sha }}
+          VERSION_INPUT: ${{ inputs.version }}
+          PROMOTED_FROM_TAG_INPUT: ${{ inputs.promoted_from_tag }}
+          ROLLBACK_VERSION_INPUT: ${{ inputs.rollback_version }}
+          GA_DATE_INPUT: ${{ inputs.ga_date }}
+          V5_EOS_DATE_INPUT: ${{ inputs.v5_eos_date }}
+          HOTFIX_EXCEPTION_INPUT: ${{ inputs.hotfix_exception }}
+          HOTFIX_REASON_INPUT: ${{ inputs.hotfix_reason }}
+          UNSIGNED_WINDOWS_EXCEPTION_INPUT: ${{ inputs.unsigned_windows_exception }}
+          UNSIGNED_WINDOWS_REASON_INPUT: ${{ inputs.unsigned_windows_reason }}
+          NOTE_INPUT: ${{ inputs.note }}
+          MOBILE_RELEASE_DECISION: ${{ inputs.mobile_release_decision }}
+          MOBILE_RELEASE_EVIDENCE: ${{ inputs.mobile_release_evidence }}
         run: |
+          set -euo pipefail
           if [[ "${GITHUB_REF}" != refs/heads/* ]]; then
             echo "::error::Release dry run must be executed from a branch ref. Current ref: ${GITHUB_REF}"
             exit 1
           fi
+          WATCHDOG=false
+          case "${EVENT_NAME}:${WATCHDOG_INPUT:-false}" in
+            schedule:false) WATCHDOG=true ;;
+            workflow_dispatch:true)
+              if [ "${GITHUB_REF}" != "refs/heads/main" ] || \
+                 [[ ! "${EXPECTED_WORKFLOW_SHA_INPUT:-}" =~ ^[0-9a-f]{40}$ ]] || \
+                 [ "${EXPECTED_WORKFLOW_SHA_INPUT}" != "${GITHUB_SHA}" ]; then
+                echo "::error::Watchdog dispatch requires the exact reviewed main workflow SHA."
+                exit 1
+              fi
+              WATCHDOG=true
+              ;;
+            workflow_dispatch:false)
+              if [ -n "${EXPECTED_WORKFLOW_SHA_INPUT:-}" ] || [ -z "${ROLLBACK_VERSION_INPUT:-}" ]; then
+                echo "::error::Candidate rehearsal requires explicit rollback and no watchdog SHA."
+                exit 1
+              fi
+              ;;
+            *) echo "::error::Unsupported rehearsal event or watchdog mode."; exit 1 ;;
+          esac
+          if [ "${WATCHDOG}" = "true" ]; then
+            for name in VERSION_INPUT PROMOTED_FROM_TAG_INPUT ROLLBACK_VERSION_INPUT \
+                        GA_DATE_INPUT V5_EOS_DATE_INPUT HOTFIX_REASON_INPUT \
+                        UNSIGNED_WINDOWS_REASON_INPUT NOTE_INPUT \
+                        MOBILE_RELEASE_DECISION MOBILE_RELEASE_EVIDENCE; do
+              if [ -n "${!name:-}" ]; then
+                echo "::error::Watchdog refuses candidate input ${name}."
+                exit 1
+              fi
+            done
+            for name in HOTFIX_EXCEPTION_INPUT UNSIGNED_WINDOWS_EXCEPTION_INPUT; do
+              if [ "${!name:-false}" != "false" ]; then
+                echo "::error::Watchdog refuses exception input ${name}."
+                exit 1
+              fi
+            done
+          fi
+          echo "watchdog=${WATCHDOG}" >> "$GITHUB_OUTPUT"
           echo "[OK] Dry run executing on branch ${GITHUB_REF_NAME}"
 
       - name: Checkout repository
@@ -145,11 +212,16 @@ jobs:
         id: source
         env:
           EVENT_NAME: ${{ github.event_name }}
+          WATCHDOG_MODE: ${{ steps.mode.outputs.watchdog }}
           REQUIRED_BRANCH: ${{ steps.branch_policy.outputs.required_branch }}
         run: |
           set -euo pipefail
           TESTED_BRANCH="${GITHUB_REF_NAME}"
-          if [ "${EVENT_NAME}" = "schedule" ]; then
+          if [ "${WATCHDOG_MODE}" = "true" ]; then
+            if [ "${EVENT_NAME}" != "schedule" ] && [ "${EVENT_NAME}" != "workflow_dispatch" ]; then
+              echo "::error::Unsupported watchdog event."
+              exit 1
+            fi
             # Resolve once; never test a moving branch or rewrite GITHUB_SHA.
             git check-ref-format "refs/heads/${REQUIRED_BRANCH}"
             git fetch --no-tags origin "refs/heads/${REQUIRED_BRANCH}"
@@ -183,6 +255,7 @@ jobs:
         id: rehearsal
         env:
           EVENT_NAME: ${{ github.event_name }}
+          WATCHDOG_MODE: ${{ steps.mode.outputs.watchdog }}
           VERSION_INPUT: ${{ inputs.version }}
           PROMOTED_FROM_TAG_INPUT: ${{ inputs.promoted_from_tag }}
           ROLLBACK_VERSION_INPUT: ${{ inputs.rollback_version }}
@@ -242,10 +315,9 @@ jobs:
           elif [ -n "${UNSIGNED_WINDOWS_REASON_INPUT:-}" ]; then
             HELPER_ARGS+=(--unsigned-windows-reason "${UNSIGNED_WINDOWS_REASON_INPUT}")
           fi
-          if [ "${EVENT_NAME}" = "schedule" ] && [ -z "${ROLLBACK_VERSION_INPUT:-}" ]; then
-            # Scheduled watchdog runs carry no dispatch inputs; derive the
-            # rollback target instead of failing. Manual dispatches never get
-            # this flag, so their explicit-rollback requirement stands.
+          if [ "${WATCHDOG_MODE}" = "true" ] && [ -z "${ROLLBACK_VERSION_INPUT:-}" ]; then
+            # Both admitted watchdog paths have no candidate inputs. Ordinary
+            # candidate rehearsals retain their explicit-rollback requirement.
             HELPER_ARGS+=(--derive-rollback-latest-stable)
             echo "[OK] Scheduled rehearsal: deriving rollback target from the latest preceding stable tag"
           fi
@@ -266,6 +338,7 @@ jobs:
         id: mobile_release
         env:
           EVENT_NAME: ${{ github.event_name }}
+          WATCHDOG_MODE: ${{ steps.mode.outputs.watchdog }}
           MOBILE_RELEASE_DECISION: ${{ inputs.mobile_release_decision }}
           MOBILE_RELEASE_EVIDENCE: ${{ inputs.mobile_release_evidence }}
           WORKFLOW_OUTPUT_1: ${{ steps.rehearsal.outputs.version }}
@@ -273,7 +346,7 @@ jobs:
           set -euo pipefail
           DECISION="${MOBILE_RELEASE_DECISION:-}"
           EVIDENCE="${MOBILE_RELEASE_EVIDENCE:-}"
-          if [ -z "${DECISION}" ] && [ "${EVENT_NAME}" = "schedule" ]; then
+          if [ -z "${DECISION}" ] && [ "${WATCHDOG_MODE}" = "true" ]; then
             DECISION="no-mobile-impact"
             EVIDENCE="Scheduled release dry-run watchdog; no mobile release packet is being dispatched."
           fi

docs/ZFS_MONITORING.md

@@ -1,17 +1,50 @@
 # 💾 ZFS Pool Monitoring
 
-Pulse automatically detects and monitors ZFS pools on your Proxmox nodes.
+Pulse adds ZFS pool health to **configured Proxmox storage** on your nodes.
+The Proxmox → Storage list is not an inventory of every local ZFS pool.
 
 > **TrueNAS users:** TrueNAS ZFS pool monitoring is handled separately via the TrueNAS integration. See [CONFIGURATION.md](CONFIGURATION.md#truenas) for setup. This page covers Proxmox-native ZFS pools.
 
 ## 🚀 Features
 
-*   **Auto-Detection**: No configuration needed.
+*   **Auto-Detection**: Matches configured storage to its backing ZFS pool, including directory storage on a ZFS dataset.
 *   **Health Status**: Tracks `ONLINE`, `DEGRADED`, and `FAULTED` states.
 *   **Error Tracking**: Monitors read, write, and checksum errors.
 *   **Dataset Inventory**: With a Unified Agent on the node, expanded pool details list ZFS filesystems and zvols with used, available, referenced, and mountpoint information.
 *   **Alerts**: Notifies you of degraded pools or failing devices.
 
+## Unregistered and data-only pools
+
+Pulse reads `/nodes/{node}/disks/zfs` and the pool-detail endpoint to enrich
+Proxmox storage rows. A pool that is not configured in `/etc/pve/storage.cfg`
+does not get its own row, even if it is `ONLINE` in the ZFS API, its physical
+disks are visible, and the API account has the required permissions. A pool
+used only through datasets and LXC bind mounts can therefore be absent from
+Proxmox → Storage without a collection error.
+
+Do not register a pool as Proxmox storage or change its datasets solely to make
+it appear in Pulse. For a pool you intentionally keep unregistered:
+
+* A **Unified Agent on the Proxmox host**, with a locally mounted ZFS dataset
+  and access to `zpool`, can report capacity and usage in the node's
+  **Overview details → Disks** panel, under the selected mount point. Without
+  usable `zpool` output, the agent falls back to mounted-dataset counters;
+  those are not necessarily whole-pool capacity.
+* That capacity view and physical-disk SMART data do **not** provide
+  `DEGRADED`/`FAULTED` pool-health alerts for an unregistered pool with no
+  matching Proxmox storage row. Keep a separate ZFS health check for it.
+* On the Proxmox host, these read-only commands show pool capacity, health and
+  device errors. Replace `tank` with your pool name:
+
+  ```bash
+  zpool list -o name,size,alloc,free,health tank
+  zpool status tank
+  ```
+
+`zpool list` reports raw pool capacity; it can differ from the agent's usable
+capacity estimate on RAIDZ. These commands do not change the pool or register
+it as Proxmox storage.
+
 ## ⚙️ Requirements
 
 The Pulse user needs `Sys.Audit` permission on `/nodes/{node}/disks` (included in the standard Pulse role).
@@ -42,6 +75,11 @@ PULSE_DISABLE_ZFS_MONITORING=true
 ## 🔍 Troubleshooting
 
 **No ZFS Data?**
-1.  Check permissions: `pveum user permissions pulse-monitor@pve`.
+1.  Check whether the pool backs configured Proxmox storage: `pvesm status`.
+    If it is intentionally unregistered, use the capacity and health paths above.
 2.  Verify pools exist: `zpool list`.
-3.  Check logs: `journalctl -u pulse -n 200 | grep -i zfs`.
+3.  For a missing configured storage row, check permissions:
+    `pveum user permissions pulse-monitor@pve`.
+4.  Check logs: `journalctl -u pulse -n 200 | grep -i zfs`.
+    A missing unregistered-pool row alone is not evidence that more API
+    permissions are needed.

docs/release-control/v6/internal/subsystems/alerts.md

@@ -3044,6 +3044,34 @@ starting database workers, including restart loading and Unix hardening. The
 active mirror's unchanged fast path alone does not cover this companion file.
 This is a narrow write contributor, not aggregate installed-write acceptance.
 
+### Coalesced asynchronous active-alert checkpoints
+
+`active_persistence.go` admits one asynchronous checkpoint worker per Manager,
+not one goroutine and full serialization per mutation. Admission is protected
+by the shutdown lock. A worker starts immediately, takes fresh state through
+`SaveActiveAlerts`, and retains one follow-up request for mutations arriving
+during its current pass. Returning to idle and accepting another request share
+the same lock, so requests cannot be stranded at the idle boundary. The worker
+never holds the shutdown lock while acquiring persistence or manager-state
+locks; callers may request a checkpoint while holding the manager-state lock.
+
+This coalesces derived snapshots, not lifecycle events or notification work.
+Synchronous lifecycle durability, checkpoint revision/failure-epoch checks,
+recovery-mirror replacement, intent timing and periodic saves are unchanged.
+An error or recovered panic releases admission after any requested follow-up;
+there is no automatic busy-loop retry. Once shutdown closes admission, the
+worker finishes only its in-flight pass; `Stop` joins it and writes the final
+current snapshot synchronously instead of draining redundant queued snapshots.
+
+`TestAsyncActiveCheckpointBurstIsBounded` in `internal/alerts/alerts_test.go`
+blocks the real checkpoint writer, bounds a 512-request burst to one worker,
+then verifies current acknowledgement, incident age, metadata and pending intent
+through actual JSON persistence and a fresh Manager. The scheduler regressions
+in `internal/alerts/async_checkpoint_test.go` pin one follow-up for 10,000
+in-flight requests, idle restart, error/panic recovery, retry after a real file
+failure, and shutdown admission/draining. These checks establish bounded
+checkpoint work, not the installed CPU/write rate or any reporter's recovery.
+
 ### Warning-level destination preference persistence
 
 Email API decoding and encoding, plus webhook edit/create and list presentation,

docs/release-control/v6/internal/subsystems/deployment-installability.md

@@ -131,11 +131,22 @@ product checks use the selected checkout; a missing branch or policy mismatch
 fails closed rather than falling back to main.
 
 Manual dispatch retains its governed event branch and exact event SHA, VERSION
-equality and explicit rollback requirement. Only scheduled runs may derive the
-preceding stable rollback target. Summaries distinguish workflow event revision
+equality and explicit rollback requirement. The one additional dispatch mode
+is a fixed no-publication watchdog: `watchdog=true` plus the exact reviewed
+main `expected_workflow_sha`, with no candidate, exception, mobile or note
+inputs. It checks the main ref and event SHA before checkout, then uses the
+same one-fetch governed-source selection and preceding stable rollback
+derivation as the schedule. A moving control head fails closed. Ordinary
+candidate dispatches cannot use the watchdog SHA or omit explicit rollback.
+Summaries distinguish workflow event revision
 from tested branch and revision; unresolved selection is not test evidence.
 This watchdog is not an admitted release candidate and cannot qualify or
-replace the fixed release packet.
+replace the fixed release packet. Its distinct `Release Watchdog at <sha>`
+title cannot satisfy an exact-version candidate rehearsal. The candidate-build
+job is always skipped in watchdog mode; actual preflight, demo resolver,
+no-mutation verification and definitive verdict must all succeed. Delivery's
+reviewed dispatcher fixes repository, workflow, main and both inputs, retains
+intent before POST and never replays a refused or uncertain watchdog attempt.
 
 Verification: `rehearsal_source_test.py` executes the source-selection shell
 against local Git fixtures. `release_promotion_policy_test.py` pins workflow
@@ -3869,13 +3880,14 @@ for stable-versus-prerelease metadata validation shared by `.github/workflows/re
 and `.github/workflows/create-release.yml`. Promotion rollback targets, promoted
 prerelease lineage, soak checks, and GA/v5 notice metadata may not drift between those
 two workflows through duplicated inline shell validation.
-One scoped exception keeps the weekly drift watchdog viable: scheduled
+One scoped exception keeps the drift watchdog viable: scheduled
 `release-dry-run.yml` runs carry no `workflow_dispatch` inputs (GitHub does
 not apply input defaults to `schedule` events), so the rehearsal step passes
 `--derive-rollback-latest-stable` and the resolver fills the empty
 `rollback_version` with the latest stable repository tag preceding the
-rehearsal version. The derivation flag is gated on the `schedule` event in
-the workflow; manual rehearsal dispatches and real promotions must still
+rehearsal version. The derivation flag is gated on admitted watchdog mode
+(schedule or the exact-main, candidate-input-free dispatch envelope described
+above); ordinary candidate rehearsal dispatches and real promotions must still
 supply an explicit stable `rollback_version`, and the resolver still fails
 closed when the input is empty and the flag is absent.
 `scripts/release_control/validate_artifact_release_line.py` is the canonical

frontend-modern/public/docs/ZFS_MONITORING.md

@@ -1,17 +1,50 @@
 # 💾 ZFS Pool Monitoring
 
-Pulse automatically detects and monitors ZFS pools on your Proxmox nodes.
+Pulse adds ZFS pool health to **configured Proxmox storage** on your nodes.
+The Proxmox → Storage list is not an inventory of every local ZFS pool.
 
 > **TrueNAS users:** TrueNAS ZFS pool monitoring is handled separately via the TrueNAS integration. See [CONFIGURATION.md](CONFIGURATION.md#truenas) for setup. This page covers Proxmox-native ZFS pools.
 
 ## 🚀 Features
 
-*   **Auto-Detection**: No configuration needed.
+*   **Auto-Detection**: Matches configured storage to its backing ZFS pool, including directory storage on a ZFS dataset.
 *   **Health Status**: Tracks `ONLINE`, `DEGRADED`, and `FAULTED` states.
 *   **Error Tracking**: Monitors read, write, and checksum errors.
 *   **Dataset Inventory**: With a Unified Agent on the node, expanded pool details list ZFS filesystems and zvols with used, available, referenced, and mountpoint information.
 *   **Alerts**: Notifies you of degraded pools or failing devices.
 
+## Unregistered and data-only pools
+
+Pulse reads `/nodes/{node}/disks/zfs` and the pool-detail endpoint to enrich
+Proxmox storage rows. A pool that is not configured in `/etc/pve/storage.cfg`
+does not get its own row, even if it is `ONLINE` in the ZFS API, its physical
+disks are visible, and the API account has the required permissions. A pool
+used only through datasets and LXC bind mounts can therefore be absent from
+Proxmox → Storage without a collection error.
+
+Do not register a pool as Proxmox storage or change its datasets solely to make
+it appear in Pulse. For a pool you intentionally keep unregistered:
+
+* A **Unified Agent on the Proxmox host**, with a locally mounted ZFS dataset
+  and access to `zpool`, can report capacity and usage in the node's
+  **Overview details → Disks** panel, under the selected mount point. Without
+  usable `zpool` output, the agent falls back to mounted-dataset counters;
+  those are not necessarily whole-pool capacity.
+* That capacity view and physical-disk SMART data do **not** provide
+  `DEGRADED`/`FAULTED` pool-health alerts for an unregistered pool with no
+  matching Proxmox storage row. Keep a separate ZFS health check for it.
+* On the Proxmox host, these read-only commands show pool capacity, health and
+  device errors. Replace `tank` with your pool name:
+
+  ```bash
+  zpool list -o name,size,alloc,free,health tank
+  zpool status tank
+  ```
+
+`zpool list` reports raw pool capacity; it can differ from the agent's usable
+capacity estimate on RAIDZ. These commands do not change the pool or register
+it as Proxmox storage.
+
 ## ⚙️ Requirements
 
 The Pulse user needs `Sys.Audit` permission on `/nodes/{node}/disks` (included in the standard Pulse role).
@@ -42,6 +75,11 @@ PULSE_DISABLE_ZFS_MONITORING=true
 ## 🔍 Troubleshooting
 
 **No ZFS Data?**
-1.  Check permissions: `pveum user permissions pulse-monitor@pve`.
+1.  Check whether the pool backs configured Proxmox storage: `pvesm status`.
+    If it is intentionally unregistered, use the capacity and health paths above.
 2.  Verify pools exist: `zpool list`.
-3.  Check logs: `journalctl -u pulse -n 200 | grep -i zfs`.
+3.  For a missing configured storage row, check permissions:
+    `pveum user permissions pulse-monitor@pve`.
+4.  Check logs: `journalctl -u pulse -n 200 | grep -i zfs`.
+    A missing unregistered-pool row alone is not evidence that more API
+    permissions are needed.

internal/alerts/active_persistence.go

@@ -20,31 +20,71 @@ import (
 )
 
 func (m *Manager) saveActiveAlertsAsync(context string) {
-	m.stopMu.RLock()
-	if m.stopping {
-		m.stopMu.RUnlock()
-		return
+	if m.queueActiveAlertsSave(context) {
+		go m.runActiveAlertsSaveWorker(m.SaveActiveAlerts)
 	}
-	m.workerWG.Add(1)
-	m.stopMu.RUnlock()
+}
 
-	go func() {
-		defer m.workerWG.Done()
-		defer func() {
-			if r := recover(); r != nil {
+// queueActiveAlertsSave reserves at most one worker and one follow-up snapshot.
+// Callers may hold m.mu; never wait for a checkpoint under stopMu.
+func (m *Manager) queueActiveAlertsSave(context string) bool {
+	m.stopMu.Lock()
+	defer m.stopMu.Unlock()
+	if m.stopping {
+		return false
+	}
+	m.activeSavePending = true
+	m.activeSaveContext = context
+	if m.activeSaveRunning {
+		return false
+	}
+	m.activeSaveRunning = true
+	// Add under the shutdown admission lock, before Stop can begin waiting.
+	m.workerWG.Add(1)
+	return true
+}
+
+// Each pass takes a fresh snapshot. Requests arriving during a pass mark
+// one follow-up, not another goroutine/full snapshot per mutation. No timer
+// delays alert persistence, and lifecycle events remain independently durable.
+func (m *Manager) runActiveAlertsSaveWorker(save func() error) {
+	defer m.workerWG.Done()
+	for {
+		m.stopMu.Lock()
+		if m.stopping || !m.activeSavePending {
+			m.activeSaveRunning = false
+			m.activeSavePending = false
+			m.activeSaveContext = ""
+			m.stopMu.Unlock()
+			// Stop drains this in-flight worker and owns the final synchronous
+			// checkpoint; do not drain redundant queued snapshots at shutdown.
+			return
+		}
+		context := m.activeSaveContext
+		m.activeSavePending = false
+		m.activeSaveContext = ""
+		m.stopMu.Unlock()
+
+		// Recover per pass so a failed checkpoint cannot strand later requests
+		// or leave worker admission permanently occupied. Retry only on another
+		// request (or the existing periodic/shutdown save), never in a busy loop.
+		func() {
+			defer func() {
+				if r := recover(); r != nil {
+					log.Error().
+						Interface("panic", r).
+						Str("context", context).
+						Msg("Panic in SaveActiveAlerts goroutine")
+				}
+			}()
+			if err := save(); err != nil {
 				log.Error().
-					Interface("panic", r).
+					Err(err).
 					Str("context", context).
-					Msg("Panic in SaveActiveAlerts goroutine")
+					Msg("Failed to save active alerts")
 			}
 		}()
-		if err := m.SaveActiveAlerts(); err != nil {
-			log.Error().
-				Err(err).
-				Str("context", context).
-				Msg("Failed to save active alerts")
-		}
-	}()
+	}
 }
 
 // SaveActiveAlerts persists active alerts to disk.

internal/alerts/alerts_test.go

@@ -1,6 +1,7 @@
 package alerts
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
 	"math"
@@ -8,6 +9,7 @@ import (
 	"path/filepath"
 	"reflect"
 	"runtime"
+	"runtime/pprof"
 	"strings"
 	"sync"
 	"testing"
@@ -21560,3 +21562,80 @@ func TestIntentCheckpointRetriesFailedWrite(t *testing.T) {
 		t.Fatalf("retry wrote %q", got)
 	}
 }
+
+// A slow checkpoint must not turn an alert burst into one blocked worker per
+// mutation. Exercise the real asynchronous entry point and real JSON writer.
+func TestAsyncActiveCheckpointBurstIsBounded(t *testing.T) {
+	dir := t.TempDir()
+	id, alert := testNewCanonicalAlert("checkpoint-host", "metric-threshold:cpu", "host", "cpu")
+	now := time.Now().UTC()
+	alert.Level, alert.StartTime, alert.LastSeen = AlertLevelWarning, now.Add(-time.Hour), now
+	m := &Manager{
+		alertsDir:     filepath.Join(dir, "alerts"),
+		activeAlerts:  map[string]*Alert{id: alert},
+		intentPending: make(map[string]IntentPendingState),
+	}
+	m.saveMu.Lock()
+	locked := true
+	defer func() {
+		if locked {
+			m.saveMu.Unlock()
+		}
+		m.workerWG.Wait()
+	}()
+	const requests = 512
+	for range requests {
+		m.saveActiveAlertsAsync("burst checkpoint")
+	}
+	var stacks bytes.Buffer
+	if err := pprof.Lookup("goroutine").WriteTo(&stacks, 2); err != nil {
+		t.Fatal(err)
+	}
+	workers := strings.Count(stacks.String(), "created by github.com/rcourtman/pulse-go-rewrite/internal/alerts.(*Manager).saveActiveAlertsAsync")
+	t.Logf("blocked asynchronous checkpoint workers: %d for %d requests", workers, requests)
+	if workers != 1 {
+		t.Errorf("burst started %d checkpoint workers, want one", workers)
+	}
+
+	// State changed while the first checkpoint was blocked must reach both
+	// restart files, not the snapshot that existed at request admission.
+	m.mu.Lock()
+	alert.Acknowledged, alert.AckUser, alert.AckTime = true, "checkpoint-operator", &now
+	alert.LastSeen = now.Add(time.Second)
+	alert.Metadata["checkpoint"] = "latest"
+	m.intentPending["pending-cpu"] = IntentPendingState{
+		TrackingKey: "pending-cpu", ResourceID: "checkpoint-host", Signal: "cpu",
+		FirstMatchedAt: now.Add(-time.Minute), LastObservedAt: now, ElapsedNanos: int64(time.Minute),
+	}
+	m.mu.Unlock()
+	m.saveActiveAlertsAsync("latest state")
+	m.saveMu.Unlock()
+	locked = false
+	m.workerWG.Wait()
+
+	data, err := os.ReadFile(filepath.Join(m.alertsDir, "active-alerts.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var persisted []*Alert
+	if err := json.Unmarshal(data, &persisted); err != nil {
+		t.Fatal(err)
+	}
+	if len(persisted) != 1 || !persisted[0].Acknowledged || persisted[0].AckUser != "checkpoint-operator" ||
+		!persisted[0].StartTime.Equal(alert.StartTime) || !persisted[0].LastSeen.Equal(alert.LastSeen) ||
+		persisted[0].Metadata["checkpoint"] != "latest" {
+		t.Fatalf("checkpoint lost latest incident state: %s", data)
+	}
+	restarted := NewManagerWithDataDir(dir)
+	t.Cleanup(restarted.Stop)
+	restored := testRequireActiveAlert(t, restarted, id)
+	if !restored.Acknowledged || restored.AckUser != "checkpoint-operator" || !restored.StartTime.Equal(alert.StartTime) {
+		t.Fatalf("restart lost acknowledgement or incident age: %+v", restored)
+	}
+	restarted.mu.RLock()
+	pending := restarted.intentPending["pending-cpu"]
+	restarted.mu.RUnlock()
+	if pending.ElapsedNanos != int64(time.Minute) || pending.ResourceID != "checkpoint-host" {
+		t.Fatalf("restart lost latest pending intent: %+v", pending)
+	}
+}

internal/alerts/async_checkpoint_test.go

@@ -0,0 +1,214 @@
+package alerts
+
+import (
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func awaitCheckpointSignal(t *testing.T, signal <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-signal:
+	case <-time.After(5 * time.Second):
+		t.Fatal("checkpoint worker did not reach its barrier")
+	}
+}
+
+func TestAsyncActiveCheckpointCoalescesInFlightRequests(t *testing.T) {
+	m := &Manager{}
+	started, release := make(chan struct{}), make(chan struct{})
+	var calls atomic.Int32
+	if !m.queueActiveAlertsSave("initial") {
+		t.Fatal("first request did not start a worker")
+	}
+	go m.runActiveAlertsSaveWorker(func() error {
+		if calls.Add(1) == 1 {
+			close(started)
+			<-release
+		}
+		return nil
+	})
+	// Always release the worker before test cleanup, even on an assertion failure.
+	var releaseOnce sync.Once
+	t.Cleanup(func() { releaseOnce.Do(func() { close(release) }); m.workerWG.Wait() })
+	awaitCheckpointSignal(t, started)
+	const requests = 10_000
+	for range requests {
+		if m.queueActiveAlertsSave("during checkpoint") {
+			t.Fatal("request during a write admitted another worker")
+		}
+	}
+	releaseOnce.Do(func() { close(release) })
+	m.workerWG.Wait()
+	if got := calls.Load(); got != 2 {
+		t.Fatalf("%d in-flight requests executed %d checkpoints, want exactly two", requests, got)
+	}
+	t.Logf("%d requests during one blocked write produced one follow-up checkpoint", requests)
+	if !m.queueActiveAlertsSave("after idle") {
+		t.Fatal("worker remained stuck after becoming idle")
+	}
+	go m.runActiveAlertsSaveWorker(func() error { calls.Add(1); return nil })
+	m.workerWG.Wait()
+	if got := calls.Load(); got != 3 {
+		t.Fatalf("new request after idle executed %d total checkpoints, want three", got)
+	}
+}
+
+func TestAsyncActiveCheckpointFailureDoesNotStrandRequests(t *testing.T) {
+	for _, failure := range []string{"error", "panic"} {
+		t.Run(failure, func(t *testing.T) {
+			m := &Manager{}
+			started, release := make(chan struct{}), make(chan struct{})
+			var calls atomic.Int32
+			if !m.queueActiveAlertsSave("initial failure") {
+				t.Fatal("first request did not start a worker")
+			}
+			go m.runActiveAlertsSaveWorker(func() error {
+				if calls.Add(1) == 1 {
+					close(started)
+					<-release
+					if failure == "panic" {
+						panic("test checkpoint panic")
+					}
+					return errors.New("test checkpoint failure")
+				}
+				return nil
+			})
+			var releaseOnce sync.Once
+			t.Cleanup(func() { releaseOnce.Do(func() { close(release) }); m.workerWG.Wait() })
+			awaitCheckpointSignal(t, started)
+			if m.queueActiveAlertsSave("after failure") {
+				t.Fatal("request during a failing write admitted another worker")
+			}
+			releaseOnce.Do(func() { close(release) })
+			m.workerWG.Wait()
+			if got := calls.Load(); got != 2 {
+				t.Fatalf("failure stranded the follow-up request: %d checkpoints", got)
+			}
+			if !m.queueActiveAlertsSave("later retry") {
+				t.Fatal("failed worker did not release admission")
+			}
+			go m.runActiveAlertsSaveWorker(func() error { calls.Add(1); return nil })
+			m.workerWG.Wait()
+			if got := calls.Load(); got != 3 {
+				t.Fatalf("later retry did not execute: %d checkpoints", got)
+			}
+		})
+	}
+}
+
+func TestAsyncActiveCheckpointStopOwnsFinalState(t *testing.T) {
+	m := &Manager{
+		alertsDir: t.TempDir(), activeAlerts: make(map[string]*Alert),
+		escalationStop: make(chan struct{}), cleanupStop: make(chan struct{}),
+	}
+	started, release, stopped := make(chan struct{}), make(chan struct{}), make(chan struct{})
+	var calls atomic.Int32
+	if !m.queueActiveAlertsSave("initial") {
+		t.Fatal("first request did not start a worker")
+	}
+	go m.runActiveAlertsSaveWorker(func() error {
+		calls.Add(1)
+		close(started)
+		<-release
+		return nil
+	})
+	var releaseOnce sync.Once
+	t.Cleanup(func() { releaseOnce.Do(func() { close(release) }); m.Stop() })
+	awaitCheckpointSignal(t, started)
+	if m.queueActiveAlertsSave("superseded at shutdown") {
+		t.Fatal("follow-up request admitted another worker")
+	}
+	go func() { m.Stop(); close(stopped) }()
+	awaitCheckpointSignal(t, m.escalationStop)
+	if m.queueActiveAlertsSave("too late") {
+		t.Fatal("shutdown admitted a new worker")
+	}
+	select {
+	case <-stopped:
+		t.Fatal("Stop returned before the in-flight checkpoint finished")
+	default:
+	}
+	m.mu.Lock()
+	m.activeAlerts["shutdown-final"] = &Alert{ID: "shutdown-final", Acknowledged: true, AckUser: "final-operator"}
+	m.mu.Unlock()
+	releaseOnce.Do(func() { close(release) })
+	awaitCheckpointSignal(t, stopped)
+	if got := calls.Load(); got != 1 {
+		t.Fatalf("shutdown drained %d redundant async checkpoints, want one in-flight", got)
+	}
+	data, err := os.ReadFile(filepath.Join(m.alertsDir, "active-alerts.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(string(data), "final-operator") {
+		t.Fatalf("Stop did not checkpoint the latest state: %s", data)
+	}
+}
+
+func TestAsyncActiveCheckpointConcurrentRequestsAndStop(t *testing.T) {
+	m := NewManagerWithDataDir(t.TempDir())
+	m.saveMu.Lock()
+	var unlockOnce sync.Once
+	t.Cleanup(func() { unlockOnce.Do(m.saveMu.Unlock); m.Stop() })
+	m.saveActiveAlertsAsync("initial blocked checkpoint")
+	start := make(chan struct{})
+	entered := make(chan struct{}, 16)
+	var requesters sync.WaitGroup
+	for range 16 {
+		requesters.Add(1)
+		go func() {
+			defer requesters.Done()
+			<-start
+			m.saveActiveAlertsAsync("before shutdown")
+			entered <- struct{}{}
+			for range 256 {
+				m.saveActiveAlertsAsync("concurrent request")
+			}
+		}()
+	}
+	close(start)
+	for range 16 {
+		<-entered
+	}
+	stopped := make(chan struct{})
+	go func() { m.Stop(); close(stopped) }()
+	awaitCheckpointSignal(t, m.escalationStop)
+	requesters.Wait()
+	unlockOnce.Do(m.saveMu.Unlock)
+	awaitCheckpointSignal(t, stopped)
+	m.stopMu.RLock()
+	defer m.stopMu.RUnlock()
+	if m.activeSaveRunning {
+		t.Fatal("checkpoint worker survived shutdown")
+	}
+}
+
+func TestAsyncActiveCheckpointWriteFailureCanRetry(t *testing.T) {
+	dir := filepath.Join(t.TempDir(), "alerts")
+	if err := os.WriteFile(dir, []byte("block checkpoint directory creation"), 0600); err != nil {
+		t.Fatal(err)
+	}
+	m := &Manager{alertsDir: dir, activeAlerts: make(map[string]*Alert)}
+	m.saveActiveAlertsAsync("unavailable directory")
+	m.workerWG.Wait()
+	data, err := os.ReadFile(dir)
+	if err != nil || string(data) != "block checkpoint directory creation" {
+		t.Fatalf("failed save changed the source: %q, %v", data, err)
+	}
+	if err := os.Remove(dir); err != nil {
+		t.Fatal(err)
+	}
+	m.saveActiveAlertsAsync("directory restored")
+	m.workerWG.Wait()
+	data, err = os.ReadFile(filepath.Join(dir, "active-alerts.json"))
+	if err != nil || string(data) != "[]" {
+		t.Fatalf("retry after a failed checkpoint did not persist: %q, %v", data, err)
+	}
+}

internal/alerts/manager.go

@@ -145,6 +145,11 @@ type Manager struct {
 	stopMu        sync.RWMutex
 	stopping      bool
 	workerWG      sync.WaitGroup
+	// stopMu protects checkpoint worker admission and its coalesced request.
+	// The worker releases stopMu before taking saveMu or the manager state lock.
+	activeSaveRunning bool
+	activeSavePending bool
+	activeSaveContext string
 }
 
 type ackRecord struct {

scripts/installtests/build_release_assets_test.go

@@ -2415,6 +2415,35 @@ func TestUpdateDemoResolverChecksOutOutputHelperBeforeUse(t *testing.T) {
 	}
 }
 
+func TestReleaseWatchdogIsControlBoundAndCannotBuildCandidate(t *testing.T) {
+	workflow, err := os.ReadFile(repoFile(".github", "workflows", "release-dry-run.yml"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	text := string(workflow)
+	for _, required := range []string{
+		"Release Watchdog at {0}",
+		"if: ${{ inputs.watchdog != true && inputs.version != '' }}",
+		`[ "${EXPECTED_WORKFLOW_SHA_INPUT}" != "${GITHUB_SHA}" ]`,
+		`[ "${GITHUB_REF}" != "refs/heads/main" ]`,
+		`Candidate rehearsal requires explicit rollback and no watchdog SHA.`,
+		`Watchdog refuses candidate input ${name}.`,
+		`Watchdog refuses exception input ${name}.`,
+		"WATCHDOG_MODE: ${{ steps.mode.outputs.watchdog }}",
+		"verify_only: true",
+		`require_result "stable demo no-mutation verification" "$DEMO_RESULT" success`,
+	} {
+		if !strings.Contains(text, required) {
+			t.Fatalf("fixed release watchdog missing safety contract: %s", required)
+		}
+	}
+	guard := strings.Index(text, "- name: Validate release ref")
+	checkout := strings.Index(text, "- name: Checkout repository")
+	if guard < 0 || checkout <= guard {
+		t.Fatal("watchdog control identity and envelope must be checked before checkout")
+	}
+}
+
 func TestDemoMutationAndRecoverySharePhysicalTargetLock(t *testing.T) {
 	updateBytes, err := os.ReadFile(repoFile(".github", "workflows", "update-demo-server.yml"))
 	if err != nil {

scripts/release_control/rehearsal_source_test.py

@@ -64,9 +64,10 @@ class RehearsalSourceTest(unittest.TestCase):
         self.git("add", ".")
         self.git("commit", "-m", "fixture")
 
-    def select(self, event="schedule", branch="main", sha=None, required="release/v6.4"):
+    def select(self, event="schedule", branch="main", sha=None, required="release/v6.4", watchdog=None):
         env = dict(self.env, EVENT_NAME=event, GITHUB_REF_NAME=branch,
                    GITHUB_SHA=sha or self.main, REQUIRED_BRANCH=required,
+                   WATCHDOG_MODE=watchdog or ("true" if event == "schedule" else "false"),
                    GITHUB_OUTPUT=str(self.output))
         return subprocess.run(
             ["bash", "-euo", "pipefail", "-c", step("Select exact rehearsal source")],
@@ -83,6 +84,69 @@ class RehearsalSourceTest(unittest.TestCase):
         self.assertIn(self.main, result.stdout)
         self.assertNotEqual(self.main, self.release)
 
+    def mode(self, **overrides):
+        env = dict(self.env, EVENT_NAME="workflow_dispatch", WATCHDOG_INPUT="true",
+                   EXPECTED_WORKFLOW_SHA_INPUT=self.main, GITHUB_SHA=self.main,
+                   GITHUB_REF="refs/heads/main", GITHUB_REF_NAME="main",
+                   GITHUB_OUTPUT=str(self.output))
+        env.update(overrides)
+        return subprocess.run(["bash", "-euo", "pipefail", "-c", step("Validate release ref")],
+                              cwd=self.repo, env=env, text=True, capture_output=True)
+
+    def test_fixed_watchdog_dispatch_selects_same_source_as_schedule(self):
+        admitted = self.mode()
+        self.assertEqual(admitted.returncode, 0, admitted.stderr)
+        self.assertEqual(self.output.read_text(), "watchdog=true\n")
+        result = self.select("workflow_dispatch", watchdog="true")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(self.git("rev-parse", "HEAD"), self.release)
+        self.assertIn(self.main, result.stdout)
+        self.assertIn("tested_sha=" + self.release, self.output.read_text())
+
+    def test_watchdog_rejects_moved_control_wrong_ref_and_unknown_event_before_selection(self):
+        for overrides in (
+            {"EXPECTED_WORKFLOW_SHA_INPUT": "a" * 40},
+            {"EXPECTED_WORKFLOW_SHA_INPUT": "not-a-sha"},
+            {"EXPECTED_WORKFLOW_SHA_INPUT": ""},
+            {"GITHUB_REF": "refs/heads/release/v6.4"},
+            {"GITHUB_REF": "refs/tags/v6.4.5"},
+            {"EVENT_NAME": "push"}, {"WATCHDOG_INPUT": "yes"},
+        ):
+            with self.subTest(overrides=overrides):
+                result = self.mode(**overrides)
+                self.assertNotEqual(result.returncode, 0, result.stdout)
+                self.assertFalse(self.output.exists())
+                self.assertEqual(self.git("rev-parse", "HEAD"), self.main)
+
+    def test_watchdog_rejects_every_candidate_and_exception_input(self):
+        for name in (
+            "VERSION_INPUT", "PROMOTED_FROM_TAG_INPUT", "ROLLBACK_VERSION_INPUT",
+            "GA_DATE_INPUT", "V5_EOS_DATE_INPUT", "HOTFIX_REASON_INPUT",
+            "UNSIGNED_WINDOWS_REASON_INPUT", "NOTE_INPUT", "MOBILE_RELEASE_DECISION",
+            "MOBILE_RELEASE_EVIDENCE", "HOTFIX_EXCEPTION_INPUT", "UNSIGNED_WINDOWS_EXCEPTION_INPUT",
+        ):
+            with self.subTest(name=name):
+                self.assertNotEqual(self.mode(**{name: "true"}).returncode, 0)
+                self.assertFalse(self.output.exists())
+        self.assertEqual(self.mode(HOTFIX_EXCEPTION_INPUT="false",
+                                   UNSIGNED_WINDOWS_EXCEPTION_INPUT="false").returncode, 0)
+
+    def test_schedule_and_manual_candidate_envelopes_remain_separate(self):
+        result = self.mode(EVENT_NAME="schedule", WATCHDOG_INPUT="", EXPECTED_WORKFLOW_SHA_INPUT="")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(self.output.read_text(), "watchdog=true\n")
+        self.output.unlink()
+        for overrides in ({}, {"ROLLBACK_VERSION_INPUT": "v6.4.1", "EXPECTED_WORKFLOW_SHA_INPUT": self.main}):
+            with self.subTest(overrides=overrides):
+                env = {"WATCHDOG_INPUT": "false", "EXPECTED_WORKFLOW_SHA_INPUT": ""}
+                env.update(overrides)
+                self.assertNotEqual(self.mode(**env).returncode, 0)
+                self.assertFalse(self.output.exists())
+        result = self.mode(WATCHDOG_INPUT="false", EXPECTED_WORKFLOW_SHA_INPUT="",
+                           ROLLBACK_VERSION_INPUT="v6.4.1")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(self.output.read_text(), "watchdog=false\n")
+
     def test_manual_wrong_branch_rejected_without_switching(self):
         result = self.select("workflow_dispatch")
         self.assertNotEqual(result.returncode, 0)
@@ -104,6 +168,17 @@ class RehearsalSourceTest(unittest.TestCase):
         self.assertEqual(self.git("rev-parse", "HEAD"), self.main)
         self.assertFalse(self.output.exists())
 
+    def test_explicit_watchdog_missing_branch_and_policy_drift_fail_closed(self):
+        self.assertNotEqual(self.select("workflow_dispatch", watchdog="true",
+                                        required="release/missing").returncode, 0)
+        self.assertFalse(self.output.exists())
+        self.git("checkout", "release/v6.4")
+        self.version("6.6.0-beta.1")
+        self.git("checkout", "main")
+        self.assertNotEqual(self.select("workflow_dispatch", watchdog="true").returncode, 0)
+        self.assertEqual(self.git("rev-parse", "HEAD"), self.main)
+        self.assertFalse(self.output.exists())
+
     def test_selected_version_must_still_belong_to_governed_branch(self):
         self.git("checkout", "release/v6.4")
         self.version("6.6.0-beta.1")
@@ -119,7 +194,7 @@ class RehearsalSourceTest(unittest.TestCase):
         metadata = step("Resolve rehearsal metadata")
         self.assertIn('if [ "${TESTED_BRANCH}" != "$REQUIRED_BRANCH" ]; then', metadata)
         self.assertIn('if [ "$FILE_VERSION" != "$VERSION" ]; then', metadata)
-        self.assertIn('if [ "${EVENT_NAME}" = "schedule" ] && [ -z "${ROLLBACK_VERSION_INPUT:-}" ]; then', metadata)
+        self.assertIn('if [ "${WATCHDOG_MODE}" = "true" ] && [ -z "${ROLLBACK_VERSION_INPUT:-}" ]; then', metadata)
         self.assertIn('--derive-rollback-latest-stable', metadata)
         workflow = WORKFLOW.read_text()
         self.assertIn('TESTED_SHA: ${{ needs.dry-run.outputs.tested_sha }}', workflow)

scripts/release_control/release_promotion_policy_test.py

@@ -408,6 +408,7 @@ class ReleasePromotionPolicyTest(unittest.TestCase):
         dry_run = jobs["dry-run"]
         steps = dry_run["steps"]
         names = [step["name"] for step in steps]
+        self.assertLess(names.index("Validate release ref"), names.index("Checkout repository"))
         self.assertLess(names.index("Resolve required release branch"),
                         names.index("Select exact rehearsal source"))
         self.assertLess(names.index("Select exact rehearsal source"),
@@ -415,6 +416,7 @@ class ReleasePromotionPolicyTest(unittest.TestCase):
         source = next(step for step in steps if step.get("id") == "source")
         self.assertEqual(source["env"]["REQUIRED_BRANCH"],
                          "${{ steps.branch_policy.outputs.required_branch }}")
+        self.assertEqual(source["env"]["WATCHDOG_MODE"], "${{ steps.mode.outputs.watchdog }}")
         self.assertIn('git checkout --detach "${TESTED_SHA}"', source["run"])
         self.assertIn('"${TESTED_SHA}" != "${GITHUB_SHA}"', source["run"])
         metadata = next(step for step in steps if step.get("id") == "rehearsal")
@@ -424,9 +426,13 @@ class ReleasePromotionPolicyTest(unittest.TestCase):
                       metadata["run"])
         self.assertIn('if [ "$FILE_VERSION" != "$VERSION" ]; then', metadata["run"])
         self.assertIn(
-            'if [ "${EVENT_NAME}" = "schedule" ] && [ -z "${ROLLBACK_VERSION_INPUT:-}" ]; then',
+            'if [ "${WATCHDOG_MODE}" = "true" ] && [ -z "${ROLLBACK_VERSION_INPUT:-}" ]; then',
             metadata["run"],
         )
+        self.assertEqual(metadata["env"]["WATCHDOG_MODE"], "${{ steps.mode.outputs.watchdog }}")
+        self.assertEqual(jobs["build_release_candidate"]["if"],
+                         "${{ inputs.watchdog != true && inputs.version != '' }}")
+        self.assertIn("Release Watchdog at {0}", yaml.safe_load(workflow)["run-name"])
         for key in ("tested_sha", "tested_branch"):
             self.assertEqual(dry_run["outputs"][key],
                              "${{ steps.source.outputs." + key + " }}")
@@ -1707,7 +1713,7 @@ class ReleasePromotionPolicyTest(unittest.TestCase):
         self.assertIn(promotion_metadata_envelope(), normalize_ws(template))
         self.assertIn("rc-to-ga-rehearsal-summary", workflow)
         self.assertIn("build_release_candidate:", workflow)
-        self.assertIn("if: ${{ inputs.version != '' }}", workflow)
+        self.assertIn("if: ${{ inputs.watchdog != true && inputs.version != '' }}", workflow)
         self.assertIn("require_macos_signing: true", workflow)
         self.assertIn(
             "require_windows_signing: false",
@@ -2194,12 +2200,19 @@ class ReleasePromotionPolicyTest(unittest.TestCase):
         dry_run_workflow = read(".github/workflows/release-dry-run.yml")
         self.assertIn("Required rollback stable version to rehearse", dry_run_workflow)
         self.assertIn("rollback_version:\n        description: 'Required rollback stable version to rehearse", dry_run_workflow)
-        self.assertIn("required: true", dry_run_workflow)
-        # Scheduled watchdog runs carry no dispatch inputs, so the rehearsal
-        # step must derive the rollback target; the derive flag stays gated on
-        # the schedule event so manual dispatches keep explicit rollback.
+        # The fixed watchdog has no rollback override, so its dispatch schema
+        # must permit omission. Ordinary manual rehearsals still require an
+        # explicit rollback in the pre-checkout admission shell.
+        dry_run = yaml.safe_load(dry_run_workflow)
+        inputs = dry_run.get("on", dry_run.get(True))["workflow_dispatch"]["inputs"]
+        self.assertIs(inputs["rollback_version"]["required"], False)
+        self.assertIn('workflow_dispatch:false)', dry_run_workflow)
+        self.assertIn('[ -z "${ROLLBACK_VERSION_INPUT:-}" ]', dry_run_workflow)
+        self.assertIn('Candidate rehearsal requires explicit rollback and no watchdog SHA.', dry_run_workflow)
+        # Both admitted watchdog modes derive rollback; a candidate rehearsal
+        # can never reach that derivation, including when the input is absent.
         self.assertIn(
-            'if [ "${EVENT_NAME}" = "schedule" ] && [ -z "${ROLLBACK_VERSION_INPUT:-}" ]; then',
+            'if [ "${WATCHDOG_MODE}" = "true" ] && [ -z "${ROLLBACK_VERSION_INPUT:-}" ]; then',
             dry_run_workflow,
         )
         self.assertIn("--derive-rollback-latest-stable", dry_run_workflow)