feat(release): publish qualified channel pointers

bot_bottle/release_qualification.py

@@ -0,0 +1,179 @@
+"""Validated release and channel pointers for qualified commit bundles."""
+
+from __future__ import annotations
+
+import json
+import re
+import urllib.error
+import urllib.request
+from typing import Any
+
+from .release_manifest import ReleaseManifestError
+from .release_publish import (
+    HTTP_TIMEOUT_SECONDS,
+    bundle_url,
+    config,
+    remote_bytes,
+    request,
+    upload,
+)
+
+_COMMIT_RE = re.compile(r"^[0-9a-f]{40}$")
+_STABLE_TAG_RE = re.compile(r"^v(\d+)\.(\d+)\.(\d+)$")
+_STAGING_TAG_RE = re.compile(r"^v(\d+)\.(\d+)\.(\d+)-rc\.(\d+)$")
+
+
+def release_channel(tag: str) -> str:
+    """Return the only channel eligible for *tag*."""
+    if _STABLE_TAG_RE.fullmatch(tag):
+        return "production"
+    if _STAGING_TAG_RE.fullmatch(tag):
+        return "staging"
+    raise ReleaseManifestError(
+        "release tag must be vX.Y.Z or vX.Y.Z-rc.N")
+
+
+def release_version(tag: str) -> tuple[int, int, int, int]:
+    """Return a monotonically comparable version tuple."""
+    stable = _STABLE_TAG_RE.fullmatch(tag)
+    if stable:
+        return *(int(value) for value in stable.groups()), 1 << 30
+    staging = _STAGING_TAG_RE.fullmatch(tag)
+    if staging:
+        return *(int(value) for value in staging.groups()[:3]), int(
+            staging.group(4))
+    raise ReleaseManifestError(
+        "release tag must be vX.Y.Z or vX.Y.Z-rc.N")
+
+
+def pointer_url(package: str, version: str, filename: str) -> str:
+    base, owner, _ = config()
+    return f"{base}/api/packages/{owner}/generic/{package}/{version}/{filename}"
+
+
+def build_release_pointer(
+    *,
+    tag: str,
+    source_commit: str,
+    workflow_run: str,
+    qualified_at: str,
+) -> dict[str, Any]:
+    """Build a qualified pointer to one immutable commit bundle."""
+    channel = release_channel(tag)
+    if _COMMIT_RE.fullmatch(source_commit) is None:
+        raise ReleaseManifestError(
+            "release source_commit must be 40 lowercase hex")
+    if not workflow_run.strip() or not qualified_at.strip():
+        raise ReleaseManifestError(
+            "release qualification provenance is incomplete")
+    return {
+        "schema": 1,
+        "qualified": True,
+        "tag": tag,
+        "channel": channel,
+        "source_commit": source_commit,
+        "bundle_index_url": bundle_url(source_commit, "bundle-index.json"),
+        "qualification": {
+            "workflow_run": workflow_run.strip(),
+            "qualified_at": qualified_at.strip(),
+        },
+    }
+
+
+def parse_release_pointer(data: Any) -> dict[str, Any]:
+    """Validate a release or channel pointer."""
+    if not isinstance(data, dict) or data.get("schema") != 1:
+        raise ReleaseManifestError("release pointer must use schema 1")
+    if data.get("qualified") is not True:
+        raise ReleaseManifestError("release pointer must be qualified")
+    tag = data.get("tag")
+    channel = data.get("channel")
+    commit = data.get("source_commit")
+    if not isinstance(tag, str) or release_channel(tag) != channel:
+        raise ReleaseManifestError("release pointer tag and channel disagree")
+    if not isinstance(commit, str) or _COMMIT_RE.fullmatch(commit) is None:
+        raise ReleaseManifestError("release pointer source commit is invalid")
+    if data.get("bundle_index_url") != bundle_url(commit, "bundle-index.json"):
+        raise ReleaseManifestError("release pointer bundle URL is invalid")
+    qualification = data.get("qualification")
+    if not isinstance(qualification, dict) or not all(
+        isinstance(qualification.get(key), str) and qualification[key].strip()
+        for key in ("workflow_run", "qualified_at")
+    ):
+        raise ReleaseManifestError("release qualification is incomplete")
+    return data
+
+
+def canonical_pointer(data: Any) -> bytes:
+    parse_release_pointer(data)
+    return (json.dumps(data, indent=2, sort_keys=True) + "\n").encode()
+
+
+def delete(url: str) -> None:
+    try:
+        with urllib.request.urlopen(
+            request(url, method="DELETE"), timeout=HTTP_TIMEOUT_SECONDS,
+        ):
+            pass
+    except urllib.error.HTTPError as exc:
+        if exc.code != 404:
+            raise ReleaseManifestError(
+                f"replacing release channel failed (HTTP {exc.code}): {url}"
+            ) from exc
+    except urllib.error.URLError as exc:
+        raise ReleaseManifestError(
+            f"release registry unreachable: {url} ({exc.reason})") from exc
+
+
+def publish_qualification(
+    pointer: dict[str, Any], *, allow_rollback: bool = False,
+) -> tuple[str, str]:
+    """Publish an immutable tag pointer and advance its channel."""
+    parse_release_pointer(pointer)
+    tag = str(pointer["tag"])
+    channel = str(pointer["channel"])
+    wanted = canonical_pointer(pointer)
+    release_url = pointer_url(
+        "bot-bottle-releases", tag, "release.json")
+    channel_url = pointer_url(
+        "bot-bottle-channels", channel, "channel.json")
+
+    if remote_bytes(str(pointer["bundle_index_url"])) is None:
+        raise ReleaseManifestError(
+            f"commit bundle does not exist for {pointer['source_commit']}")
+    existing_release = remote_bytes(release_url)
+    if existing_release is not None and existing_release != wanted:
+        raise ReleaseManifestError(
+            f"immutable release pointer {tag} already differs")
+    if existing_release is None:
+        upload(release_url, wanted)
+
+    existing_channel = remote_bytes(channel_url)
+    if existing_channel == wanted:
+        return release_url, channel_url
+    if existing_channel is not None:
+        try:
+            current = parse_release_pointer(json.loads(existing_channel))
+        except (json.JSONDecodeError, UnicodeDecodeError) as exc:
+            raise ReleaseManifestError(
+                f"existing {channel} channel pointer is malformed") from exc
+        if (
+            not allow_rollback
+            and release_version(tag) <= release_version(str(current["tag"]))
+        ):
+            raise ReleaseManifestError(
+                f"{channel} channel update is not monotonic; "
+                "use the explicit rollback operation")
+        delete(channel_url)
+    upload(channel_url, wanted)
+    return release_url, channel_url
+
+
+__all__ = [
+    "build_release_pointer",
+    "canonical_pointer",
+    "parse_release_pointer",
+    "publish_qualification",
+    "release_channel",
+    "release_version",
+]

scripts/publish_release_qualification.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+"""Publish a qualified release and advance its matching channel."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from bot_bottle.release_qualification import (
+    build_release_pointer,
+    canonical_pointer,
+    publish_qualification,
+)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--tag", required=True)
+    parser.add_argument("--source-commit", required=True)
+    parser.add_argument("--workflow-run", required=True)
+    parser.add_argument("--qualified-at", required=True)
+    parser.add_argument("--output", type=Path, default=Path("release.json"))
+    parser.add_argument("--allow-rollback", action="store_true")
+    args = parser.parse_args()
+    pointer = build_release_pointer(
+        tag=args.tag,
+        source_commit=args.source_commit,
+        workflow_run=args.workflow_run,
+        qualified_at=args.qualified_at,
+    )
+    args.output.write_bytes(canonical_pointer(pointer))
+    release_url, channel_url = publish_qualification(
+        pointer, allow_rollback=args.allow_rollback)
+    print(json.dumps({
+        "release": release_url,
+        "channel": channel_url,
+    }, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

tests/unit/test_release_qualification.py

@@ -0,0 +1,74 @@
+"""Tests for release qualification and channel promotion."""
+
+from __future__ import annotations
+
+import json
+import unittest
+from unittest import mock
+
+from bot_bottle.release_manifest import ReleaseManifestError
+from bot_bottle.release_qualification import (
+    build_release_pointer,
+    publish_qualification,
+    release_channel,
+)
+
+COMMIT = "a" * 40
+
+
+class ReleaseQualificationTest(unittest.TestCase):
+    def test_tag_selects_channel(self) -> None:
+        self.assertEqual(release_channel("v1.2.3"), "production")
+        self.assertEqual(release_channel("v1.2.3-rc.4"), "staging")
+        with self.assertRaises(ReleaseManifestError):
+            release_channel("latest")
+
+    def test_builds_qualified_pointer(self) -> None:
+        pointer = build_release_pointer(
+            tag="v1.2.3",
+            source_commit=COMMIT,
+            workflow_run="https://example.test/run/1",
+            qualified_at="2026-07-27T00:00:00Z",
+        )
+        self.assertEqual(pointer["channel"], "production")
+        self.assertTrue(pointer["qualified"])
+
+    @mock.patch("bot_bottle.release_qualification.upload")
+    @mock.patch("bot_bottle.release_qualification.remote_bytes")
+    def test_publishes_bundle_release_and_channel(
+        self, remote: mock.Mock, upload: mock.Mock,
+    ) -> None:
+        remote.side_effect = [b"bundle", None, None]
+        pointer = build_release_pointer(
+            tag="v1.2.3",
+            source_commit=COMMIT,
+            workflow_run="run",
+            qualified_at="now",
+        )
+        publish_qualification(pointer)
+        self.assertEqual(upload.call_count, 2)
+
+    @mock.patch("bot_bottle.release_qualification.upload")
+    @mock.patch("bot_bottle.release_qualification.remote_bytes")
+    def test_rejects_non_monotonic_channel(
+        self, remote: mock.Mock, _upload: mock.Mock,
+    ) -> None:
+        old = build_release_pointer(
+            tag="v2.0.0",
+            source_commit="b" * 40,
+            workflow_run="old-run",
+            qualified_at="old-time",
+        )
+        remote.side_effect = [b"bundle", None, json.dumps(old).encode()]
+        pointer = build_release_pointer(
+            tag="v1.2.3",
+            source_commit=COMMIT,
+            workflow_run="run",
+            qualified_at="now",
+        )
+        with self.assertRaisesRegex(ReleaseManifestError, "not monotonic"):
+            publish_qualification(pointer)
+
+
+if __name__ == "__main__":
+    unittest.main()