Reconcile convergence with current controls

GitHub reruns preserve the failed workflow SHA, so a repaired convergence control cannot resolve an already committed release. A missed workflow_run event can also leave mutable aliases stranded without another attempt.

Change-source: pulse-maintainer

.github/workflows/retry-release-convergence.yml

@@ -1,8 +1,9 @@
 name: Retry Release Convergence
-run-name: Retry release convergence run ${{ github.event.workflow_run.id }} attempt ${{ github.event.workflow_run.run_attempt }}
+run-name: Reconcile release convergence debt
 
 # A committed release is never returned to draft because a mutable surface is
-# temporarily unavailable. Re-run the complete convergence workflow so it
+# temporarily unavailable. Re-run current controls, or dispatch them afresh
+# when the failed run predates a repair, so the complete convergence workflow
 # reacquires the global lease and replays every idempotent surface. Successful
 # surfaces are safe to repeat; retrying only failed jobs would bypass the lease.
 on:
@@ -13,80 +14,54 @@ on:
       - completed
     branches:
       - main
+  schedule:
+    # workflow_run delivery is not durable. Reconcile the current stable and
+    # preview channel heads hourly so a dropped event cannot strand aliases.
+    - cron: '37 * * * *'
+  workflow_dispatch:
 
 permissions:
   actions: write
   contents: read
 
+concurrency:
+  group: retry-release-convergence
+  cancel-in-progress: false
+
 jobs:
   retry_committed_convergence:
-    if: ${{ github.event.workflow_run.conclusion != 'success' }}
+    if: ${{ github.event_name == 'workflow_run' && github.event.workflow_run.conclusion != 'success' }}
     runs-on: ubuntu-24.04
     timeout-minutes: 10
     steps:
-      - name: Re-run the complete committed convergence
+      - name: Checkout current reconciliation controls
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+
+      - name: Reconcile the failed convergence
         env:
           GH_TOKEN: ${{ github.token }}
           RUN_ID: ${{ github.event.workflow_run.id }}
-          RUN_ATTEMPT: ${{ github.event.workflow_run.run_attempt }}
-          DISPLAY_TITLE: ${{ github.event.workflow_run.display_title }}
-          MAX_ATTEMPTS: "5"
-        run: |
-          set -euo pipefail
-          if [[ ! "${DISPLAY_TITLE}" =~ ^Release\ convergence\ (v[^[:space:]]+)\ source\ ([0-9]+)$ ]]; then
-            echo "::error::Cannot derive the convergence tag from display title ${DISPLAY_TITLE}."
-            exit 1
-          fi
-          tag="${BASH_REMATCH[1]}"
-          source_release_run_id="${BASH_REMATCH[2]}"
-          marker_url="https://github.com/${{ github.repository }}/releases/download/${tag}/release-activation.json"
-          release_state="$(
-            gh api "repos/${{ github.repository }}/releases/tags/${tag}" \
-              --jq '[(.draft | tostring), (.published_at // "")] | @tsv' \
-              2>/dev/null || true
-          )"
-          is_draft="$(awk -F '\t' '{print $1}' <<<"${release_state}")"
-          published_at="$(awk -F '\t' '{print $2}' <<<"${release_state}")"
-          marker="$(mktemp)"
-          committed=false
-          if [ "${is_draft}" = "false" ] && [ -n "${published_at}" ] && \
-             curl -fsSL --retry 2 --retry-delay 2 --retry-all-errors \
-               -o "${marker}" "${marker_url}" && \
-             jq -e \
-               --arg tag "${tag}" \
-               --arg source_release_run_id "${source_release_run_id}" \
-               --arg convergence_run_id "${RUN_ID}" \
-               '.schema_version == 1 and .tag == $tag and .source_release_run_id == $source_release_run_id and .convergence_run_id == $convergence_run_id' \
-               "${marker}" >/dev/null; then
-            committed=true
-          fi
-          if [ "${committed}" != "true" ]; then
-            rm -f "${marker}"
-            source_status="$(
-              gh api "repos/${{ github.repository }}/actions/runs/${source_release_run_id}" \
-                --jq '.status' 2>/dev/null || true
-            )"
-            if [ "${source_status}" != "completed" ] && (( RUN_ATTEMPT < 50 )); then
-              echo "Activation source run ${source_release_run_id} remains ${source_status:-unknown}; renewing the pre-commit convergence owner."
-              gh api \
-                --method POST \
-                -H "Accept: application/vnd.github+json" \
-                -H "X-GitHub-Api-Version: 2026-03-10" \
-                "repos/${{ github.repository }}/actions/runs/${RUN_ID}/rerun"
-              exit 0
-            fi
-            echo "${tag} never crossed the irreversible activation commit and source run ${source_release_run_id} is ${source_status:-unavailable}; convergence will not retry."
-            exit 0
-          fi
-          rm -f "${marker}"
+        run: >-
+          python3 scripts/release_control/reconcile_release_convergence.py
+          --repository "${GITHUB_REPOSITORY}"
+          --run-id "${RUN_ID}"
 
-          if (( RUN_ATTEMPT >= MAX_ATTEMPTS )); then
-            echo "::error::Committed release ${tag} still has convergence debt after ${RUN_ATTEMPT} attempts. Re-run the complete convergence workflow after repairing the failing surface."
-            exit 1
-          fi
-          echo "Re-running complete convergence run ${RUN_ID} for committed ${tag}; next attempt is $((RUN_ATTEMPT + 1))/${MAX_ATTEMPTS}."
-          gh api \
-            --method POST \
-            -H "Accept: application/vnd.github+json" \
-            -H "X-GitHub-Api-Version: 2026-03-10" \
-            "repos/${{ github.repository }}/actions/runs/${RUN_ID}/rerun"
+  reconcile_missed_event:
+    if: ${{ github.event_name != 'workflow_run' }}
+    runs-on: ubuntu-24.04
+    timeout-minutes: 10
+    steps:
+      - name: Checkout current reconciliation controls
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+
+      - name: Reconcile current channel heads
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: >-
+          python3 scripts/release_control/reconcile_release_convergence.py
+          --repository "${GITHUB_REPOSITORY}"
+          --latest

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

@@ -1847,7 +1847,7 @@ artifact-selection behaviour.
    the full workflow is retried so it reacquires the lease and safely replays
    all idempotent surfaces.
    The activation marker preserves its original convergence run ID. Once that
-   run is completed, a fresh manual convergence dispatch from fixed workflow
+   run is completed, a fresh successor convergence dispatch from fixed workflow
    code may adopt the same immutable tag, source run, target commit, release ID,
    R2 prefix, and activation-marker digest. Adoption happens only after lease
    acquisition and writes a unique owner record into the exact lease commit.
@@ -1858,6 +1858,21 @@ artifact-selection behaviour.
    publication, while every successor gets immutable, run-scoped Git evidence.
    Reading a floating ref or a clobbered constant record is forbidden because
    cached prior bytes could authorize stale ownership.
+   Convergence retry is a current-control reconciliation, not unconditional
+   replay of the failed run's checkout. A terminal failure delivered through
+   `workflow_run` must be reconciled, and an hourly or manual pass must inspect
+   the current immutable stable and preview channel heads so a missed event
+   cannot strand mutable aliases. Discovery must not fall back behind a mutable
+   channel head. Before mutation, reconciliation must bind the exact mainline
+   workflow-dispatch identity, release, source run, activation marker and
+   original convergence owner; reject malformed marker digests; suppress a
+   newer matching run; and enforce one aggregate attempt budget across reruns
+   and successor runs. If the failed run used the current default-branch commit,
+   the complete run is rerun. If its checkout predates current controls, a fresh
+   default-branch convergence run is dispatched using only immutable identity
+   inputs recovered from the activation marker. Pre-commit owner renewal remains
+   limited to the original run while its exact source release run is active and
+   does not consume the post-commit convergence-debt budget.
    A support-only private Pro prerelease image is a narrower exception for
    customer verification of an already-fixed defect. It may dispatch the private
    `Build Pro Release` workflow with `publish_docker_image=true`,

scripts/release_control/reconcile_release_convergence.py

@@ -0,0 +1,510 @@
+#!/usr/bin/env python3
+"""Reconcile a failed release convergence run without replaying stale controls."""
+
+from __future__ import annotations
+
+import argparse
+from datetime import datetime
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+import tempfile
+from typing import Any, Iterable
+
+
+CONVERGENCE_PATH = ".github/workflows/release-convergence.yml"
+CREATE_RELEASE_PATH = ".github/workflows/create-release.yml"
+DISPLAY_TITLE = re.compile(r"^Release convergence (v[^\s]+) source ([1-9][0-9]*)$")
+RELEASE_TAG = re.compile(
+    r"^v(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)"
+    r"(?:-(?:alpha|beta|rc)\.[1-9][0-9]*)?$"
+)
+DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$")
+EXACT_SHA = re.compile(r"^[0-9a-f]{40}$")
+TERMINAL_FAILURES = {
+    "action_required",
+    "cancelled",
+    "failure",
+    "stale",
+    "startup_failure",
+    "timed_out",
+}
+
+
+class ReconciliationError(ValueError):
+    """The remote evidence is incomplete or inconsistent."""
+
+
+class GitHubNotFound(ReconciliationError):
+    """GitHub returned a definite 404 for an object that may not exist yet."""
+
+
+def positive_int(value: object, subject: str) -> int:
+    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
+        raise ReconciliationError(f"{subject} is not a positive integer")
+    return value
+
+
+def timestamp(value: object, subject: str) -> datetime:
+    if not isinstance(value, str):
+        raise ReconciliationError(f"{subject} has no timestamp")
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise ReconciliationError(f"{subject} has an invalid timestamp") from exc
+    if parsed.tzinfo is None:
+        raise ReconciliationError(f"{subject} timestamp has no timezone")
+    return parsed
+
+
+def require_run_identity(
+    run: dict[str, Any],
+    *,
+    repository: str,
+    workflow_id: int,
+    path: str,
+    display_title: str | None = None,
+) -> None:
+    checks = {
+        "repository": run.get("repository", {}).get("full_name") == repository,
+        "workflow": run.get("workflow_id") == workflow_id,
+        "workflow path": run.get("path") == path,
+        "event": run.get("event") == "workflow_dispatch",
+    }
+    if display_title is not None:
+        checks["display title"] = run.get("display_title") == display_title
+    failed = [name for name, valid in checks.items() if not valid]
+    if failed:
+        raise ReconciliationError("run has mismatched " + ", ".join(failed))
+
+
+def validate_marker(
+    marker: object,
+    *,
+    release: dict[str, Any],
+    tag: str,
+    source_run_id: int,
+) -> int:
+    if not isinstance(marker, dict):
+        raise ReconciliationError("activation marker is not an object")
+    release_id = positive_int(release.get("id"), "release ID")
+    source_sha = release.get("target_commitish")
+    if not isinstance(source_sha, str) or EXACT_SHA.fullmatch(source_sha) is None:
+        raise ReconciliationError("release target is not an exact commit")
+    checks = {
+        "schema": marker.get("schema_version") == 1,
+        "tag": marker.get("tag") == tag,
+        "release ID": marker.get("release_id") == str(release_id),
+        "source commit": marker.get("target_commitish") == source_sha,
+        "source release run": marker.get("source_release_run_id") == str(source_run_id),
+        "R2 prefix": isinstance(marker.get("r2_prefix"), str),
+        "server digest": bool(DIGEST.fullmatch(str(marker.get("server_image_digest", "")))),
+        "control-plane digest": bool(
+            DIGEST.fullmatch(str(marker.get("control_plane_image_digest", "")))
+        ),
+        "Helm digest": bool(DIGEST.fullmatch(str(marker.get("helm_chart_digest", "")))),
+    }
+    failed = [name for name, valid in checks.items() if not valid]
+    if failed:
+        raise ReconciliationError("activation marker has mismatched " + ", ".join(failed))
+    owner = marker.get("convergence_run_id")
+    if not isinstance(owner, str) or not owner.isdigit() or int(owner) <= 0:
+        raise ReconciliationError("activation marker has an invalid convergence owner")
+    return int(owner)
+
+
+def latest_failed_runs(
+    releases: Iterable[object],
+    runs: Iterable[object],
+    *,
+    workflow_id: int,
+    default_branch: str,
+) -> list[int]:
+    """Return the latest failed convergence for each current immutable channel."""
+    latest_release: dict[bool, tuple[datetime, dict[str, Any]]] = {}
+    for index, value in enumerate(releases):
+        if not isinstance(value, dict) or value.get("draft") is not False:
+            continue
+        prerelease = value.get("prerelease")
+        tag = value.get("tag_name")
+        if (
+            not isinstance(prerelease, bool)
+            or not isinstance(tag, str)
+            or RELEASE_TAG.fullmatch(tag) is None
+        ):
+            continue
+        published = timestamp(value.get("published_at"), f"release {index}")
+        if prerelease not in latest_release or published > latest_release[prerelease][0]:
+            latest_release[prerelease] = (published, value)
+
+    # Never fall back to an older release when the advertised channel head is
+    # mutable. That is continuity debt requiring a replacement, not a target
+    # whose aliases should be promoted again.
+    current_tags = {
+        str(release["tag_name"])
+        for _, release in latest_release.values()
+        if release.get("immutable") is True
+    }
+    newest: dict[str, tuple[datetime, dict[str, Any]]] = {}
+    for index, value in enumerate(runs):
+        if not isinstance(value, dict):
+            continue
+        title = value.get("display_title")
+        match = DISPLAY_TITLE.fullmatch(title) if isinstance(title, str) else None
+        if match is None or match.group(1) not in current_tags:
+            continue
+        if (
+            value.get("workflow_id") != workflow_id
+            or value.get("path") != CONVERGENCE_PATH
+            or value.get("event") != "workflow_dispatch"
+            or value.get("head_branch") != default_branch
+        ):
+            continue
+        created = timestamp(value.get("created_at"), f"convergence run {index}")
+        tag = match.group(1)
+        if tag not in newest or created > newest[tag][0]:
+            newest[tag] = (created, value)
+
+    result: list[int] = []
+    for _, run in newest.values():
+        if run.get("status") == "completed" and run.get("conclusion") in TERMINAL_FAILURES:
+            result.append(positive_int(run.get("id"), "convergence run ID"))
+    return sorted(result)
+
+
+class GitHub:
+    def __init__(self, repository: str, gh: str, *, mutate: bool = True) -> None:
+        self.repository = repository
+        self.gh = gh
+        self.mutate = mutate
+
+    def _run(self, arguments: list[str], *, output: bool = True) -> str:
+        result = subprocess.run(
+            [self.gh, *arguments],
+            check=False,
+            capture_output=output,
+            text=True,
+        )
+        if result.returncode != 0:
+            detail = result.stderr.strip().splitlines() if output else []
+            suffix = f": {detail[-1]}" if detail else ""
+            if "HTTP 404" in result.stderr:
+                raise GitHubNotFound(
+                    f"GitHub object was not found ({' '.join(arguments[:3])}){suffix}"
+                )
+            raise ReconciliationError(f"GitHub command failed ({' '.join(arguments[:3])}){suffix}")
+        return result.stdout if output else ""
+
+    def api(self, endpoint: str) -> dict[str, Any]:
+        try:
+            value = json.loads(
+                self._run(
+                    [
+                        "api",
+                        "-H",
+                        "Accept: application/vnd.github+json",
+                        "-H",
+                        "X-GitHub-Api-Version: 2026-03-10",
+                        endpoint,
+                    ]
+                )
+            )
+        except json.JSONDecodeError as exc:
+            raise ReconciliationError(f"GitHub returned invalid JSON for {endpoint}") from exc
+        if not isinstance(value, dict):
+            raise ReconciliationError(f"GitHub returned a non-object for {endpoint}")
+        return value
+
+    def pages(self, endpoint: str) -> list[object]:
+        output = self._run(["api", "--paginate", endpoint])
+        decoder = json.JSONDecoder()
+        value: list[object] = []
+        offset = 0
+        while offset < len(output):
+            while offset < len(output) and output[offset].isspace():
+                offset += 1
+            if offset == len(output):
+                break
+            try:
+                page, offset = decoder.raw_decode(output, offset)
+            except json.JSONDecodeError as exc:
+                raise ReconciliationError(
+                    f"GitHub returned invalid pages for {endpoint}"
+                ) from exc
+            value.append(page)
+        if not value:
+            raise ReconciliationError(f"GitHub returned no pages for {endpoint}")
+        return value
+
+    def post(self, endpoint: str, payload: dict[str, Any] | None = None) -> None:
+        if not self.mutate:
+            print(f"DRY RUN: POST {endpoint}")
+            return
+        arguments = [
+            "api",
+            "--method",
+            "POST",
+            "-H",
+            "Accept: application/vnd.github+json",
+            "-H",
+            "X-GitHub-Api-Version: 2026-03-10",
+            endpoint,
+        ]
+        if payload is not None:
+            arguments.extend(["--input", "-"])
+            result = subprocess.run(
+                [self.gh, *arguments],
+                input=json.dumps(payload, separators=(",", ":")),
+                check=False,
+                capture_output=True,
+                text=True,
+            )
+            if result.returncode != 0:
+                detail = result.stderr.strip().splitlines()
+                suffix = f": {detail[-1]}" if detail else ""
+                raise ReconciliationError(f"GitHub mutation failed for {endpoint}{suffix}")
+            return
+        self._run(arguments)
+
+    def download_marker(self, tag: str, directory: Path) -> dict[str, Any]:
+        self._run(
+            [
+                "release",
+                "download",
+                tag,
+                "--repo",
+                self.repository,
+                "--pattern",
+                "release-activation.json",
+                "--dir",
+                str(directory),
+            ]
+        )
+        try:
+            value = json.loads((directory / "release-activation.json").read_text())
+        except (OSError, json.JSONDecodeError) as exc:
+            raise ReconciliationError("downloaded activation marker is unreadable") from exc
+        if not isinstance(value, dict):
+            raise ReconciliationError("downloaded activation marker is not an object")
+        return value
+
+
+def flatten_pages(pages: Iterable[object], key: str | None = None) -> list[object]:
+    values: list[object] = []
+    for index, page in enumerate(pages):
+        if key is None:
+            if not isinstance(page, list):
+                raise ReconciliationError(f"GitHub page {index} is not a list")
+            values.extend(page)
+        else:
+            if not isinstance(page, dict) or not isinstance(page.get(key), list):
+                raise ReconciliationError(f"GitHub page {index} has no {key} list")
+            values.extend(page[key])
+    return values
+
+
+def reconcile(github: GitHub, run_id: int, max_attempts: int) -> None:
+    repository = github.repository
+    repository_state = github.api(f"repos/{repository}")
+    default_branch = repository_state.get("default_branch")
+    if not isinstance(default_branch, str) or not default_branch:
+        raise ReconciliationError("repository has no default branch")
+    convergence_workflow = github.api(
+        f"repos/{repository}/actions/workflows/release-convergence.yml"
+    )
+    create_workflow = github.api(f"repos/{repository}/actions/workflows/create-release.yml")
+    convergence_workflow_id = positive_int(convergence_workflow.get("id"), "workflow ID")
+    create_workflow_id = positive_int(create_workflow.get("id"), "create-release workflow ID")
+
+    run = github.api(f"repos/{repository}/actions/runs/{run_id}")
+    require_run_identity(
+        run,
+        repository=repository,
+        workflow_id=convergence_workflow_id,
+        path=CONVERGENCE_PATH,
+    )
+    if run.get("status") != "completed" or run.get("conclusion") not in TERMINAL_FAILURES:
+        print(f"Convergence run {run_id} no longer has terminal convergence debt; no action.")
+        return
+    title = run.get("display_title")
+    match = DISPLAY_TITLE.fullmatch(title) if isinstance(title, str) else None
+    if match is None:
+        raise ReconciliationError("convergence display title has an invalid identity")
+    tag, source_text = match.groups()
+    if RELEASE_TAG.fullmatch(tag) is None:
+        raise ReconciliationError("convergence display title has an invalid release tag")
+    source_run_id = int(source_text)
+
+    all_runs = flatten_pages(
+        github.pages(
+            f"repos/{repository}/actions/workflows/release-convergence.yml/runs"
+            "?event=workflow_dispatch&per_page=100"
+        ),
+        "workflow_runs",
+    )
+    matching_runs = [
+        value
+        for value in all_runs
+        if isinstance(value, dict)
+        and value.get("workflow_id") == convergence_workflow_id
+        and value.get("display_title") == title
+    ]
+    if not matching_runs:
+        raise ReconciliationError("convergence history omitted the requested run")
+    newest = max(matching_runs, key=lambda item: timestamp(item.get("created_at"), "run"))
+    if newest.get("id") != run_id:
+        print(f"A newer convergence run already owns {title}; no action.")
+        return
+    try:
+        release = github.api(f"repos/{repository}/releases/tags/{tag}")
+    except GitHubNotFound:
+        release = {}
+    committed = (
+        release.get("tag_name") == tag
+        and release.get("draft") is False
+        and release.get("immutable") is True
+        and isinstance(release.get("published_at"), str)
+        and bool(release.get("published_at"))
+    )
+    if not committed:
+        source = github.api(f"repos/{repository}/actions/runs/{source_run_id}")
+        require_run_identity(
+            source,
+            repository=repository,
+            workflow_id=create_workflow_id,
+            path=CREATE_RELEASE_PATH,
+        )
+        if source.get("status") != "completed" and positive_int(
+            run.get("run_attempt"), "run attempt"
+        ) < 50:
+            github.post(f"repos/{repository}/actions/runs/{run_id}/rerun")
+            print(f"Renewed pre-commit convergence owner {run_id} for active source {source_run_id}.")
+            return
+        print(f"{tag} has no immutable activation commit; no convergence retry was dispatched.")
+        return
+
+    attempts = sum(
+        positive_int(item.get("run_attempt"), "run attempt") for item in matching_runs
+    )
+    if attempts >= max_attempts:
+        raise ReconciliationError(
+            f"{title} still has convergence debt after {attempts} attempts"
+        )
+
+    with tempfile.TemporaryDirectory(prefix="release-convergence-") as raw:
+        marker = github.download_marker(tag, Path(raw))
+    owner_run_id = validate_marker(
+        marker, release=release, tag=tag, source_run_id=source_run_id
+    )
+    owner_run = github.api(f"repos/{repository}/actions/runs/{owner_run_id}")
+    require_run_identity(
+        owner_run,
+        repository=repository,
+        workflow_id=convergence_workflow_id,
+        path=CONVERGENCE_PATH,
+        display_title=title,
+    )
+    if owner_run.get("status") != "completed":
+        raise ReconciliationError("activation marker owner is not terminal")
+
+    default_commit = github.api(f"repos/{repository}/commits/{default_branch}").get("sha")
+    if not isinstance(default_commit, str) or EXACT_SHA.fullmatch(default_commit) is None:
+        raise ReconciliationError("default branch did not resolve to an exact commit")
+    if run.get("head_sha") == default_commit:
+        github.post(f"repos/{repository}/actions/runs/{run_id}/rerun")
+        print(f"Re-ran current-control convergence {run_id} for committed {tag}.")
+        return
+
+    prerelease = release.get("prerelease")
+    if not isinstance(prerelease, bool):
+        raise ReconciliationError("release has no prerelease classification")
+    payload = {
+        "ref": default_branch,
+        "inputs": {
+            "tag": tag,
+            "version": tag.removeprefix("v"),
+            # The workflow-dispatch API models input values as strings even
+            # when the receiving workflow declares a boolean input.
+            "prerelease": "true" if prerelease else "false",
+            "target_commitish": marker["target_commitish"],
+            "release_id": marker["release_id"],
+            "r2_prefix": marker["r2_prefix"],
+            "source_release_run_id": marker["source_release_run_id"],
+        },
+    }
+    github.post(
+        f"repos/{repository}/actions/workflows/release-convergence.yml/dispatches",
+        payload,
+    )
+    print(
+        f"Dispatched fresh convergence controls after observing {default_commit} for committed {tag}; "
+        f"the failed run used {run.get('head_sha')}."
+    )
+
+
+def discover(github: GitHub) -> list[int]:
+    repository = github.repository
+    repository_state = github.api(f"repos/{repository}")
+    default_branch = repository_state.get("default_branch")
+    if not isinstance(default_branch, str) or not default_branch:
+        raise ReconciliationError("repository has no default branch")
+    workflow = github.api(f"repos/{repository}/actions/workflows/release-convergence.yml")
+    workflow_id = positive_int(workflow.get("id"), "workflow ID")
+    releases = flatten_pages(
+        github.pages(f"repos/{repository}/releases?per_page=100")
+    )
+    runs = flatten_pages(
+        github.pages(
+            f"repos/{repository}/actions/workflows/release-convergence.yml/runs"
+            "?event=workflow_dispatch&per_page=100"
+        ),
+        "workflow_runs",
+    )
+    return latest_failed_runs(
+        releases,
+        runs,
+        workflow_id=workflow_id,
+        default_branch=default_branch,
+    )
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--repository", required=True)
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument("--run-id", type=int)
+    group.add_argument("--latest", action="store_true")
+    parser.add_argument("--max-attempts", type=int, default=5)
+    parser.add_argument("--dry-run", action="store_true")
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    if args.run_id is not None and args.run_id <= 0:
+        print("run ID must be positive", file=sys.stderr)
+        return 2
+    if not 1 <= args.max_attempts <= 20:
+        print("max attempts must be between 1 and 20", file=sys.stderr)
+        return 2
+    gh = os.environ.get("GH_BIN", "gh")
+    github = GitHub(args.repository, gh, mutate=not args.dry_run)
+    try:
+        run_ids = discover(github) if args.latest else [args.run_id]
+        if not run_ids:
+            print("No current immutable release has unattended convergence debt.")
+            return 0
+        for run_id in run_ids:
+            assert run_id is not None
+            reconcile(github, run_id, args.max_attempts)
+    except (OSError, ReconciliationError) as exc:
+        print(f"release convergence reconciliation failed: {exc}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/release_control/reconcile_release_convergence_test.py

@@ -0,0 +1,295 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import unittest
+
+import reconcile_release_convergence as subject
+
+
+WORKFLOW_ID = 42
+
+
+def release(tag: str, published_at: str, *, prerelease: bool, immutable: bool = True):
+    return {
+        "id": abs(hash(tag)),
+        "tag_name": tag,
+        "draft": False,
+        "prerelease": prerelease,
+        "immutable": immutable,
+        "published_at": published_at,
+    }
+
+
+def run(
+    run_id: int,
+    tag: str,
+    created_at: str,
+    *,
+    status: str = "completed",
+    conclusion: str = "failure",
+):
+    return {
+        "id": run_id,
+        "workflow_id": WORKFLOW_ID,
+        "path": subject.CONVERGENCE_PATH,
+        "event": "workflow_dispatch",
+        "head_branch": "main",
+        "display_title": f"Release convergence {tag} source 1234",
+        "created_at": created_at,
+        "status": status,
+        "conclusion": conclusion,
+    }
+
+
+class LatestFailedRunsTests(unittest.TestCase):
+    def select(self, releases, runs):
+        return subject.latest_failed_runs(
+            releases, runs, workflow_id=WORKFLOW_ID, default_branch="main"
+        )
+
+    def test_selects_failed_heads_of_stable_and_preview_channels(self):
+        releases = [
+            release("v6.4.1", "2026-09-01T00:00:00Z", prerelease=False),
+            release("v6.5.0-rc.1", "2026-09-02T00:00:00Z", prerelease=True),
+        ]
+        runs = [
+            run(10, "v6.4.1", "2026-09-01T00:01:00Z"),
+            run(11, "v6.5.0-rc.1", "2026-09-02T00:01:00Z"),
+        ]
+        self.assertEqual([10, 11], self.select(releases, runs))
+
+    def test_does_not_fall_back_behind_a_mutable_channel_head(self):
+        releases = [
+            release("v6.4.1", "2026-09-01T00:00:00Z", prerelease=False),
+            release(
+                "v6.4.2", "2026-09-02T00:00:00Z", prerelease=False, immutable=False
+            ),
+        ]
+        self.assertEqual([], self.select(releases, [run(10, "v6.4.1", "2026-09-01T00:01:00Z")]))
+
+    def test_ignores_newer_companion_chart_releases(self):
+        releases = [
+            release("v6.5.0-rc.1", "2026-09-02T00:00:00Z", prerelease=True),
+            release(
+                "helm-chart-6.5.0-rc.1",
+                "2026-09-02T00:01:00Z",
+                prerelease=True,
+            ),
+        ]
+        self.assertEqual(
+            [10],
+            self.select(
+                releases, [run(10, "v6.5.0-rc.1", "2026-09-02T00:02:00Z")]
+            ),
+        )
+
+    def test_newer_success_or_active_run_clears_old_debt(self):
+        releases = [release("v6.5.0-rc.1", "2026-09-02T00:00:00Z", prerelease=True)]
+        failed = run(10, "v6.5.0-rc.1", "2026-09-02T00:01:00Z")
+        success = run(
+            11,
+            "v6.5.0-rc.1",
+            "2026-09-02T00:02:00Z",
+            conclusion="success",
+        )
+        self.assertEqual([], self.select(releases, [failed, success]))
+        active = run(
+            12,
+            "v6.5.0-rc.1",
+            "2026-09-02T00:03:00Z",
+            status="in_progress",
+            conclusion="",
+        )
+        self.assertEqual([], self.select(releases, [failed, active]))
+
+    def test_ignores_wrong_workflow_branch_and_title(self):
+        releases = [release("v6.5.0-rc.1", "2026-09-02T00:00:00Z", prerelease=True)]
+        wrong_workflow = run(10, "v6.5.0-rc.1", "2026-09-02T00:01:00Z")
+        wrong_workflow["workflow_id"] = 99
+        wrong_branch = run(11, "v6.5.0-rc.1", "2026-09-02T00:02:00Z")
+        wrong_branch["head_branch"] = "release/v6.5"
+        wrong_title = run(12, "v6.5.0-rc.1", "2026-09-02T00:03:00Z")
+        wrong_title["display_title"] += " injected"
+        self.assertEqual(
+            [], self.select(releases, [wrong_workflow, wrong_branch, wrong_title])
+        )
+
+
+class MarkerTests(unittest.TestCase):
+    def test_marker_binds_release_source_and_all_customer_digests(self):
+        current_release = {
+            "id": 88,
+            "target_commitish": "a" * 40,
+        }
+        marker = {
+            "schema_version": 1,
+            "tag": "v6.5.0-rc.1",
+            "release_id": "88",
+            "target_commitish": "a" * 40,
+            "source_release_run_id": "1234",
+            "convergence_run_id": "5678",
+            "r2_prefix": "packet",
+            "server_image_digest": "sha256:" + "b" * 64,
+            "control_plane_image_digest": "sha256:" + "c" * 64,
+            "helm_chart_digest": "sha256:" + "d" * 64,
+        }
+        self.assertEqual(
+            5678,
+            subject.validate_marker(
+                marker,
+                release=current_release,
+                tag="v6.5.0-rc.1",
+                source_run_id=1234,
+            ),
+        )
+        marker["server_image_digest"] = "sha256:bad"
+        with self.assertRaisesRegex(subject.ReconciliationError, "server digest"):
+            subject.validate_marker(
+                marker,
+                release=current_release,
+                tag="v6.5.0-rc.1",
+                source_run_id=1234,
+            )
+
+
+class FakeGitHub:
+    repository = "rcourtman/Pulse"
+
+    def __init__(self, *, current_controls: bool = False, committed: bool = True):
+        self.run_id = 100
+        self.source_run_id = 200
+        self.owner_run_id = 90
+        self.title = "Release convergence v6.5.0-rc.1 source 200"
+        self.old_sha = "a" * 40
+        self.main_sha = self.old_sha if current_controls else "b" * 40
+        self.committed = committed
+        self.posts = []
+        self.runs = [self.run(self.run_id, self.old_sha)]
+        self.marker = {
+            "schema_version": 1,
+            "tag": "v6.5.0-rc.1",
+            "release_id": "300",
+            "target_commitish": "c" * 40,
+            "source_release_run_id": "200",
+            "convergence_run_id": "90",
+            "r2_prefix": "packet",
+            "server_image_digest": "sha256:" + "d" * 64,
+            "control_plane_image_digest": "sha256:" + "e" * 64,
+            "helm_chart_digest": "sha256:" + "f" * 64,
+        }
+
+    def run(self, run_id, head_sha, *, attempt=1, status="completed", conclusion="failure"):
+        return {
+            "id": run_id,
+            "workflow_id": 42,
+            "path": subject.CONVERGENCE_PATH,
+            "event": "workflow_dispatch",
+            "repository": {"full_name": self.repository},
+            "display_title": self.title,
+            "head_branch": "main",
+            "head_sha": head_sha,
+            "status": status,
+            "conclusion": conclusion,
+            "run_attempt": attempt,
+            "created_at": f"2026-09-02T00:{run_id % 60:02d}:00Z",
+        }
+
+    def api(self, endpoint):
+        if endpoint == f"repos/{self.repository}":
+            return {"default_branch": "main"}
+        suffix = endpoint.removeprefix(f"repos/{self.repository}/")
+        if suffix == "actions/workflows/release-convergence.yml":
+            return {"id": 42}
+        if suffix == "actions/workflows/create-release.yml":
+            return {"id": 43}
+        if suffix == "commits/main":
+            return {"sha": self.main_sha}
+        if suffix == "releases/tags/v6.5.0-rc.1":
+            if not self.committed:
+                return {"tag_name": "v6.5.0-rc.1", "draft": True}
+            return {
+                "id": 300,
+                "tag_name": "v6.5.0-rc.1",
+                "target_commitish": "c" * 40,
+                "draft": False,
+                "immutable": True,
+                "published_at": "2026-09-02T00:00:00Z",
+                "prerelease": True,
+            }
+        if suffix.startswith("actions/runs/"):
+            run_id = int(suffix.rsplit("/", 1)[1])
+            if run_id == self.owner_run_id:
+                owner = self.run(self.owner_run_id, self.old_sha, conclusion="success")
+                owner["status"] = "completed"
+                return owner
+            if run_id == self.source_run_id:
+                return {
+                    "id": self.source_run_id,
+                    "workflow_id": 43,
+                    "path": subject.CREATE_RELEASE_PATH,
+                    "event": "workflow_dispatch",
+                    "repository": {"full_name": self.repository},
+                    "status": "in_progress",
+                }
+            return next(value for value in self.runs if value["id"] == run_id)
+        raise AssertionError(endpoint)
+
+    def pages(self, endpoint):
+        self.assert_endpoint = endpoint
+        return [{"workflow_runs": self.runs}]
+
+    def download_marker(self, tag, directory):
+        self.download = (tag, directory)
+        return self.marker
+
+    def post(self, endpoint, payload=None):
+        self.posts.append((endpoint, payload))
+
+
+class ReconciliationTests(unittest.TestCase):
+    def test_stale_failed_run_dispatches_current_controls_with_bound_inputs(self):
+        github = FakeGitHub()
+        subject.reconcile(github, github.run_id, 5)
+        self.assertEqual(1, len(github.posts))
+        endpoint, payload = github.posts[0]
+        self.assertTrue(endpoint.endswith("release-convergence.yml/dispatches"))
+        self.assertEqual("main", payload["ref"])
+        self.assertEqual("v6.5.0-rc.1", payload["inputs"]["tag"])
+        self.assertEqual("200", payload["inputs"]["source_release_run_id"])
+        self.assertEqual("true", payload["inputs"]["prerelease"])
+
+    def test_current_failed_run_uses_exact_rerun(self):
+        github = FakeGitHub(current_controls=True)
+        subject.reconcile(github, github.run_id, 5)
+        self.assertEqual(
+            [(f"repos/{github.repository}/actions/runs/{github.run_id}/rerun", None)],
+            github.posts,
+        )
+
+    def test_newer_sibling_prevents_duplicate_mutation(self):
+        github = FakeGitHub()
+        github.runs.append(github.run(101, github.main_sha, status="in_progress", conclusion=""))
+        subject.reconcile(github, github.run_id, 5)
+        self.assertEqual([], github.posts)
+
+    def test_attempt_budget_counts_distinct_runs_and_reruns(self):
+        github = FakeGitHub()
+        github.runs[0]["run_attempt"] = 3
+        github.runs.append(github.run(101, github.old_sha, attempt=2))
+        with self.assertRaisesRegex(subject.ReconciliationError, "after 5 attempts"):
+            subject.reconcile(github, 101, 5)
+        self.assertEqual([], github.posts)
+
+    def test_precommit_owner_is_renewed_without_using_committed_budget(self):
+        github = FakeGitHub(committed=False)
+        github.runs[0]["run_attempt"] = 6
+        subject.reconcile(github, github.run_id, 5)
+        self.assertEqual(
+            [(f"repos/{github.repository}/actions/runs/{github.run_id}/rerun", None)],
+            github.posts,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/release_control/release_promotion_policy_test.py

@@ -585,6 +585,9 @@ class ReleasePromotionPolicyTest(unittest.TestCase):
         release_workflow = read(".github/workflows/create-release.yml")
         convergence = read(".github/workflows/release-convergence.yml")
         retry = read(".github/workflows/retry-release-convergence.yml")
+        reconciler = read(
+            "scripts/release_control/reconcile_release_convergence.py"
+        )
         commit_verdict = workflow_job_block(release_workflow, "release_commit_verdict")
         convergence_verdict = workflow_job_block(convergence, "convergence_verdict")
 
@@ -612,10 +615,16 @@ class ReleasePromotionPolicyTest(unittest.TestCase):
                 self.assertIn("the committed release remains public", convergence_verdict)
 
         self.assertIn("github.event.workflow_run.conclusion != 'success'", retry)
-        self.assertIn("actions/runs/${RUN_ID}/rerun", retry)
+        self.assertIn("cron: '37 * * * *'", retry)
+        self.assertIn("workflow_dispatch:", retry)
+        self.assertIn("reconcile_release_convergence.py", retry)
+        self.assertIn('--run-id "${RUN_ID}"', retry)
+        self.assertIn("--latest", retry)
         self.assertNotIn("rerun-failed-jobs", retry)
-        self.assertIn("RUN_ATTEMPT >= MAX_ATTEMPTS", retry)
-        self.assertIn("release-activation.json", retry)
+        self.assertIn('actions/runs/{run_id}/rerun', reconciler)
+        self.assertIn('release-convergence.yml/dispatches', reconciler)
+        self.assertIn("attempts >= max_attempts", reconciler)
+        self.assertIn("validate_marker(", reconciler)
 
     def test_mutating_reusable_workflows_have_no_direct_dispatch_lock_bypass(self) -> None:
         convergence = read(".github/workflows/release-convergence.yml")
@@ -805,6 +814,9 @@ class ReleasePromotionPolicyTest(unittest.TestCase):
         release_workflow = read(".github/workflows/create-release.yml")
         convergence = read(".github/workflows/release-convergence.yml")
         retry = read(".github/workflows/retry-release-convergence.yml")
+        reconciler = read(
+            "scripts/release_control/reconcile_release_convergence.py"
+        )
         activation = workflow_job_block(release_workflow, "activate_release")
 
         self.assertIn(
@@ -830,11 +842,13 @@ class ReleasePromotionPolicyTest(unittest.TestCase):
         self.assertIn("successor adoption is not yet allowed", await_commit)
         self.assertIn("activation_marker_sha256", await_commit)
 
-        self.assertIn("source_release_run_id=\"${BASH_REMATCH[2]}\"", retry)
-        self.assertIn("actions/runs/${source_release_run_id}", retry)
-        self.assertIn('source_status}" != "completed"', retry)
-        self.assertIn("RUN_ATTEMPT < 50", retry)
-        self.assertIn("renewing the pre-commit convergence owner", retry)
+        self.assertIn("--run-id", retry)
+        self.assertIn("DISPLAY_TITLE = re.compile", reconciler)
+        self.assertIn("source_run_id = int(source_text)", reconciler)
+        self.assertIn('actions/runs/{source_run_id}', reconciler)
+        self.assertIn('source.get("status") != "completed"', reconciler)
+        self.assertIn('< 50', reconciler)
+        self.assertIn("Renewed pre-commit convergence owner", reconciler)
 
         self.assertIn("require_viable_convergence_owner()", activation)
         self.assertEqual(activation.count("require_viable_convergence_owner"), 3)