(MOT-4299) feat: stage worker releases before promotion

.github/scripts/parse_release_tag.py

@@ -1,16 +1,17 @@
 #!/usr/bin/env python3
 """Parse a release tag and emit setup outputs for release.yml.
 
-Writes 12 keys to $GITHUB_OUTPUT:
+Writes 13 keys to $GITHUB_OUTPUT:
     tag, worker, version, deploy, language, bin, manifest,
-    registry_tag, is_prerelease, dry_run, targets, experimental
+    registry_tag, is_prerelease, dry_run, targets, experimental, tag_sha
 """
 from __future__ import annotations
 
 import argparse
 import os
 import pathlib
 import re
+import subprocess
 import sys
 
 sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
@@ -19,7 +20,7 @@
 
 TAG_RE = re.compile(r"^([a-z0-9][a-z0-9_-]*)/v(.+)$")
 DRY_RUN_RE = re.compile(r"-dry-run\.\d+$")
-PRERELEASE_RE = re.compile(r"-[a-z]+\.\d+$")
+STABLE_VERSION_RE = re.compile(r"^[0-9]+\.[0-9]+\.[0-9]+$")
 
 
 def main(argv: list[str] | None = None) -> int:
@@ -36,10 +37,10 @@ def main(argv: list[str] | None = None) -> int:
 
     if DRY_RUN_RE.search(version):
         dry_run, is_pre = "true", "true"
-    elif PRERELEASE_RE.search(version):
-        dry_run, is_pre = "false", "true"
-    else:
+    elif STABLE_VERSION_RE.fullmatch(version):
         dry_run, is_pre = "false", "false"
+    else:
+        dry_run, is_pre = "false", "true"
 
     worker_dir = pathlib.Path(worker)
     try:
@@ -57,6 +58,9 @@ def main(argv: list[str] | None = None) -> int:
 
     annotation = _lib.read_tag_annotation(raw)
     registry_tag = annotation.get("registry-tag", "latest") or "latest"
+    tag_sha = subprocess.check_output(
+        ["git", "rev-list", "-n", "1", raw], text=True
+    ).strip()
 
     # Anything but a literal `true` is false: a lightweight tag, a missing
     # line, or a typo publishes as stable. Marking a worker experimental is
@@ -87,6 +91,7 @@ def main(argv: list[str] | None = None) -> int:
         ("dry_run", dry_run),
         ("targets", targets),
         ("experimental", experimental),
+        ("tag_sha", tag_sha),
     ]
 
     gh_out = os.environ.get("GITHUB_OUTPUT")

.github/scripts/registry_release.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""Promote a staged worker release through the Registry HTTP API."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+
+class RegistryError(RuntimeError):
+    pass
+
+
+def request_json(
+    method: str,
+    url: str,
+    payload: dict[str, Any],
+    *,
+    api_key: str | None = None,
+) -> tuple[int, dict[str, Any]]:
+    body = json.dumps(payload).encode()
+    headers = {"Content-Type": "application/json"}
+    if api_key:
+        headers["X-API-Key"] = api_key
+    request = urllib.request.Request(url, data=body, headers=headers, method=method)
+    try:
+        with urllib.request.urlopen(request, timeout=60) as response:
+            return response.status, json.loads(response.read().decode())
+    except urllib.error.HTTPError as error:
+        try:
+            response = json.loads(error.read().decode())
+        except (json.JSONDecodeError, UnicodeDecodeError):
+            response = {"error": f"HTTP {error.code}"}
+        return error.code, response
+
+
+def resolved_root_version(response: dict[str, Any]) -> str:
+    root = response.get("root")
+    version = root.get("version") if isinstance(root, dict) else None
+    if not isinstance(version, str) or not version:
+        raise RegistryError("Registry resolve response has no root.version")
+    return version
+
+
+def resolve_version(api_url: str, worker: str, selector: str, *, allow_missing: bool = False) -> str | None:
+    status, response = request_json(
+        "POST",
+        f"{api_url.rstrip('/')}/resolve",
+        {"worker": worker, "version": selector},
+    )
+    if status == 200:
+        return resolved_root_version(response)
+    error = response.get("error")
+    code = error.get("code") if isinstance(error, dict) else None
+    if allow_missing and code in {"version_not_found", "worker_not_found"}:
+        return None
+    raise RegistryError(f"resolve {worker}@{selector} failed with HTTP {status}: {json.dumps(response)}")
+
+
+def promotion_payload(version: str, current_latest: str | None) -> dict[str, str]:
+    payload = {"version": version, "expected_tag": "next"}
+    if current_latest is not None:
+        payload["expected_current_version"] = current_latest
+    return payload
+
+
+def promote(api_url: str, api_key: str, worker: str, version: str) -> dict[str, Any]:
+    current_latest = resolve_version(api_url, worker, "latest", allow_missing=True)
+    current_next = resolve_version(api_url, worker, "next", allow_missing=True)
+    # A first promotion must still own `next`. Once Registry latest already
+    # points at the requested immutable version, allow an idempotent retry to
+    # repair GitHub/GHCR/Slack even if a newer candidate has moved `next`.
+    if current_latest != version and current_next != version:
+        raise RegistryError(f"next points to {current_next}, expected {version}")
+    encoded_worker = urllib.parse.quote(worker, safe="")
+    status, response = request_json(
+        "PUT",
+        f"{api_url.rstrip('/')}/w/{encoded_worker}/tags/latest",
+        promotion_payload(version, current_latest),
+        api_key=api_key,
+    )
+    if status != 200:
+        raise RegistryError(f"promotion failed with HTTP {status}: {json.dumps(response)}")
+
+    promoted = resolve_version(api_url, worker, "latest")
+    if promoted != version:
+        raise RegistryError(f"promotion verification resolved {promoted}, expected {version}")
+
+    return {
+        "worker": worker,
+        "version": version,
+        "previous_latest": current_latest,
+        "next": current_next,
+        "latest": promoted,
+        "changed": bool(response.get("changed")),
+        "registry_response": response,
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--api-url", default="https://api.workers.iii.dev")
+    parser.add_argument("--worker", required=True)
+    parser.add_argument("--version", required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+
+    api_key = os.environ.get("WORKERS_REGISTRY_API_KEY", "")
+    if not api_key:
+        raise SystemExit("WORKERS_REGISTRY_API_KEY is required")
+    try:
+        result = promote(args.api_url, api_key, args.worker, args.version)
+    except RegistryError as error:
+        raise SystemExit(str(error)) from error
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n")
+    print(json.dumps(result, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

.github/scripts/release_candidate.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+"""Build and validate evidence for a staged worker release."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+from pathlib import Path
+
+
+SHA_RE = re.compile(r"^[0-9a-f]{40}$")
+VERSION_RE = re.compile(r"^[0-9]+\.[0-9]+\.[0-9]+$")
+
+
+def parse_bool(value: str) -> bool:
+    normalized = value.strip().lower()
+    if normalized == "true":
+        return True
+    if normalized == "false":
+        return False
+    raise argparse.ArgumentTypeError("expected true or false")
+
+
+def build_evidence(args: argparse.Namespace) -> dict:
+    results = {
+        "publish": args.publish_result,
+        "candidate_smoke": args.candidate_smoke_result,
+        "harness_quickstart": args.harness_quickstart_result,
+        "harness_e2e": args.harness_e2e_result,
+    }
+    candidate_ready = results["publish"] == "success" and results["candidate_smoke"] == "success"
+    if args.harness_gate_required:
+        candidate_ready = (
+            candidate_ready
+            and results["harness_quickstart"] == "success"
+            and results["harness_e2e"] == "success"
+        )
+
+    return {
+        "schema_version": 1,
+        "repository": args.repository,
+        "release_run_id": args.release_run_id,
+        "run_attempt": args.run_attempt,
+        "tag_sha": args.tag_sha,
+        "release_tag": args.release_tag,
+        "worker": args.worker,
+        "version": args.version,
+        "deploy": args.deploy,
+        "registry_tag": args.registry_tag,
+        "harness_gate_required": args.harness_gate_required,
+        "promotable": args.promotable,
+        "candidate_ready": candidate_ready,
+        "results": results,
+    }
+
+
+def validate_evidence(args: argparse.Namespace) -> dict:
+    evidence = json.loads(args.evidence.read_text())
+    failures: list[str] = []
+
+    expected = {
+        "schema_version": 1,
+        "repository": args.repository,
+        "release_run_id": args.release_run_id,
+        "release_tag": f"{args.worker}/v{args.version}",
+        "worker": args.worker,
+        "version": args.version,
+        "registry_tag": "next",
+        "candidate_ready": True,
+        "promotable": True,
+    }
+    for key, value in expected.items():
+        if evidence.get(key) != value:
+            failures.append(f"{key}: expected {value!r}, got {evidence.get(key)!r}")
+
+    if not VERSION_RE.fullmatch(args.version):
+        failures.append("version must be stable semver MAJOR.MINOR.PATCH")
+    if not SHA_RE.fullmatch(str(evidence.get("tag_sha", ""))):
+        failures.append("tag_sha must be a full lowercase commit SHA")
+    if not isinstance(evidence.get("run_attempt"), int) or evidence["run_attempt"] < 1:
+        failures.append("run_attempt must be a positive integer")
+
+    results = evidence.get("results")
+    if not isinstance(results, dict):
+        failures.append("results must be an object")
+    else:
+        if results.get("publish") != "success":
+            failures.append("publish gate did not succeed")
+        if results.get("candidate_smoke") != "success":
+            failures.append("candidate smoke gate did not succeed")
+        if evidence.get("harness_gate_required"):
+            if results.get("harness_quickstart") != "success":
+                failures.append("Harness quickstart gate did not succeed")
+            if results.get("harness_e2e") != "success":
+                failures.append("Harness E2E gate did not succeed")
+
+    if failures:
+        raise SystemExit("invalid release candidate evidence:\n- " + "\n- ".join(failures))
+    return evidence
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser()
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    build = subparsers.add_parser("build")
+    build.add_argument("--repository", required=True)
+    build.add_argument("--release-run-id", required=True)
+    build.add_argument("--run-attempt", type=int, required=True)
+    build.add_argument("--tag-sha", required=True)
+    build.add_argument("--release-tag", required=True)
+    build.add_argument("--worker", required=True)
+    build.add_argument("--version", required=True)
+    build.add_argument("--deploy", choices=("binary", "image", "bundle"), required=True)
+    build.add_argument("--registry-tag", required=True)
+    build.add_argument("--harness-gate-required", type=parse_bool, required=True)
+    build.add_argument("--promotable", type=parse_bool, required=True)
+    build.add_argument("--publish-result", required=True)
+    build.add_argument("--candidate-smoke-result", required=True)
+    build.add_argument("--harness-quickstart-result", required=True)
+    build.add_argument("--harness-e2e-result", required=True)
+    build.add_argument("--output", type=Path, required=True)
+
+    validate = subparsers.add_parser("validate")
+    validate.add_argument("--evidence", type=Path, required=True)
+    validate.add_argument("--repository", required=True)
+    validate.add_argument("--release-run-id", required=True)
+    validate.add_argument("--worker", required=True)
+    validate.add_argument("--version", required=True)
+    validate.add_argument("--output", type=Path)
+    return parser
+
+
+def main() -> None:
+    args = build_parser().parse_args()
+    if args.command == "build":
+        evidence = build_evidence(args)
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        args.output.write_text(json.dumps(evidence, indent=2, sort_keys=True) + "\n")
+    else:
+        evidence = validate_evidence(args)
+        rendered = json.dumps(evidence, sort_keys=True)
+        if args.output:
+            args.output.parent.mkdir(parents=True, exist_ok=True)
+            args.output.write_text(rendered + "\n")
+        print(rendered)
+
+
+if __name__ == "__main__":
+    main()

.github/scripts/tests/test_parse_release_tag.py

@@ -69,6 +69,7 @@ def test_stable_binary_tag(self, tmp_path):
         assert out["registry_tag"] == "latest"
         assert out["is_prerelease"] == "false"
         assert out["dry_run"] == "false"
+        assert len(out["tag_sha"]) == 40
 
     def test_prerelease_sets_is_prerelease(self, tmp_path):
         repo = make_repo_with_tagged_worker(tmp_path, "smoke/v1.2.3-rc.1", "1.2.3-rc.1",
@@ -82,6 +83,19 @@ def test_prerelease_sets_is_prerelease(self, tmp_path):
         assert out["dry_run"] == "false"
         assert out["registry_tag"] == "next"
 
+    def test_non_numeric_prerelease_suffix_is_not_promotable_stable(self, tmp_path):
+        repo = make_repo_with_tagged_worker(
+            tmp_path,
+            "smoke/v1.2.3-preview",
+            "1.2.3-preview",
+            registry_tag_line="registry-tag: next",
+        )
+        out_path = tmp_path / "gh_output"
+        out_path.touch()
+        r = run_script(repo, "smoke/v1.2.3-preview", out_path)
+        assert r.returncode == 0
+        assert parse_outputs(out_path)["is_prerelease"] == "true"
+
     def test_dry_run_tag(self, tmp_path):
         repo = make_repo_with_tagged_worker(tmp_path, "smoke/v9.9.9-dry-run.1", "9.9.9-dry-run.1")
         out_path = tmp_path / "gh_output"