Make release continuity failures actionable

Change-source: pulse-maintainer
2026-09-10 02:25:56 +00:00 · 2026-08-31 19:31:41 +01:00
parent 86f8073c0a
commit bd146586e8
6 changed files with 743 additions and 65 deletions
@@ -180,6 +180,9 @@ jobs:
      - name: Run immutable release setting unit tests
        run: python3 scripts/release_control/check_github_release_immutability_test.py

+      - name: Run stable release continuity diagnostic unit tests
+        run: python3 scripts/release_control/release_continuity_test.py
+
      - name: Run status audit unit tests
        run: python3 scripts/release_control/status_audit_test.py

@@ -61,59 +61,56 @@ jobs:
          set -euo pipefail
          mkdir -p release-continuity-evidence
          release_json=release-continuity-evidence/release.json
+
+          if ! gh api \
+              -H 'Accept: application/vnd.github+json' \
+              -H 'X-GitHub-Api-Version: 2026-03-10' \
+              "repos/${REPOSITORY}/releases/latest" > "${release_json}"; then
+            python3 scripts/release_control/release_continuity.py release \
+              --release-json "${release_json}" \
+              --diagnostic release-continuity-evidence/release-diagnostic.json \
+              --github-output "${GITHUB_OUTPUT}" || true
+            exit 1
+          fi
+
+          python3 scripts/release_control/release_continuity.py release \
+            --release-json "${release_json}" \
+            --diagnostic release-continuity-evidence/release-diagnostic.json \
+            --github-output "${GITHUB_OUTPUT}"
+
+      - name: Bind the release activation marker
+        id: activation
+        if: ${{ steps.release.outcome == 'success' }}
+        env:
+          GH_TOKEN: ${{ github.token }}
+          REPOSITORY: ${{ github.repository }}
+          TAG: ${{ steps.release.outputs.tag }}
+        run: |
+          set -euo pipefail
+          release_json=release-continuity-evidence/release.json
          marker=release-continuity-evidence/release-activation.json

-          gh api \
-            -H 'Accept: application/vnd.github+json' \
-            -H 'X-GitHub-Api-Version: 2026-03-10' \
-            "repos/${REPOSITORY}/releases/latest" > "${release_json}"
-          if ! jq -e '
-              (.id | type == "number") and
-              (.tag_name | type == "string" and test("^v[0-9]+\\.[0-9]+\\.[0-9]+$")) and
-              (.target_commitish | type == "string" and test("^[0-9a-f]{40}$")) and
-              .draft == false and .prerelease == false and .immutable == true and
-              (.published_at | type == "string" and length > 0)
-            ' "${release_json}" >/dev/null; then
-            echo "::error::Advertised latest release failed the immutable stable identity contract. It must be a published, non-draft vX.Y.Z release locked immutable and bound to a 40-character source commit; inspect release.json in the continuity evidence."
+          if ! gh release download "${TAG}" \
+              --repo "${REPOSITORY}" \
+              --pattern release-activation.json \
+              --dir release-continuity-evidence; then
+            python3 scripts/release_control/release_continuity.py activation \
+              --release-json "${release_json}" \
+              --activation-json "${marker}" \
+              --diagnostic release-continuity-evidence/activation-diagnostic.json \
+              --github-output "${GITHUB_OUTPUT}" || true
            exit 1
          fi

-          tag="$(jq -r '.tag_name' "${release_json}")"
-          release_id="$(jq -r '.id' "${release_json}")"
-          source_sha="$(jq -r '.target_commitish' "${release_json}")"
-          gh release download "${tag}" \
-            --repo "${REPOSITORY}" \
-            --pattern release-activation.json \
-            --dir release-continuity-evidence
-
-          if ! jq -e \
-              --arg tag "${tag}" \
-              --arg release_id "${release_id}" \
-              --arg source_sha "${source_sha}" \
-              '.schema_version == 1 and .tag == $tag and
-               .release_id == $release_id and .target_commitish == $source_sha and
-               (.source_release_run_id | test("^[0-9]+$")) and
-               (.convergence_run_id | test("^[0-9]+$")) and
-               (.r2_prefix | type == "string" and length > 0) and
-               (.server_image_digest | test("^sha256:[0-9a-f]{64}$")) and
-               (.control_plane_image_digest | test("^sha256:[0-9a-f]{64}$")) and
-               (.helm_chart_digest | test("^sha256:[0-9a-f]{64}$"))' \
-              "${marker}" >/dev/null; then
-            echo "::error::release-activation.json does not bind the advertised stable tag, release, source commit, convergence runs, and published artifact digests."
-            exit 1
-          fi
-
-          echo "tag=${tag}" >> "${GITHUB_OUTPUT}"
-          echo "release_id=${release_id}" >> "${GITHUB_OUTPUT}"
-          echo "source_sha=${source_sha}" >> "${GITHUB_OUTPUT}"
-          echo "activation_sha256=$(sha256sum "${marker}" | awk '{print $1}')" >> "${GITHUB_OUTPUT}"
-          echo "server_image_digest=$(jq -r '.server_image_digest' "${marker}")" >> "${GITHUB_OUTPUT}"
-          echo "control_plane_image_digest=$(jq -r '.control_plane_image_digest' "${marker}")" >> "${GITHUB_OUTPUT}"
-          echo "helm_chart_digest=$(jq -r '.helm_chart_digest' "${marker}")" >> "${GITHUB_OUTPUT}"
+          python3 scripts/release_control/release_continuity.py activation \
+            --release-json "${release_json}" \
+            --activation-json "${marker}" \
+            --diagnostic release-continuity-evidence/activation-diagnostic.json \
+            --github-output "${GITHUB_OUTPUT}"

      - name: Verify immutable release and build provenance
        id: packet
-        if: ${{ always() && steps.release.outcome == 'success' && github.event.schedule != '17 */6 * * *' }}
+        if: ${{ always() && steps.release.outcome == 'success' && steps.activation.outcome == 'success' && github.event.schedule != '17 */6 * * *' }}
        env:
          GH_TOKEN: ${{ github.token }}
          REPOSITORY: ${{ github.repository }}
@@ -126,7 +123,7 @@ jobs:

      - name: Authenticate every published release asset
        id: assets
-        if: ${{ always() && steps.release.outcome == 'success' && github.event.schedule != '17 */6 * * *' }}
+        if: ${{ always() && steps.release.outcome == 'success' && steps.activation.outcome == 'success' && github.event.schedule != '17 */6 * * *' }}
        env:
          PULSE_UPDATE_SIGNING_PUBLIC_KEY: ${{ vars.PULSE_UPDATE_SIGNING_PUBLIC_KEY }}
          REPOSITORY: ${{ github.repository }}
@@ -135,14 +132,14 @@ jobs:

      - name: Verify exact-version container identities
        id: containers
-        if: ${{ always() && steps.release.outcome == 'success' && github.event.schedule != '17 */6 * * *' }}
+        if: ${{ always() && steps.release.outcome == 'success' && steps.activation.outcome == 'success' && github.event.schedule != '17 */6 * * *' }}
        env:
          GH_TOKEN: ${{ github.token }}
          REPOSITORY: ${{ github.repository }}
          TAG: ${{ steps.release.outputs.tag }}
          SOURCE_SHA: ${{ steps.release.outputs.source_sha }}
-          EXPECTED_SERVER_DIGEST: ${{ steps.release.outputs.server_image_digest }}
-          EXPECTED_CONTROL_PLANE_DIGEST: ${{ steps.release.outputs.control_plane_image_digest }}
+          EXPECTED_SERVER_DIGEST: ${{ steps.activation.outputs.server_image_digest }}
+          EXPECTED_CONTROL_PLANE_DIGEST: ${{ steps.activation.outputs.control_plane_image_digest }}
        run: |
          set -euo pipefail
          proof="$(./scripts/verify-release-container-images.sh "${TAG}" "${SOURCE_SHA}" "${REPOSITORY}")"
@@ -157,11 +154,11 @@ jobs:

      - name: Verify stable container discovery aliases
        id: aliases
-        if: ${{ always() && steps.release.outcome == 'success' && github.event.schedule != '17 */6 * * *' }}
+        if: ${{ always() && steps.release.outcome == 'success' && steps.activation.outcome == 'success' && github.event.schedule != '17 */6 * * *' }}
        env:
          TAG: ${{ steps.release.outputs.tag }}
-          EXPECTED_SERVER_DIGEST: ${{ steps.release.outputs.server_image_digest }}
-          EXPECTED_CONTROL_PLANE_DIGEST: ${{ steps.release.outputs.control_plane_image_digest }}
+          EXPECTED_SERVER_DIGEST: ${{ steps.activation.outputs.server_image_digest }}
+          EXPECTED_CONTROL_PLANE_DIGEST: ${{ steps.activation.outputs.control_plane_image_digest }}
          REGISTRY_OWNER: ${{ github.repository_owner }}
        run: >-
          ./scripts/verify-stable-container-aliases.sh
@@ -170,13 +167,13 @@ jobs:

      - name: Verify exact-version Helm identity
        id: helm
-        if: ${{ always() && steps.release.outcome == 'success' && github.event.schedule != '17 */6 * * *' }}
+        if: ${{ always() && steps.release.outcome == 'success' && steps.activation.outcome == 'success' && github.event.schedule != '17 */6 * * *' }}
        env:
          GH_TOKEN: ${{ github.token }}
          REPOSITORY: ${{ github.repository }}
          TAG: ${{ steps.release.outputs.tag }}
          SOURCE_SHA: ${{ steps.release.outputs.source_sha }}
-          EXPECTED_HELM_DIGEST: ${{ steps.release.outputs.helm_chart_digest }}
+          EXPECTED_HELM_DIGEST: ${{ steps.activation.outputs.helm_chart_digest }}
        run: >-
          ./scripts/verify-release-helm-chart.sh
          "${TAG}" "${SOURCE_SHA}" "${REPOSITORY}" "${EXPECTED_HELM_DIGEST}"
@@ -196,11 +193,12 @@ jobs:
          TAG: ${{ steps.release.outputs.tag }}
          RELEASE_ID: ${{ steps.release.outputs.release_id }}
          SOURCE_SHA: ${{ steps.release.outputs.source_sha }}
-          ACTIVATION_SHA256: ${{ steps.release.outputs.activation_sha256 }}
-          SERVER_IMAGE_DIGEST: ${{ steps.release.outputs.server_image_digest }}
-          CONTROL_PLANE_IMAGE_DIGEST: ${{ steps.release.outputs.control_plane_image_digest }}
-          HELM_CHART_DIGEST: ${{ steps.release.outputs.helm_chart_digest }}
+          ACTIVATION_SHA256: ${{ steps.activation.outputs.activation_sha256 }}
+          SERVER_IMAGE_DIGEST: ${{ steps.activation.outputs.server_image_digest }}
+          CONTROL_PLANE_IMAGE_DIGEST: ${{ steps.activation.outputs.control_plane_image_digest }}
+          HELM_CHART_DIGEST: ${{ steps.activation.outputs.helm_chart_digest }}
          RELEASE_RESULT: ${{ steps.release.outcome }}
+          ACTIVATION_RESULT: ${{ steps.activation.outcome }}
          PACKET_RESULT: ${{ steps.packet.outcome }}
          ASSET_RESULT: ${{ steps.assets.outcome }}
          CONTAINER_RESULT: ${{ steps.containers.outcome }}
@@ -214,7 +212,18 @@ jobs:
          if [ "${TRIGGER_SCHEDULE}" = '17 */6 * * *' ]; then
            mode=release_lock
          fi
+          for diagnostic in release activation; do
+            path="release-continuity-evidence/${diagnostic}-diagnostic.json"
+            if [ ! -s "${path}" ]; then
+              jq -n \
+                --arg check "${diagnostic}" \
+                '{schema_version: 1, check: $check, status: "not_run", identity: {}, violations: []}' \
+                > "${path}"
+            fi
+          done
          jq -n \
+            --slurpfile release_diagnostic release-continuity-evidence/release-diagnostic.json \
+            --slurpfile activation_diagnostic release-continuity-evidence/activation-diagnostic.json \
            --arg schema_version "1" \
            --arg checked_at "${checked_at}" \
            --arg repository "${REPOSITORY}" \
@@ -235,6 +244,7 @@ jobs:
            --arg control_plane_image_digest "${CONTROL_PLANE_IMAGE_DIGEST}" \
            --arg helm_chart_digest "${HELM_CHART_DIGEST}" \
            --arg release_result "${RELEASE_RESULT}" \
+            --arg activation_result "${ACTIVATION_RESULT}" \
            --arg packet_result "${PACKET_RESULT}" \
            --arg asset_result "${ASSET_RESULT}" \
            --arg container_result "${CONTAINER_RESULT}" \
@@ -265,8 +275,13 @@ jobs:
                control_plane_image_digest: $control_plane_image_digest,
                helm_chart_digest: $helm_chart_digest
              },
+              diagnostics: {
+                release_identity: $release_diagnostic[0],
+                activation_binding: $activation_diagnostic[0]
+              },
              checks: {
                release_resolution: $release_result,
+                activation_binding: $activation_result,
                immutable_packet_and_provenance: $packet_result,
                authenticated_assets: $asset_result,
                container_identities: $container_result,
@@ -253,6 +253,8 @@ scripts/release_control/*
 !scripts/release_control/relay_registration_reconnect_drain_proof.py
 !scripts/release_control/registry_audit.py
 !scripts/release_control/registry_audit_test.py
+!scripts/release_control/release_continuity.py
+!scripts/release_control/release_continuity_test.py
 !scripts/release_control/release_promotion_policy_support.py
 !scripts/release_control/release_promotion_policy_support_test.py
 !scripts/release_control/release_promotion_policy_test.py
@@ -159,25 +159,29 @@ func TestSecurityScanRevalidatesLatestStableDelivery(t *testing.T) {
 		"Latest stable release continuity",
 		"docker/setup-buildx-action@d7f5e7f509e45cec5c76c4d5afdd7de93d0b3df5",
 		`"repos/${REPOSITORY}/releases/latest"`,
-		`.draft == false and .prerelease == false and .immutable == true`,
-		"Advertised latest release failed the immutable stable identity contract",
-		`.release_id == $release_id and .target_commitish == $source_sha`,
-		"release-activation.json does not bind the advertised stable tag",
+		`scripts/release_control/release_continuity.py release`,
+		"release-diagnostic.json",
+		"Bind the release activation marker",
+		`scripts/release_control/release_continuity.py activation`,
+		"activation-diagnostic.json",
+		`steps.activation.outcome == 'success'`,
 		`./scripts/verify-github-release-integrity.sh`,
 		`./scripts/validate-published-release.sh`,
 		`PULSE_UPDATE_SIGNING_PUBLIC_KEY: ${{ vars.PULSE_UPDATE_SIGNING_PUBLIC_KEY }}`,
 		`./scripts/verify-release-container-images.sh`,
-		`EXPECTED_SERVER_DIGEST: ${{ steps.release.outputs.server_image_digest }}`,
-		`EXPECTED_CONTROL_PLANE_DIGEST: ${{ steps.release.outputs.control_plane_image_digest }}`,
+		`EXPECTED_SERVER_DIGEST: ${{ steps.activation.outputs.server_image_digest }}`,
+		`EXPECTED_CONTROL_PLANE_DIGEST: ${{ steps.activation.outputs.control_plane_image_digest }}`,
 		`./scripts/verify-stable-container-aliases.sh`,
 		`stable_container_aliases: $alias_result`,
+		`activation_binding: $activation_result`,
+		`release_identity: $release_diagnostic[0]`,
 		`CONVERGENCE_RUN_ID: ${{ github.event.workflow_run.id }}`,
 		`TRIGGER_SCHEDULE: ${{ github.event.schedule }}`,
 		`mode=release_lock`,
 		`mode: $mode`,
 		`release_convergence_run: {`,
 		`./scripts/verify-release-helm-chart.sh`,
-		`EXPECTED_HELM_DIGEST: ${{ steps.release.outputs.helm_chart_digest }}`,
+		`EXPECTED_HELM_DIGEST: ${{ steps.activation.outputs.helm_chart_digest }}`,
 		"continuity-evidence.json",
 		"retention-days: 90",
 	}
@@ -0,0 +1,458 @@
+#!/usr/bin/env python3
+"""Validate and explain the public stable-release continuity identity."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+
+STABLE_TAG = re.compile(r"^v[0-9]+\.[0-9]+\.[0-9]+$")
+SOURCE_SHA = re.compile(r"^[0-9a-f]{40}$")
+RUN_ID = re.compile(r"^[0-9]+$")
+SHA256 = re.compile(r"^sha256:[0-9a-f]{64}$")
+IMMUTABLE_REPLACEMENT_ACTION = (
+    "Do not edit the immutable release; restore the last known-good stable target if "
+    "needed, then publish a corrected replacement through convergence."
+)
+
+
+@dataclass(frozen=True)
+class Violation:
+    code: str
+    field: str
+    expected: str
+    actual: Any
+    action: str
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "code": self.code,
+            "field": self.field,
+            "expected": self.expected,
+            "actual": diagnostic_value(self.actual),
+            "action": self.action,
+        }
+
+
+RELEASE_RULES = {
+    "release_payload_invalid": (
+        "GitHub did not return one release object.",
+        "Inspect the releases/latest API response and API availability before retrying.",
+    ),
+    "release_id_invalid": (
+        "The release id is absent or malformed.",
+        "Do not activate the release; inspect how the release was created.",
+    ),
+    "stable_tag_invalid": (
+        "The advertised release is not a stable vX.Y.Z tag.",
+        "Restore the latest stable pointer to an exact stable release.",
+    ),
+    "source_identity_invalid": (
+        "The release is not bound to a full lowercase source commit.",
+        "Publish from an exact 40-character source commit.",
+    ),
+    "release_is_draft": (
+        "The advertised release is still a draft.",
+        "Keep drafts outside the stable channel until convergence completes.",
+    ),
+    "release_is_prerelease": (
+        "The advertised release is marked as a prerelease.",
+        "Keep prereleases outside the stable latest-release pointer.",
+    ),
+    "release_mutable": (
+        "GitHub reports immutable=false for the advertised stable release.",
+        "Publish a replacement through the immutable-release-gated pipeline; "
+        "never repair the packet in place.",
+    ),
+    "publication_time_invalid": (
+        "The advertised release has no publication timestamp.",
+        "Do not treat the release as activated until GitHub reports publication.",
+    ),
+}
+
+
+ACTIVATION_RULES = {
+    "activation_payload_invalid": (
+        "The activation marker is not one JSON object.",
+        IMMUTABLE_REPLACEMENT_ACTION,
+    ),
+    "activation_schema_invalid": (
+        "The activation marker schema is unsupported.",
+        IMMUTABLE_REPLACEMENT_ACTION,
+    ),
+    "activation_tag_mismatch": (
+        "The activation marker names a different release tag.",
+        IMMUTABLE_REPLACEMENT_ACTION,
+    ),
+    "activation_release_mismatch": (
+        "The activation marker names a different GitHub release id.",
+        IMMUTABLE_REPLACEMENT_ACTION,
+    ),
+    "activation_source_mismatch": (
+        "The activation marker names a different source commit.",
+        IMMUTABLE_REPLACEMENT_ACTION,
+    ),
+    "source_run_invalid": (
+        "The activation marker has no valid source release run id.",
+        IMMUTABLE_REPLACEMENT_ACTION,
+    ),
+    "convergence_run_invalid": (
+        "The activation marker has no valid convergence run id.",
+        IMMUTABLE_REPLACEMENT_ACTION,
+    ),
+    "delivery_prefix_invalid": (
+        "The activation marker has no customer-delivery prefix.",
+        IMMUTABLE_REPLACEMENT_ACTION,
+    ),
+    "server_digest_invalid": (
+        "The activation marker has no valid server image digest.",
+        IMMUTABLE_REPLACEMENT_ACTION,
+    ),
+    "control_plane_digest_invalid": (
+        "The activation marker has no valid control-plane image digest.",
+        IMMUTABLE_REPLACEMENT_ACTION,
+    ),
+    "helm_digest_invalid": (
+        "The activation marker has no valid Helm chart digest.",
+        IMMUTABLE_REPLACEMENT_ACTION,
+    ),
+}
+
+
+def diagnostic_value(value: Any) -> Any:
+    if value is None or isinstance(value, (bool, int, float)):
+        return value
+    if isinstance(value, str):
+        return value if len(value) <= 160 else value[:157] + "..."
+    return f"<{type(value).__name__}>"
+
+
+def read_json(path: Path) -> Any:
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
+        raise ValueError(f"cannot read JSON from {path}: {exc}") from exc
+
+
+def violation(
+    code: str, field: str, expected: str, actual: Any, rules: dict[str, tuple[str, str]]
+) -> Violation:
+    _, action = rules[code]
+    return Violation(code, field, expected, actual, action)
+
+
+def release_violations(payload: Any) -> list[Violation]:
+    if not isinstance(payload, dict):
+        return [
+            violation(
+                "release_payload_invalid",
+                "$",
+                "object",
+                payload,
+                RELEASE_RULES,
+            )
+        ]
+
+    failures: list[Violation] = []
+    release_id = payload.get("id")
+    if not isinstance(release_id, int) or isinstance(release_id, bool) or release_id <= 0:
+        failures.append(
+            violation("release_id_invalid", "id", "positive integer", release_id, RELEASE_RULES)
+        )
+    tag = payload.get("tag_name")
+    if not isinstance(tag, str) or STABLE_TAG.fullmatch(tag) is None:
+        failures.append(
+            violation("stable_tag_invalid", "tag_name", "vX.Y.Z", tag, RELEASE_RULES)
+        )
+    source = payload.get("target_commitish")
+    if not isinstance(source, str) or SOURCE_SHA.fullmatch(source) is None:
+        failures.append(
+            violation(
+                "source_identity_invalid",
+                "target_commitish",
+                "40 lowercase hexadecimal characters",
+                source,
+                RELEASE_RULES,
+            )
+        )
+    if payload.get("draft") is not False:
+        failures.append(
+            violation("release_is_draft", "draft", "false", payload.get("draft"), RELEASE_RULES)
+        )
+    if payload.get("prerelease") is not False:
+        failures.append(
+            violation(
+                "release_is_prerelease",
+                "prerelease",
+                "false",
+                payload.get("prerelease"),
+                RELEASE_RULES,
+            )
+        )
+    if payload.get("immutable") is not True:
+        failures.append(
+            violation(
+                "release_mutable", "immutable", "true", payload.get("immutable"), RELEASE_RULES
+            )
+        )
+    published_at = payload.get("published_at")
+    if not isinstance(published_at, str) or not published_at:
+        failures.append(
+            violation(
+                "publication_time_invalid",
+                "published_at",
+                "non-empty timestamp",
+                published_at,
+                RELEASE_RULES,
+            )
+        )
+    return failures
+
+
+def release_identity(payload: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "id": payload["id"],
+        "tag": payload["tag_name"],
+        "source_sha": payload["target_commitish"],
+        "draft": payload["draft"],
+        "prerelease": payload["prerelease"],
+        "immutable": payload["immutable"],
+        "published_at": payload["published_at"],
+    }
+
+
+def activation_violations(
+    payload: Any, expected_release: dict[str, Any]
+) -> list[Violation]:
+    if not isinstance(payload, dict):
+        return [
+            violation(
+                "activation_payload_invalid",
+                "$",
+                "object",
+                payload,
+                ACTIVATION_RULES,
+            )
+        ]
+
+    failures: list[Violation] = []
+    checks = (
+        ("activation_schema_invalid", "schema_version", 1, "integer 1"),
+        (
+            "activation_tag_mismatch",
+            "tag",
+            expected_release["tag_name"],
+            expected_release["tag_name"],
+        ),
+        (
+            "activation_release_mismatch",
+            "release_id",
+            str(expected_release["id"]),
+            str(expected_release["id"]),
+        ),
+        (
+            "activation_source_mismatch",
+            "target_commitish",
+            expected_release["target_commitish"],
+            expected_release["target_commitish"],
+        ),
+    )
+    for code, field, expected, expected_description in checks:
+        actual = payload.get(field)
+        if actual != expected or (field == "schema_version" and isinstance(actual, bool)):
+            failures.append(
+                violation(code, field, str(expected_description), actual, ACTIVATION_RULES)
+            )
+
+    for code, field in (
+        ("source_run_invalid", "source_release_run_id"),
+        ("convergence_run_invalid", "convergence_run_id"),
+    ):
+        actual = payload.get(field)
+        if not isinstance(actual, str) or RUN_ID.fullmatch(actual) is None:
+            failures.append(
+                violation(code, field, "decimal run id string", actual, ACTIVATION_RULES)
+            )
+
+    prefix = payload.get("r2_prefix")
+    if not isinstance(prefix, str) or not prefix:
+        failures.append(
+            violation(
+                "delivery_prefix_invalid",
+                "r2_prefix",
+                "non-empty string",
+                prefix,
+                ACTIVATION_RULES,
+            )
+        )
+
+    for code, field in (
+        ("server_digest_invalid", "server_image_digest"),
+        ("control_plane_digest_invalid", "control_plane_image_digest"),
+        ("helm_digest_invalid", "helm_chart_digest"),
+    ):
+        actual = payload.get(field)
+        if not isinstance(actual, str) or SHA256.fullmatch(actual) is None:
+            failures.append(
+                violation(code, field, "sha256:<64 lowercase hex>", actual, ACTIVATION_RULES)
+            )
+    return failures
+
+
+def write_diagnostic(
+    path: Path,
+    check: str,
+    identity: dict[str, Any],
+    failures: list[Violation],
+) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    result = {
+        "schema_version": 1,
+        "check": check,
+        "status": "failure" if failures else "success",
+        "identity": {key: diagnostic_value(value) for key, value in identity.items()},
+        "violations": [item.as_dict() for item in failures],
+    }
+    path.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+
+def append_outputs(path: Path, outputs: dict[str, str]) -> None:
+    with path.open("a", encoding="utf-8") as handle:
+        for key, value in outputs.items():
+            if "\n" in value or "\r" in value:
+                raise ValueError(f"output {key} contains a line break")
+            handle.write(f"{key}={value}\n")
+
+
+def report_failures(failures: list[Violation], rules: dict[str, tuple[str, str]]) -> None:
+    for item in failures:
+        message, action = rules[item.code]
+        print(
+            f"::error title=Stable release continuity [{item.code}]::{message} {action}",
+            file=sys.stderr,
+        )
+
+
+def validate_release(args: argparse.Namespace) -> int:
+    try:
+        payload = read_json(args.release_json)
+    except ValueError as exc:
+        payload = None
+        failures = [
+            violation("release_payload_invalid", "$", "object", str(exc), RELEASE_RULES)
+        ]
+    else:
+        failures = release_violations(payload)
+
+    identity = release_identity(payload) if isinstance(payload, dict) and not failures else {
+        key: payload.get(source) if isinstance(payload, dict) else None
+        for key, source in (
+            ("id", "id"),
+            ("tag", "tag_name"),
+            ("source_sha", "target_commitish"),
+            ("draft", "draft"),
+            ("prerelease", "prerelease"),
+            ("immutable", "immutable"),
+            ("published_at", "published_at"),
+        )
+    }
+    write_diagnostic(args.diagnostic, "stable_release_identity", identity, failures)
+    if failures:
+        report_failures(failures, RELEASE_RULES)
+        return 1
+
+    append_outputs(
+        args.github_output,
+        {
+            "tag": payload["tag_name"],
+            "release_id": str(payload["id"]),
+            "source_sha": payload["target_commitish"],
+        },
+    )
+    return 0
+
+
+def validate_activation(args: argparse.Namespace) -> int:
+    try:
+        release = read_json(args.release_json)
+        release_failures = release_violations(release)
+        if release_failures or not isinstance(release, dict):
+            raise ValueError("release identity did not pass validation")
+        activation = read_json(args.activation_json)
+    except ValueError as exc:
+        activation = None
+        failures = [
+            violation("activation_payload_invalid", "$", "object", str(exc), ACTIVATION_RULES)
+        ]
+        identity: dict[str, Any] = {}
+    else:
+        failures = activation_violations(activation, release)
+        identity = (
+            {
+                key: activation.get(key)
+                for key in (
+                    "schema_version",
+                    "tag",
+                    "release_id",
+                    "target_commitish",
+                    "source_release_run_id",
+                    "convergence_run_id",
+                    "r2_prefix",
+                    "server_image_digest",
+                    "control_plane_image_digest",
+                    "helm_chart_digest",
+                )
+            }
+            if isinstance(activation, dict)
+            else {}
+        )
+    write_diagnostic(args.diagnostic, "release_activation_binding", identity, failures)
+    if failures:
+        report_failures(failures, ACTIVATION_RULES)
+        return 1
+
+    append_outputs(
+        args.github_output,
+        {
+            "activation_sha256": hashlib.sha256(args.activation_json.read_bytes()).hexdigest(),
+            "server_image_digest": activation["server_image_digest"],
+            "control_plane_image_digest": activation["control_plane_image_digest"],
+            "helm_chart_digest": activation["helm_chart_digest"],
+        },
+    )
+    return 0
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    commands = parser.add_subparsers(dest="command", required=True)
+
+    release = commands.add_parser("release")
+    release.add_argument("--release-json", type=Path, required=True)
+    release.add_argument("--diagnostic", type=Path, required=True)
+    release.add_argument("--github-output", type=Path, required=True)
+
+    activation = commands.add_parser("activation")
+    activation.add_argument("--release-json", type=Path, required=True)
+    activation.add_argument("--activation-json", type=Path, required=True)
+    activation.add_argument("--diagnostic", type=Path, required=True)
+    activation.add_argument("--github-output", type=Path, required=True)
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    if args.command == "release":
+        return validate_release(args)
+    return validate_activation(args)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,196 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[2]
+SCRIPT = ROOT / "scripts" / "release_control" / "release_continuity.py"
+SOURCE_SHA = "a" * 40
+DIGEST = "sha256:" + "b" * 64
+
+
+def valid_release() -> dict[str, object]:
+    return {
+        "id": 12345,
+        "tag_name": "v6.4.2",
+        "target_commitish": SOURCE_SHA,
+        "draft": False,
+        "prerelease": False,
+        "immutable": True,
+        "published_at": "2026-08-31T17:00:00Z",
+    }
+
+
+def valid_activation() -> dict[str, object]:
+    return {
+        "schema_version": 1,
+        "tag": "v6.4.2",
+        "release_id": "12345",
+        "target_commitish": SOURCE_SHA,
+        "source_release_run_id": "1001",
+        "convergence_run_id": "1002",
+        "r2_prefix": "releases/v6.4.2",
+        "server_image_digest": DIGEST,
+        "control_plane_image_digest": DIGEST,
+        "helm_chart_digest": DIGEST,
+    }
+
+
+class ReleaseContinuityTest(unittest.TestCase):
+    def run_command(
+        self,
+        command: str,
+        release: object,
+        activation: object | None = None,
+    ) -> tuple[subprocess.CompletedProcess[str], dict[str, object], str]:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            release_path = root / "release.json"
+            release_path.write_text(json.dumps(release), encoding="utf-8")
+            diagnostic = root / "diagnostic.json"
+            output = root / "github-output"
+            output.touch()
+            args = [
+                sys.executable,
+                str(SCRIPT),
+                command,
+                "--release-json",
+                str(release_path),
+                "--diagnostic",
+                str(diagnostic),
+                "--github-output",
+                str(output),
+            ]
+            if command == "activation":
+                activation_path = root / "activation.json"
+                activation_path.write_text(json.dumps(activation), encoding="utf-8")
+                args.extend(["--activation-json", str(activation_path)])
+            result = subprocess.run(
+                args,
+                cwd=ROOT,
+                text=True,
+                capture_output=True,
+                check=False,
+            )
+            return (
+                result,
+                json.loads(diagnostic.read_text(encoding="utf-8")),
+                output.read_text(encoding="utf-8"),
+            )
+
+    def test_accepts_exact_immutable_stable_release(self) -> None:
+        result, diagnostic, output = self.run_command("release", valid_release())
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(diagnostic["status"], "success")
+        self.assertEqual(diagnostic["violations"], [])
+        self.assertIn("tag=v6.4.2\n", output)
+        self.assertIn("release_id=12345\n", output)
+        self.assertIn(f"source_sha={SOURCE_SHA}\n", output)
+
+    def test_mutable_release_fails_with_one_actionable_reason(self) -> None:
+        release = valid_release()
+        release["immutable"] = False
+        result, diagnostic, output = self.run_command("release", release)
+        self.assertEqual(result.returncode, 1)
+        self.assertEqual(output, "")
+        self.assertEqual(diagnostic["status"], "failure")
+        self.assertEqual(
+            [item["code"] for item in diagnostic["violations"]],
+            ["release_mutable"],
+        )
+        self.assertIn("immutable=false", result.stderr)
+        self.assertIn("never repair the packet in place", result.stderr)
+
+    def test_reports_every_release_identity_violation(self) -> None:
+        result, diagnostic, output = self.run_command(
+            "release",
+            {
+                "id": True,
+                "tag_name": "v6.4.2-rc.1\nforged",
+                "target_commitish": "main",
+                "draft": True,
+                "prerelease": True,
+                "immutable": None,
+                "published_at": "",
+            },
+        )
+        self.assertEqual(result.returncode, 1)
+        self.assertEqual(output, "")
+        self.assertEqual(
+            {item["code"] for item in diagnostic["violations"]},
+            {
+                "release_id_invalid",
+                "stable_tag_invalid",
+                "source_identity_invalid",
+                "release_is_draft",
+                "release_is_prerelease",
+                "release_mutable",
+                "publication_time_invalid",
+            },
+        )
+        self.assertNotIn("forged", result.stderr)
+
+    def test_accepts_exact_activation_binding_and_emits_digests(self) -> None:
+        result, diagnostic, output = self.run_command(
+            "activation", valid_release(), valid_activation()
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(diagnostic["status"], "success")
+        self.assertRegex(output, r"activation_sha256=[0-9a-f]{64}\n")
+        self.assertIn(f"server_image_digest={DIGEST}\n", output)
+        self.assertIn(f"control_plane_image_digest={DIGEST}\n", output)
+        self.assertIn(f"helm_chart_digest={DIGEST}\n", output)
+
+    def test_activation_mismatches_are_classified_without_outputs(self) -> None:
+        activation = valid_activation()
+        activation.update(
+            {
+                "schema_version": True,
+                "tag": "v6.4.1",
+                "release_id": 12345,
+                "target_commitish": "c" * 40,
+                "source_release_run_id": "",
+                "convergence_run_id": 1002,
+                "r2_prefix": "",
+                "server_image_digest": "latest",
+                "control_plane_image_digest": None,
+                "helm_chart_digest": "sha256:ABC",
+            }
+        )
+        result, diagnostic, output = self.run_command(
+            "activation", valid_release(), activation
+        )
+        self.assertEqual(result.returncode, 1)
+        self.assertEqual(output, "")
+        self.assertEqual(
+            {item["code"] for item in diagnostic["violations"]},
+            {
+                "activation_schema_invalid",
+                "activation_tag_mismatch",
+                "activation_release_mismatch",
+                "activation_source_mismatch",
+                "source_run_invalid",
+                "convergence_run_invalid",
+                "delivery_prefix_invalid",
+                "server_digest_invalid",
+                "control_plane_digest_invalid",
+                "helm_digest_invalid",
+            },
+        )
+        self.assertTrue(
+            all(
+                "publish a corrected replacement" in item["action"]
+                for item in diagnostic["violations"]
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()