diff --git a/docs/site/scripts/build-archives.mjs b/docs/site/scripts/build-archives.mjs
index fb57cfaec..d2585c507 100644
--- a/docs/site/scripts/build-archives.mjs
+++ b/docs/site/scripts/build-archives.mjs
@@ -5,6 +5,12 @@ import { tmpdir } from 'node:os';
 import { dirname, relative, resolve, sep } from 'node:path';
 import { fileURLToPath } from 'node:url';
 import { promisify } from 'node:util';
+import {
+  constants as zlibConstants,
+  crc32,
+  deflateRawSync,
+  gunzipSync,
+} from 'node:zlib';
 import { applyArchiveSeo } from './apply-archive-seo.mjs';
 import {
   archiveOutputDirectory,
@@ -170,6 +176,28 @@ async function replaceFile(path, contents) {
   }
 }
 
+const deterministicGzipHeader = Buffer.from([
+  0x1f, 0x8b, 0x08, 0x00,
+  0x00, 0x00, 0x00, 0x00,
+  0x02, 0xff,
+]);
+
+function deterministicGzip(contents) {
+  // zlib records the host OS in its gzip header, so gzipSync emits different
+  // bytes on macOS and Linux. Build the framing explicitly with no optional
+  // fields, zero mtime, maximum-compression XFL, and the unknown-OS marker.
+  const compressed = deflateRawSync(contents, {
+    level: 9,
+    memLevel: 8,
+    strategy: zlibConstants.Z_DEFAULT_STRATEGY,
+    windowBits: 15,
+  });
+  const trailer = Buffer.alloc(8);
+  trailer.writeUInt32LE(crc32(contents), 0);
+  trailer.writeUInt32LE(contents.length >>> 0, 4);
+  return Buffer.concat([deterministicGzipHeader, compressed, trailer]);
+}
+
 export async function normalizePagefindGzipMetadata(outputRoot) {
   const pagefindRoot = resolve(outputRoot, 'pagefind');
   let pagefindInfo;
@@ -204,9 +232,16 @@ export async function normalizePagefindGzipMetadata(outputRoot) {
       throw new Error(`generated Pagefind WebAssembly must use gzip framing: ${path}`);
     }
     files += 1;
-    if (contents.subarray(4, 8).some((byte) => byte !== 0)) {
-      const updated = Buffer.from(contents);
-      updated.fill(0, 4, 8);
+    let uncompressed;
+    try {
+      uncompressed = gunzipSync(contents);
+    } catch (error) {
+      throw new Error(`generated Pagefind WebAssembly must be valid gzip: ${path}`, {
+        cause: error,
+      });
+    }
+    const updated = deterministicGzip(uncompressed);
+    if (!contents.equals(updated)) {
       await replaceFile(path, updated);
       normalized += 1;
     }
diff --git a/docs/site/scripts/build-archives.test.mjs b/docs/site/scripts/build-archives.test.mjs
index f3635866d..76dd94d97 100644
--- a/docs/site/scripts/build-archives.test.mjs
+++ b/docs/site/scripts/build-archives.test.mjs
@@ -13,7 +13,7 @@ import { dirname, resolve } from 'node:path';
 import test from 'node:test';
 import { fileURLToPath } from 'node:url';
 import { promisify } from 'node:util';
-import { gzipSync } from 'node:zlib';
+import { gunzipSync, gzipSync } from 'node:zlib';
 
 import {
   buildDocsetArchive,
@@ -65,7 +65,7 @@ test('archive snapshot reads one no-follow regular-file descriptor', async (t) =
   assert.equal(await readOptionalRegularFile(resolve(root, 'missing.json')), null);
 });
 
-test('Pagefind gzip metadata normalizes to a stable archive tree', async (t) => {
+test('Pagefind gzip streams normalize across platforms without changing content', async (t) => {
   const root = await mkdtemp(resolve(tmpdir(), 'registry-docs-pagefind-gzip-'));
   t.after(() => rm(root, { recursive: true, force: true }));
   const left = resolve(root, 'left');
@@ -73,15 +73,21 @@ test('Pagefind gzip metadata normalizes to a stable archive tree', async (t) =>
   await mkdir(resolve(left, 'pagefind'), { recursive: true });
   await mkdir(resolve(right, 'pagefind'), { recursive: true });
 
-  const compressed = gzipSync('architecture-independent WebAssembly');
+  const wasm = Buffer.from(
+    `\0asm\x01\0\0\0${'architecture-independent WebAssembly'.repeat(256)}`,
+  );
   for (const name of ['wasm.en.pagefind', 'wasm.unknown.pagefind']) {
-    const leftContents = Buffer.from(compressed);
-    const rightContents = Buffer.from(compressed);
+    const leftContents = gzipSync(wasm, { level: 1 });
+    const rightContents = gzipSync(wasm, { level: 9 });
     leftContents.writeUInt32LE(1_700_000_000, 4);
     rightContents.writeUInt32LE(1_800_000_000, 4);
+    leftContents[9] = 0x03;
+    rightContents[9] = 0x13;
     await writeFile(resolve(left, 'pagefind', name), leftContents);
     await writeFile(resolve(right, 'pagefind', name), rightContents);
     assert.notDeepEqual(leftContents, rightContents);
+    assert.deepEqual(gunzipSync(leftContents), wasm);
+    assert.deepEqual(gunzipSync(rightContents), wasm);
   }
 
   assert.deepEqual(
@@ -95,10 +101,18 @@ test('Pagefind gzip metadata normalizes to a stable archive tree', async (t) =>
   for (const name of ['wasm.en.pagefind', 'wasm.unknown.pagefind']) {
     const normalizedLeft = await readFile(resolve(left, 'pagefind', name));
     const normalizedRight = await readFile(resolve(right, 'pagefind', name));
-    assert.deepEqual(normalizedLeft.subarray(4, 8), Buffer.alloc(4));
+    assert.deepEqual(
+      normalizedLeft.subarray(0, 10),
+      Buffer.from([0x1f, 0x8b, 0x08, 0x00, 0, 0, 0, 0, 0x02, 0xff]),
+    );
     assert.deepEqual(normalizedLeft, normalizedRight);
+    assert.deepEqual(gunzipSync(normalizedLeft), wasm);
   }
   assert.equal(await treeDigest(left), await treeDigest(right));
+  assert.deepEqual(
+    await normalizePagefindGzipMetadata(left),
+    { files: 2, normalized: 0 },
+  );
 });
 
 test('Pagefind metadata normalization rejects an unexpected WASM format', async (t) => {
diff --git a/release/OPERATIONS.md b/release/OPERATIONS.md
index 428acf818..833403c09 100644
--- a/release/OPERATIONS.md
+++ b/release/OPERATIONS.md
@@ -208,6 +208,12 @@ The command prints the exact candidate run ID and URL immediately after the
 dispatch is correlated. `--wait-for-ci` waits only for protected-main `ci.yml`
 at the exact source SHA, then refreshes protected `main` again immediately
 before dispatch. `--wait` follows only that uniquely identified candidate run.
+Both waits print only workflow state changes by default. If a protected
+environment is waiting for approval, the command names the environment, links
+the exact run, and prints a read-only command for inspecting the pending
+deployment. An authorized reviewer must approve it through **Review
+deployments** in that run. Add `--verbose-wait` to retain the raw `gh run watch`
+display when detailed live job output is useful.
 Omit either flag when another operator or monitor owns the corresponding wait.
 
 The request is accepted only when `source_sha` is the exact protected-main
@@ -443,6 +449,23 @@ SHA-512 integrity and PyPI SHA-256 digest of every client package before
 docs promotion. The public verifier is read-only and can be rerun
 independently.
 
+For an interrupted publication after the annotated tag exists, classify the
+exact recovery state before retrying:
+
+```sh
+release/scripts/registry-release verify-recovery --tag v<version>
+```
+
+This command is read-only. For an absent release or a bound draft, it verifies
+that the local annotated tag exactly matches `origin`, revalidates the original
+candidate and its lifetime, and checks the draft's candidate binding. It then
+prints the exact protected-main `release.yml` retry command. The workflow owns
+the fail-closed reconciliation of draft assets, OCI digests, npm packages, and
+PyPI wheels immediately before each write. If the release is already
+published, the command runs `verify-public`, reports the release complete, and
+does not recommend a retry. It never approves environments or dispatches a
+workflow, and it adds no release gate.
+
 ## Failure handling
 
 | Failure state | Response |
diff --git a/release/scripts/registry-release b/release/scripts/registry-release
index b3dfe08c6..fdea421f0 100755
--- a/release/scripts/registry-release
+++ b/release/scripts/registry-release
@@ -2012,7 +2012,12 @@ def workflow_run_identity(run: dict[str, Any], label: str) -> tuple[int, str]:
     return run_id, url
 
 
-def wait_for_exact_protected_ci(repository: str, source_sha: str) -> dict[str, Any]:
+def wait_for_exact_protected_ci(
+    repository: str,
+    source_sha: str,
+    *,
+    verbose_wait: bool = False,
+) -> dict[str, Any]:
     deadline = time.monotonic() + 120
     while True:
         runs = workflow_runs(
@@ -2032,7 +2037,12 @@ def wait_for_exact_protected_ci(repository: str, source_sha: str) -> dict[str, A
         if active:
             run = max(active, key=lambda item: int(item.get("id", 0)))
             run_id, _ = workflow_run_identity(run, "protected-main CI")
-            watch_workflow_run(repository, run_id, "protected-main CI")
+            watch_workflow_run(
+                repository,
+                run_id,
+                "protected-main CI",
+                verbose=verbose_wait,
+            )
             repeated = workflow_runs(
                 repository,
                 "ci.yml",
@@ -2095,7 +2105,97 @@ def wait_for_dispatched_run(
         time.sleep(1)
 
 
-def watch_workflow_run(repository: str, run_id: int, label: str) -> None:
+def workflow_run(repository: str, run_id: int, label: str) -> dict[str, Any]:
+    document = json.loads(
+        run_checked(["gh", "api", f"repos/{repository}/actions/runs/{run_id}"])
+    )
+    if not isinstance(document, dict):
+        raise ReleasePlanError(f"GitHub returned malformed state for {label} run {run_id}")
+    observed_id, _ = workflow_run_identity(document, label)
+    if observed_id != run_id:
+        raise ReleasePlanError(f"GitHub returned the wrong run for {label} run {run_id}")
+    return document
+
+
+def pending_deployments(repository: str, run_id: int) -> list[dict[str, Any]]:
+    document = json.loads(
+        run_checked(
+            [
+                "gh",
+                "api",
+                f"repos/{repository}/actions/runs/{run_id}/pending_deployments",
+            ]
+        )
+    )
+    if not isinstance(document, list) or any(
+        not isinstance(deployment, dict) for deployment in document
+    ):
+        raise ReleasePlanError(
+            f"GitHub returned malformed pending deployments for run {run_id}"
+        )
+    return document
+
+
+def pending_environment_names(deployments: list[dict[str, Any]]) -> tuple[str, ...]:
+    names = []
+    for deployment in deployments:
+        environment = deployment.get("environment")
+        name = environment.get("name") if isinstance(environment, dict) else None
+        if not isinstance(name, str) or not name:
+            raise ReleasePlanError("GitHub returned a pending deployment without an environment")
+        names.append(name)
+    return tuple(sorted(set(names)))
+
+
+def watch_workflow_run(
+    repository: str,
+    run_id: int,
+    label: str,
+    *,
+    verbose: bool = False,
+) -> None:
+    if not verbose:
+        previous: tuple[str, str | None, tuple[str, ...]] | None = None
+        while True:
+            run = workflow_run(repository, run_id, label)
+            status = run.get("status")
+            conclusion = run.get("conclusion")
+            if not isinstance(status, str) or (
+                conclusion is not None and not isinstance(conclusion, str)
+            ):
+                raise ReleasePlanError(f"{label} run {run_id} has malformed state")
+            environments: tuple[str, ...] = ()
+            if status in {"queued", "in_progress", "pending", "waiting"}:
+                environments = pending_environment_names(
+                    pending_deployments(repository, run_id)
+                )
+            state = (status, conclusion, environments)
+            if state != previous:
+                detail = status if conclusion is None else f"{status}/{conclusion}"
+                if previous is None:
+                    print(f"{label} run {run_id}: {detail} ({run['html_url']})")
+                else:
+                    print(f"{label} run {run_id}: {detail}")
+                if environments:
+                    names = ", ".join(environments)
+                    print(
+                        "pending protected-environment approval for "
+                        f"{names}; an authorized reviewer must open the run URL above "
+                        "and use Review deployments. Inspect the pending request with:"
+                    )
+                    print(
+                        "gh api "
+                        f"repos/{repository}/actions/runs/{run_id}/pending_deployments"
+                    )
+                previous = state
+            if status == "completed":
+                if conclusion != "success":
+                    raise ReleasePlanError(
+                        f"{label} run {run_id} concluded {conclusion!r}"
+                    )
+                return
+            time.sleep(5)
+
     try:
         result = subprocess.run(
             [
@@ -2150,6 +2250,7 @@ def request_release_candidate(
     print_request: bool,
     wait_for_ci: bool = False,
     wait_for_candidate: bool = False,
+    verbose_wait: bool = False,
 ) -> int:
     try:
         context = prepare_release_context(repo, version, release_id)
@@ -2178,7 +2279,11 @@ def request_release_candidate(
                 "and stack.status"
             )
         if wait_for_ci:
-            ci_run = wait_for_exact_protected_ci(repository, resolved_source)
+            ci_run = wait_for_exact_protected_ci(
+                repository,
+                resolved_source,
+                verbose_wait=verbose_wait,
+            )
             ci_run_id, ci_url = workflow_run_identity(ci_run, "protected-main CI")
             print(
                 f"protected-main CI passed for {resolved_source}: "
@@ -2243,6 +2348,7 @@ def request_release_candidate(
                     repository,
                     candidate_run_id,
                     "release candidate",
+                    verbose=verbose_wait,
                 )
                 print(f"release candidate run {candidate_run_id} passed")
     except (OSError, UnicodeError, ReleasePlanError) as exc:
@@ -2363,7 +2469,7 @@ def verify_candidate_run(
     repository: str,
     candidate_run: int,
     version: str,
-    release_id: str,
+    release_id: str | None,
     source_sha: str | None,
 ) -> tuple[dict[str, Any], str]:
     with tempfile.TemporaryDirectory(prefix="registry-candidate-") as directory:
@@ -2466,6 +2572,183 @@ def verify_candidate_run(
         return manifest, binding
 
 
+def tagged_candidate_binding(
+    repo: Path,
+    tag: str,
+) -> tuple[str, dict[str, Any]]:
+    reference = f"refs/tags/{tag}"
+    if run_checked(["git", "cat-file", "-t", reference], cwd=repo).strip() != "tag":
+        raise ReleasePlanError(f"{tag} is not one immutable annotated source tag")
+    local_object = run_checked(["git", "rev-parse", reference], cwd=repo).strip()
+    source_sha = resolve_commit(repo, reference, f"annotated tag {tag} target")
+    peeled = f"{reference}^{{}}"
+    remote_lines = run_checked(
+        ["git", "ls-remote", "--tags", "origin", reference, peeled],
+        cwd=repo,
+    ).splitlines()
+    remote: dict[str, str] = {}
+    for line in remote_lines:
+        parts = line.split("\t", 1)
+        if (
+            len(parts) != 2
+            or HEX40.fullmatch(parts[0]) is None
+            or parts[1] in remote
+        ):
+            raise ReleasePlanError(f"origin returned malformed state for tag {tag}")
+        remote[parts[1]] = parts[0]
+    if remote != {reference: local_object, peeled: source_sha}:
+        raise ReleasePlanError(
+            f"local annotated tag {tag} does not exactly match origin; fetch and "
+            "inspect the immutable remote tag before recovery"
+        )
+    message = run_checked(
+        ["git", "for-each-ref", "--format=%(contents)", reference],
+        cwd=repo,
+    )
+    return source_sha, release_candidate.parse_tag_binding(message)
+
+
+def release_for_tag(repository: str, tag: str) -> dict[str, Any] | None:
+    pages = json.loads(
+        run_checked(
+            [
+                "gh",
+                "api",
+                f"repos/{repository}/releases?per_page=100",
+                "--paginate",
+                "--slurp",
+            ]
+        )
+    )
+    if not isinstance(pages, list) or any(not isinstance(page, list) for page in pages):
+        raise ReleasePlanError("GitHub Releases response is malformed")
+    matches = [
+        release
+        for page in pages
+        for release in page
+        if isinstance(release, dict) and release.get("tag_name") == tag
+    ]
+    if len(matches) > 1:
+        raise ReleasePlanError(f"GitHub Release destination {tag} is ambiguous")
+    return matches[0] if matches else None
+
+
+def validate_recovery_release(
+    release: dict[str, Any] | None,
+    *,
+    tag: str,
+    manifest_sha256: str,
+) -> str:
+    if release is None:
+        return "absent"
+    marker = f"registry-stack-release-candidate-v2 manifest_sha256:{manifest_sha256}"
+    if (
+        release.get("tag_name") != tag
+        or release.get("name") != f"RegistryStack {tag}"
+        or release.get("prerelease") is not False
+        or marker not in str(release.get("body", ""))
+    ):
+        raise ReleasePlanError(f"GitHub Release {tag} is not bound to this candidate")
+    if release.get("draft") is not True or release.get("published_at") is not None:
+        raise ReleasePlanError(f"GitHub Release {tag} has malformed draft state")
+    return "draft"
+
+
+def verify_release_recovery(
+    repo: Path,
+    *,
+    tag: str,
+    repository: str,
+) -> int:
+    try:
+        if not tag.startswith("v"):
+            raise ReleasePlanError("tag must be canonical v<major>.<minor>.<patch> text")
+        tag_version = strict_semver(tag.removeprefix("v"))
+        verify_origin_repository(repo, repository)
+        release = release_for_tag(repository, tag)
+        if release is not None and release.get("draft") is False:
+            result = verify_public_release.verify(
+                repo=repo,
+                repository=repository,
+                tag=tag,
+            )
+            result["operation"] = "verify-recovery"
+            result["release_state"] = "published"
+            result["status"] = "complete"
+            print(json.dumps(result, indent=2, sort_keys=True))
+            print(
+                f"Release {tag} is already published and verified; do not retry it.",
+                file=sys.stderr,
+            )
+            return 0
+        if tag_version[0] != 0:
+            raise ReleasePlanError("Beta publication accepts only v0.x.y release tags")
+        if tag_version < RELAY_V2_RELEASE_MINIMUM_VERSION:
+            raise ReleasePlanError(
+                "pre-v0.19 releases are immutable historical evidence; use the "
+                f"corresponding {tag} Git tag and archived assets"
+            )
+        source_sha, tag_binding = tagged_candidate_binding(repo, tag)
+        manifest, rendered_binding = verify_candidate_run(
+            repo,
+            repository=repository,
+            candidate_run=tag_binding["run_id"],
+            version=tag.removeprefix("v"),
+            release_id=None,
+            source_sha=source_sha,
+        )
+        if release_candidate.parse_tag_binding(rendered_binding) != tag_binding:
+            raise ReleasePlanError("annotated tag binding differs from the candidate attempt")
+        protected_main = refresh_protected_main(repo, repository)
+        workflow_revision = resolve_commit(
+            repo,
+            manifest["workflow"]["revision"],
+            "candidate workflow revision",
+        )
+        validate_candidate_ancestry(
+            repo,
+            source_sha=source_sha,
+            workflow_revision=workflow_revision,
+            protected_main_sha=protected_main,
+        )
+        release_state = validate_recovery_release(
+            release,
+            tag=tag,
+            manifest_sha256=tag_binding["manifest_sha256"],
+        )
+        retry_command = (
+            "gh workflow run release.yml "
+            f"--repo {shlex.quote(repository)} --ref main -f {shlex.quote(f'tag={tag}')}"
+        )
+        result = {
+            "operation": "verify-recovery",
+            "status": "verified",
+            "tag": tag,
+            "source_sha": source_sha,
+            "candidate_run": tag_binding["run_id"],
+            "release_state": release_state,
+            "retry_command": retry_command,
+        }
+        print(json.dumps(result, indent=2, sort_keys=True))
+        print(
+            "The exact tag and candidate are valid for a fail-closed release "
+            "workflow retry:\n"
+            f"{retry_command}",
+            file=sys.stderr,
+        )
+    except (
+        OSError,
+        UnicodeError,
+        json.JSONDecodeError,
+        ReleasePlanError,
+        release_candidate.CandidateError,
+        verify_public_release.PublicReleaseError,
+    ) as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 1
+    return 0
+
+
 def verify_candidate_for_tag(
     repo: Path,
     *,
@@ -2632,6 +2915,11 @@ def main() -> int:
         action="store_true",
         help="wait for the exact dispatched candidate run to complete",
     )
+    candidate_parser.add_argument(
+        "--verbose-wait",
+        action="store_true",
+        help="show raw gh run watch output instead of compact state changes",
+    )
     verify_candidate_parser = subparsers.add_parser("verify-candidate")
     verify_candidate_parser.add_argument("--version", required=True)
     verify_candidate_parser.add_argument("--release-id", required=True)
@@ -2648,6 +2936,12 @@ def main() -> int:
     verify_public_parser.add_argument(
         "--repository", default="registrystack/registry-stack"
     )
+    recovery_parser = subparsers.add_parser("verify-recovery")
+    recovery_parser.add_argument("--tag", required=True)
+    recovery_parser.add_argument("--repo", type=Path, default=ROOT)
+    recovery_parser.add_argument(
+        "--repository", default="registrystack/registry-stack"
+    )
     args = parser.parse_args()
 
     if args.command == "validate":
@@ -2682,6 +2976,7 @@ def main() -> int:
             print_request=args.print_request,
             wait_for_ci=args.wait_for_ci,
             wait_for_candidate=args.wait,
+            verbose_wait=args.verbose_wait,
         )
     if args.command == "verify-candidate":
         return verify_candidate_for_tag(
@@ -2712,6 +3007,12 @@ def main() -> int:
             print(f"error: {exc}", file=sys.stderr)
             return 1
         return 0
+    if args.command == "verify-recovery":
+        return verify_release_recovery(
+            args.repo.resolve(),
+            tag=args.tag,
+            repository=args.repository,
+        )
     raise AssertionError(args.command)
 
 
diff --git a/release/scripts/test_registry_release.py b/release/scripts/test_registry_release.py
index 01038ee25..04ee57c15 100755
--- a/release/scripts/test_registry_release.py
+++ b/release/scripts/test_registry_release.py
@@ -271,8 +271,338 @@ def test_wait_for_ci_watches_only_the_exact_source_run(self) -> None:
             "registrystack/registry-stack",
             77,
             "protected-main CI",
+            verbose=False,
         )
 
+    def test_compact_wait_finds_approval_while_run_is_in_progress(self) -> None:
+        registry_release = load_registry_release()
+        url = "https://github.com/registrystack/registry-stack/actions/runs/77"
+        queued = {"id": 77, "html_url": url, "status": "queued", "conclusion": None}
+        in_progress = {**queued, "status": "in_progress"}
+        passed = {**queued, "status": "completed", "conclusion": "success"}
+        with (
+            mock.patch.object(
+                registry_release,
+                "workflow_run",
+                side_effect=[queued, queued, in_progress, passed],
+            ),
+            mock.patch.object(
+                registry_release,
+                "pending_deployments",
+                side_effect=[[], [], [{"environment": {"name": "npm"}}]],
+            ),
+            mock.patch.object(registry_release.time, "sleep"),
+            redirect_stdout(io.StringIO()) as output,
+        ):
+            registry_release.watch_workflow_run(
+                "registrystack/registry-stack",
+                77,
+                "release",
+            )
+
+        text = output.getvalue()
+        self.assertEqual(1, text.count("release run 77: queued"))
+        self.assertEqual(1, text.count(url))
+        self.assertIn("release run 77: in_progress", text)
+        self.assertIn("pending protected-environment approval for npm", text)
+        self.assertIn("authorized reviewer", text)
+        self.assertIn("pending_deployments", text)
+        self.assertIn("release run 77: completed/success", text)
+
+    def test_verbose_wait_uses_raw_gh_watcher(self) -> None:
+        registry_release = load_registry_release()
+        with (
+            mock.patch.object(
+                registry_release.subprocess,
+                "run",
+                return_value=subprocess.CompletedProcess([], 0),
+            ) as run,
+            mock.patch.object(registry_release, "workflow_run") as compact,
+        ):
+            registry_release.watch_workflow_run(
+                "registrystack/registry-stack",
+                77,
+                "release",
+                verbose=True,
+            )
+
+        compact.assert_not_called()
+        self.assertEqual(
+            [
+                "gh",
+                "run",
+                "watch",
+                "77",
+                "--repo",
+                "registrystack/registry-stack",
+                "--exit-status",
+            ],
+            run.call_args.args[0],
+        )
+
+    def test_recovery_draft_must_keep_the_candidate_binding(self) -> None:
+        registry_release = load_registry_release()
+        manifest_sha = "b" * 64
+        marker = f"registry-stack-release-candidate-v2 manifest_sha256:{manifest_sha}"
+        release = {
+            "tag_name": "v1.2.3",
+            "name": "RegistryStack v1.2.3",
+            "prerelease": False,
+            "draft": True,
+            "published_at": None,
+            "body": marker,
+        }
+        self.assertEqual(
+            "draft",
+            registry_release.validate_recovery_release(
+                release,
+                tag="v1.2.3",
+                manifest_sha256=manifest_sha,
+            ),
+        )
+
+        release["body"] = "different candidate"
+        with self.assertRaisesRegex(
+            registry_release.ReleasePlanError,
+            "not bound to this candidate",
+        ):
+            registry_release.validate_recovery_release(
+                release,
+                tag="v1.2.3",
+                manifest_sha256=manifest_sha,
+            )
+
+    def test_recovery_requires_the_local_tag_to_match_origin(self) -> None:
+        registry_release = load_registry_release()
+        reference = "refs/tags/v1.2.3"
+        local_object = "a" * 40
+        remote_object = "b" * 40
+        source = "c" * 40
+
+        def checked(command, **_kwargs):
+            if command == ["git", "cat-file", "-t", reference]:
+                return "tag\n"
+            if command == ["git", "rev-parse", reference]:
+                return f"{local_object}\n"
+            if command[:3] == ["git", "ls-remote", "--tags"]:
+                return (
+                    f"{remote_object}\t{reference}\n"
+                    f"{source}\t{reference}^{{}}\n"
+                )
+            raise AssertionError(command)
+
+        with self.assertRaisesRegex(
+            registry_release.ReleasePlanError,
+            "does not exactly match origin",
+        ):
+            with (
+                mock.patch.object(registry_release, "run_checked", side_effect=checked),
+                mock.patch.object(
+                    registry_release,
+                    "resolve_commit",
+                    return_value=source,
+                ),
+            ):
+                registry_release.tagged_candidate_binding(ROOT, "v1.2.3")
+
+    def test_recovery_verification_emits_exact_retry_command(self) -> None:
+        registry_release = load_registry_release()
+        manifest_sha = "b" * 64
+        source_sha = "a" * 40
+        workflow_revision = "c" * 40
+        protected_main = "d" * 40
+        binding = registry_release.release_candidate.render_tag_binding(
+            77,
+            1,
+            manifest_sha,
+        )
+        with (
+            mock.patch.object(registry_release, "verify_origin_repository"),
+            mock.patch.object(
+                registry_release,
+                "tagged_candidate_binding",
+                return_value=(
+                    source_sha,
+                    registry_release.release_candidate.parse_tag_binding(binding),
+                ),
+            ),
+            mock.patch.object(registry_release, "release_for_tag", return_value=None),
+            mock.patch.object(
+                registry_release,
+                "verify_candidate_run",
+                return_value=(
+                    {
+                        "release": {"version": "0.22.0"},
+                        "workflow": {"revision": workflow_revision},
+                    },
+                    binding,
+                ),
+            ),
+            mock.patch.object(
+                registry_release,
+                "refresh_protected_main",
+                return_value=protected_main,
+            ),
+            mock.patch.object(
+                registry_release,
+                "resolve_commit",
+                return_value=workflow_revision,
+            ),
+            mock.patch.object(
+                registry_release,
+                "validate_candidate_ancestry",
+            ) as validate_ancestry,
+            redirect_stdout(io.StringIO()) as output,
+            redirect_stderr(io.StringIO()) as errors,
+        ):
+            result = registry_release.verify_release_recovery(
+                ROOT,
+                tag="v0.22.0",
+                repository="registrystack/registry-stack",
+            )
+
+        self.assertEqual(0, result)
+        expected = (
+            "gh workflow run release.yml --repo registrystack/registry-stack "
+            "--ref main -f tag=v0.22.0"
+        )
+        self.assertIn(expected, output.getvalue())
+        self.assertIn(expected, errors.getvalue())
+        validate_ancestry.assert_called_once_with(
+            ROOT,
+            source_sha=source_sha,
+            workflow_revision=workflow_revision,
+            protected_main_sha=protected_main,
+        )
+
+    def test_recovery_verification_rejects_unreachable_candidate_source(self) -> None:
+        registry_release = load_registry_release()
+        source_sha = "a" * 40
+        workflow_revision = "c" * 40
+        protected_main = "d" * 40
+        manifest_sha = "b" * 64
+        binding = registry_release.release_candidate.render_tag_binding(
+            77,
+            1,
+            manifest_sha,
+        )
+        with (
+            mock.patch.object(registry_release, "verify_origin_repository"),
+            mock.patch.object(registry_release, "release_for_tag", return_value=None),
+            mock.patch.object(
+                registry_release,
+                "tagged_candidate_binding",
+                return_value=(
+                    source_sha,
+                    registry_release.release_candidate.parse_tag_binding(binding),
+                ),
+            ),
+            mock.patch.object(
+                registry_release,
+                "verify_candidate_run",
+                return_value=(
+                    {
+                        "release": {"version": "0.22.0"},
+                        "workflow": {"revision": workflow_revision},
+                    },
+                    binding,
+                ),
+            ),
+            mock.patch.object(
+                registry_release,
+                "refresh_protected_main",
+                return_value=protected_main,
+            ),
+            mock.patch.object(
+                registry_release,
+                "resolve_commit",
+                return_value=workflow_revision,
+            ),
+            mock.patch.object(
+                registry_release,
+                "validate_candidate_ancestry",
+                side_effect=registry_release.ReleasePlanError(
+                    f"candidate source {source_sha} is not reachable from protected main"
+                ),
+            ),
+            redirect_stdout(io.StringIO()) as output,
+            redirect_stderr(io.StringIO()) as errors,
+        ):
+            result = registry_release.verify_release_recovery(
+                ROOT,
+                tag="v0.22.0",
+                repository="registrystack/registry-stack",
+            )
+
+        self.assertEqual(1, result)
+        self.assertNotIn("workflow run", output.getvalue())
+        self.assertIn("not reachable from protected main", errors.getvalue())
+
+    def test_recovery_verification_rejects_tags_publication_cannot_dispatch(
+        self,
+    ) -> None:
+        registry_release = load_registry_release()
+        for tag, expected_error in (
+            ("v1.2.3", "Beta publication accepts only v0.x.y release tags"),
+            ("v0.18.0", "pre-v0.19 releases are immutable historical evidence"),
+        ):
+            with self.subTest(tag=tag):
+                with (
+                    mock.patch.object(registry_release, "verify_origin_repository"),
+                    mock.patch.object(
+                        registry_release,
+                        "release_for_tag",
+                        return_value=None,
+                    ),
+                    mock.patch.object(
+                        registry_release,
+                        "tagged_candidate_binding",
+                    ) as candidate,
+                    redirect_stdout(io.StringIO()),
+                    redirect_stderr(io.StringIO()) as errors,
+                ):
+                    result = registry_release.verify_release_recovery(
+                        ROOT,
+                        tag=tag,
+                        repository="registrystack/registry-stack",
+                    )
+
+                self.assertEqual(1, result)
+                candidate.assert_not_called()
+                self.assertIn(expected_error, errors.getvalue())
+
+    def test_published_recovery_routes_to_public_verification_without_retry(self) -> None:
+        registry_release = load_registry_release()
+        public = {"tag": "v1.2.3", "status": "verified"}
+        with (
+            mock.patch.object(registry_release, "verify_origin_repository"),
+            mock.patch.object(
+                registry_release,
+                "release_for_tag",
+                return_value={"draft": False},
+            ),
+            mock.patch.object(
+                registry_release.verify_public_release,
+                "verify",
+                return_value=public,
+            ) as verify_public,
+            mock.patch.object(registry_release, "verify_candidate_run") as candidate,
+            redirect_stdout(io.StringIO()) as output,
+            redirect_stderr(io.StringIO()) as errors,
+        ):
+            result = registry_release.verify_release_recovery(
+                ROOT,
+                tag="v1.2.3",
+                repository="registrystack/registry-stack",
+            )
+
+        self.assertEqual(0, result)
+        verify_public.assert_called_once()
+        candidate.assert_not_called()
+        self.assertIn('"status": "complete"', output.getvalue())
+        self.assertNotIn("workflow run", output.getvalue())
+        self.assertIn("do not retry", errors.getvalue())
+
     def test_candidate_ancestry_accepts_main_advancement_and_rejects_unreachable_source(
         self,
     ) -> None: