Stop retrying unchanged credential blocks

A committed release with an unchanged operator-owned containment failure cannot converge through unattended retries. Classify that evidence without weakening the block, and rearm only when the relevant private inputs or public controls change.

Change-source: pulse-maintainer

.github/workflows/canonical-governance.yml

@@ -170,6 +170,9 @@ jobs:
       - name: Run release promotion policy unit tests
         run: python3 scripts/release_control/release_promotion_policy_test.py
 
+      - name: Run release convergence reconciler unit tests
+        run: python3 scripts/release_control/reconcile_release_convergence_test.py
+
       - name: Run release container identity verifier unit tests
         run: python3 scripts/release_control/verify_release_container_images_test.py
 

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

@@ -42,6 +42,7 @@ jobs:
       - name: Reconcile the failed convergence
         env:
           GH_TOKEN: ${{ github.token }}
+          PRO_REPOSITORY_TOKEN: ${{ secrets.WORKFLOW_PAT }}
           RUN_ID: ${{ github.event.workflow_run.id }}
         run: >-
           python3 scripts/release_control/reconcile_release_convergence.py
@@ -61,6 +62,7 @@ jobs:
       - name: Reconcile current channel heads
         env:
           GH_TOKEN: ${{ github.token }}
+          PRO_REPOSITORY_TOKEN: ${{ secrets.WORKFLOW_PAT }}
         run: >-
           python3 scripts/release_control/reconcile_release_convergence.py
           --repository "${GITHUB_REPOSITORY}"

.husky/pre-commit

@@ -151,6 +151,7 @@ python3 scripts/release_control/release_promotion_policy_support_test.py
 python3 scripts/release_control/registry_audit_test.py
 python3 scripts/release_control/readiness_assertion_guard_test.py
 (cd scripts/release_control && git -C ../.. show :scripts/release_control/release_promotion_policy_test.py | python3 -)
+(cd scripts/release_control && git -C ../.. show :scripts/release_control/reconcile_release_convergence_test.py | python3 -)
 python3 scripts/release_control/repo_file_io_test.py
 python3 scripts/release_control/staged_commit_shape_guard_test.py
 python3 scripts/release_control/status_audit_test.py

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

@@ -1871,6 +1871,18 @@ artifact-selection behaviour.
    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.
+   An immutable release blocked only by the private paid-runtime credential-
+   containment gate must not consume attempts through unattended replay while
+   the failed private run's containment checker and operator checklist are
+   byte-identical to their current private-main versions. That terminal
+   classification requires authenticated private-repository evidence tying the
+   public paid-runtime job to exactly one canonical failed private promotion run,
+   exactly one failed containment job, and the explicit blocked marker in that
+   job's log. Missing, inaccessible, malformed, or ambiguous evidence remains a
+   normal fail-closed convergence failure. A change to either private
+   containment input or to the public default-branch controls rearms the bounded
+   retry budget for that control revision; it never weakens the containment gate
+   or marks the customer surface converged.
    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

@@ -18,6 +18,18 @@ 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]*)$")
+PRIVATE_PROMOTION_RUN = re.compile(
+    r"https://github\.com/rcourtman/pulse-pro/actions/runs/([1-9][0-9]*)"
+)
+PRIVATE_REPOSITORY = "rcourtman/pulse-pro"
+PRIVATE_PROMOTION_PATH = ".github/workflows/promote-paid-runtime-release.yml"
+PAID_RUNTIME_JOB = "Converge paid-runtime broker / promote"
+CREDENTIAL_CONTAINMENT_JOB = "Require credential containment"
+CREDENTIAL_BLOCK_MARKER = "credential containment gate: BLOCKED"
+CREDENTIAL_CONTAINMENT_PATHS = (
+    "scripts/check_credential_containment.py",
+    "docs/security-rotation.md",
+)
 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]*)?$"
@@ -176,17 +188,36 @@ def latest_failed_runs(
 
 
 class GitHub:
-    def __init__(self, repository: str, gh: str, *, mutate: bool = True) -> None:
+    def __init__(
+        self,
+        repository: str,
+        gh: str,
+        *,
+        mutate: bool = True,
+        private_token: str = "",
+    ) -> None:
         self.repository = repository
         self.gh = gh
         self.mutate = mutate
+        self.private_token = private_token
 
-    def _run(self, arguments: list[str], *, output: bool = True) -> str:
+    def _run(
+        self,
+        arguments: list[str],
+        *,
+        output: bool = True,
+        token: str = "",
+    ) -> str:
+        env = None
+        if token:
+            env = os.environ.copy()
+            env["GH_TOKEN"] = token
         result = subprocess.run(
             [self.gh, *arguments],
             check=False,
             capture_output=output,
             text=True,
+            env=env,
         )
         if result.returncode != 0:
             detail = result.stderr.strip().splitlines() if output else []
@@ -198,7 +229,7 @@ class GitHub:
             raise ReconciliationError(f"GitHub command failed ({' '.join(arguments[:3])}){suffix}")
         return result.stdout if output else ""
 
-    def api(self, endpoint: str) -> dict[str, Any]:
+    def api(self, endpoint: str, *, token: str = "") -> dict[str, Any]:
         try:
             value = json.loads(
                 self._run(
@@ -209,7 +240,8 @@ class GitHub:
                         "-H",
                         "X-GitHub-Api-Version: 2026-03-10",
                         endpoint,
-                    ]
+                    ],
+                    token=token,
                 )
             )
         except json.JSONDecodeError as exc:
@@ -218,8 +250,8 @@ class GitHub:
             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])
+    def pages(self, endpoint: str, *, token: str = "") -> list[object]:
+        output = self._run(["api", "--paginate", endpoint], token=token)
         decoder = json.JSONDecoder()
         value: list[object] = []
         offset = 0
@@ -291,6 +323,118 @@ class GitHub:
             raise ReconciliationError("downloaded activation marker is not an object")
         return value
 
+    def job_log(self, repository: str, job_id: int, *, token: str = "") -> str:
+        return self._run(
+            [
+                "api",
+                "-H",
+                "Accept: application/vnd.github+json",
+                "-H",
+                "X-GitHub-Api-Version: 2026-03-10",
+                f"repos/{repository}/actions/jobs/{job_id}/logs",
+            ],
+            token=token,
+        )
+
+    def unchanged_credential_containment_block(self, run_id: int) -> bool:
+        """Whether the paid-runtime failure is an unchanged operator-owned block."""
+        if not self.private_token:
+            return False
+        jobs = flatten_pages(
+            self.pages(
+                f"repos/{self.repository}/actions/runs/{run_id}/jobs?per_page=100"
+            ),
+            "jobs",
+        )
+        paid_jobs = [
+            value
+            for value in jobs
+            if isinstance(value, dict)
+            and value.get("name") == PAID_RUNTIME_JOB
+            and value.get("conclusion") == "failure"
+        ]
+        if len(paid_jobs) != 1:
+            return False
+        paid_job_id = positive_int(paid_jobs[0].get("id"), "paid-runtime job ID")
+        annotations = flatten_pages(
+            self.pages(
+                f"repos/{self.repository}/check-runs/{paid_job_id}/annotations?per_page=100"
+            )
+        )
+        private_run_ids = {
+            int(match.group(1))
+            for value in annotations
+            if isinstance(value, dict) and isinstance(value.get("message"), str)
+            for match in PRIVATE_PROMOTION_RUN.finditer(value["message"])
+        }
+        if len(private_run_ids) != 1:
+            return False
+        private_run_id = private_run_ids.pop()
+        private_run = self.api(
+            f"repos/{PRIVATE_REPOSITORY}/actions/runs/{private_run_id}",
+            token=self.private_token,
+        )
+        if (
+            private_run.get("repository", {}).get("full_name") != PRIVATE_REPOSITORY
+            or private_run.get("path") != PRIVATE_PROMOTION_PATH
+            or private_run.get("event") != "workflow_dispatch"
+            or private_run.get("status") != "completed"
+            or private_run.get("conclusion") != "failure"
+        ):
+            return False
+        private_jobs = flatten_pages(
+            self.pages(
+                f"repos/{PRIVATE_REPOSITORY}/actions/runs/{private_run_id}/jobs?per_page=100",
+                token=self.private_token,
+            ),
+            "jobs",
+        )
+        containment_jobs = [
+            value
+            for value in private_jobs
+            if isinstance(value, dict)
+            and value.get("name") == CREDENTIAL_CONTAINMENT_JOB
+            and value.get("conclusion") == "failure"
+        ]
+        if len(containment_jobs) != 1:
+            return False
+        containment_job_id = positive_int(
+            containment_jobs[0].get("id"), "credential-containment job ID"
+        )
+        containment_log = self.job_log(
+            PRIVATE_REPOSITORY, containment_job_id, token=self.private_token
+        )
+        if CREDENTIAL_BLOCK_MARKER not in containment_log:
+            return False
+        blocked_head = private_run.get("head_sha")
+        current_head = self.api(
+            f"repos/{PRIVATE_REPOSITORY}/commits/main", token=self.private_token
+        ).get("sha")
+        if (
+            not isinstance(blocked_head, str)
+            or EXACT_SHA.fullmatch(blocked_head) is None
+            or not isinstance(current_head, str)
+            or EXACT_SHA.fullmatch(current_head) is None
+        ):
+            raise ReconciliationError("private repository head is not an exact commit")
+
+        def containment_state(ref: str) -> tuple[str, ...]:
+            blobs: list[str] = []
+            for path in CREDENTIAL_CONTAINMENT_PATHS:
+                value = self.api(
+                    f"repos/{PRIVATE_REPOSITORY}/contents/{path}?ref={ref}",
+                    token=self.private_token,
+                )
+                blob = value.get("sha")
+                if value.get("type") != "file" or not isinstance(blob, str) or not blob:
+                    raise ReconciliationError(
+                        f"private credential-containment input is invalid: {path}"
+                    )
+                blobs.append(blob)
+            return tuple(blobs)
+
+        return containment_state(blocked_head) == containment_state(current_head)
+
 
 def flatten_pages(pages: Iterable[object], key: str | None = None) -> list[object]:
     values: list[object] = []
@@ -386,8 +530,20 @@ def reconcile(github: GitHub, run_id: int, max_attempts: int) -> None:
         print(f"{tag} has no immutable activation commit; no convergence retry was dispatched.")
         return
 
+    if github.unchanged_credential_containment_block(run_id):
+        print(
+            f"Convergence run {run_id} is held by unchanged private credential containment; "
+            "no unattended retry was dispatched."
+        )
+        return
+
+    # A new control revision is itself a repair candidate. Bound churn for the
+    # exact revision without allowing failures from superseded controls to make
+    # current controls permanently unretriable.
     attempts = sum(
-        positive_int(item.get("run_attempt"), "run attempt") for item in matching_runs
+        positive_int(item.get("run_attempt"), "run attempt")
+        for item in matching_runs
+        if item.get("head_sha") == run.get("head_sha")
     )
     if attempts >= max_attempts:
         raise ReconciliationError(
@@ -491,7 +647,12 @@ def main() -> int:
         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)
+    github = GitHub(
+        args.repository,
+        gh,
+        mutate=not args.dry_run,
+        private_token=os.environ.get("PRO_REPOSITORY_TOKEN", ""),
+    )
     try:
         run_ids = discover(github) if args.latest else [args.run_id]
         if not run_ids:

scripts/release_control/reconcile_release_convergence_test.py

@@ -246,6 +246,131 @@ class FakeGitHub:
     def post(self, endpoint, payload=None):
         self.posts.append((endpoint, payload))
 
+    def unchanged_credential_containment_block(self, run_id):
+        self.containment_probe = run_id
+        return False
+
+
+class CredentialContainmentTests(unittest.TestCase):
+    def github(
+        self,
+        *,
+        current_head=None,
+        containment="failure",
+        containment_log=subject.CREDENTIAL_BLOCK_MARKER,
+        containment_state_changed=False,
+    ):
+        private_run_id = 700
+        paid_job_id = 800
+        private_head = "d" * 40
+        current_head = current_head or private_head
+        github = subject.GitHub(
+            "rcourtman/Pulse", "gh", mutate=False, private_token="private-token"
+        )
+
+        def pages(endpoint, *, token=""):
+            if endpoint == (
+                "repos/rcourtman/Pulse/actions/runs/100/jobs?per_page=100"
+            ):
+                self.assertEqual("", token)
+                return [
+                    {
+                        "jobs": [
+                            {
+                                "id": paid_job_id,
+                                "name": subject.PAID_RUNTIME_JOB,
+                                "conclusion": "failure",
+                            }
+                        ]
+                    }
+                ]
+            if endpoint == (
+                f"repos/rcourtman/Pulse/check-runs/{paid_job_id}/annotations?per_page=100"
+            ):
+                self.assertEqual("", token)
+                return [
+                    [
+                        {
+                            "message": "private Pro live promotion failed: "
+                            f"https://github.com/rcourtman/pulse-pro/actions/runs/{private_run_id}"
+                        }
+                    ]
+                ]
+            if endpoint == (
+                f"repos/{subject.PRIVATE_REPOSITORY}/actions/runs/{private_run_id}/jobs?per_page=100"
+            ):
+                self.assertEqual("private-token", token)
+                return [
+                    {
+                        "jobs": [
+                            {
+                                "id": 900,
+                                "name": subject.CREDENTIAL_CONTAINMENT_JOB,
+                                "conclusion": containment,
+                            }
+                        ]
+                    }
+                ]
+            raise AssertionError(endpoint)
+
+        def api(endpoint, *, token=""):
+            self.assertEqual("private-token", token)
+            if endpoint == f"repos/{subject.PRIVATE_REPOSITORY}/actions/runs/{private_run_id}":
+                return {
+                    "repository": {"full_name": subject.PRIVATE_REPOSITORY},
+                    "path": subject.PRIVATE_PROMOTION_PATH,
+                    "event": "workflow_dispatch",
+                    "status": "completed",
+                    "conclusion": "failure",
+                    "head_sha": private_head,
+                }
+            if endpoint == f"repos/{subject.PRIVATE_REPOSITORY}/commits/main":
+                return {"sha": current_head}
+            content_prefix = f"repos/{subject.PRIVATE_REPOSITORY}/contents/"
+            if endpoint.startswith(content_prefix):
+                path, ref = endpoint.removeprefix(content_prefix).split("?ref=", 1)
+                blob = "1" * 40 if path == subject.CREDENTIAL_CONTAINMENT_PATHS[0] else "2" * 40
+                if containment_state_changed and ref == current_head:
+                    blob = "3" * 40
+                return {"type": "file", "sha": blob}
+            raise AssertionError(endpoint)
+
+        github.pages = pages
+        github.api = api
+        github.job_log = lambda repository, job_id, token="": (
+            containment_log
+            if (repository, job_id, token)
+            == (subject.PRIVATE_REPOSITORY, 900, "private-token")
+            else self.fail((repository, job_id, token))
+        )
+        return github
+
+    def test_recognises_unchanged_private_credential_block(self):
+        self.assertTrue(self.github().unchanged_credential_containment_block(100))
+
+    def test_private_change_rearms_convergence(self):
+        github = self.github(
+            current_head="e" * 40, containment_state_changed=True
+        )
+        self.assertFalse(github.unchanged_credential_containment_block(100))
+
+    def test_unrelated_private_change_does_not_rearm_convergence(self):
+        github = self.github(current_head="e" * 40)
+        self.assertTrue(github.unchanged_credential_containment_block(100))
+
+    def test_other_private_failure_remains_retriable(self):
+        github = self.github(containment="success")
+        self.assertFalse(github.unchanged_credential_containment_block(100))
+
+    def test_containment_job_error_without_block_marker_remains_retriable(self):
+        github = self.github(containment_log="checkout failed")
+        self.assertFalse(github.unchanged_credential_containment_block(100))
+
+    def test_missing_private_token_cannot_weaken_retry(self):
+        github = subject.GitHub("rcourtman/Pulse", "gh", mutate=False)
+        github.pages = lambda endpoint: self.fail(endpoint)
+        self.assertFalse(github.unchanged_credential_containment_block(100))
+
 
 class ReconciliationTests(unittest.TestCase):
     def test_stale_failed_run_dispatches_current_controls_with_bound_inputs(self):
@@ -281,6 +406,22 @@ class ReconciliationTests(unittest.TestCase):
             subject.reconcile(github, 101, 5)
         self.assertEqual([], github.posts)
 
+    def test_attempt_budget_resets_for_repaired_controls(self):
+        github = FakeGitHub()
+        github.runs[0]["run_attempt"] = 5
+        github.runs.append(github.run(101, github.main_sha))
+        subject.reconcile(github, 101, 5)
+        self.assertEqual(
+            [(f"repos/{github.repository}/actions/runs/101/rerun", None)],
+            github.posts,
+        )
+
+    def test_unchanged_credential_block_is_not_retried(self):
+        github = FakeGitHub(current_controls=True)
+        github.unchanged_credential_containment_block = lambda run_id: True
+        subject.reconcile(github, github.run_id, 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

scripts/release_control/release_promotion_policy_test.py

@@ -620,12 +620,24 @@ class ReleasePromotionPolicyTest(unittest.TestCase):
         self.assertIn("reconcile_release_convergence.py", retry)
         self.assertIn('--run-id "${RUN_ID}"', retry)
         self.assertIn("--latest", retry)
+        self.assertEqual(2, retry.count("PRO_REPOSITORY_TOKEN: ${{ secrets.WORKFLOW_PAT }}"))
         self.assertNotIn("rerun-failed-jobs", 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)
 
+        canonical = read(".github/workflows/canonical-governance.yml")
+        self.assertIn(
+            "python3 scripts/release_control/reconcile_release_convergence_test.py",
+            canonical,
+        )
+        precommit = read(".husky/pre-commit")
+        self.assertIn(
+            "show :scripts/release_control/reconcile_release_convergence_test.py",
+            precommit,
+        )
+
     def test_mutating_reusable_workflows_have_no_direct_dispatch_lock_bypass(self) -> None:
         convergence = read(".github/workflows/release-convergence.yml")
         for workflow_path, convergence_job in (