"""Run the complete release gate from an anonymous clone at one exact commit.""" from __future__ import annotations import argparse import hashlib import json import os import re import subprocess import sys import tempfile import time from pathlib import Path ROOT = Path(__file__).resolve().parents[1] PUBLIC_REPOSITORY = "https://repo.andraxion.net/administrator/DocForge2.git" EXPECTED_ORIGINS = { "forgejo@repo.andraxion.net:administrator/DocForge2.git", PUBLIC_REPOSITORY, PUBLIC_REPOSITORY.removesuffix(".git"), } COMMIT_PATTERN = re.compile(r"^[0-9a-f]{40}$") COMMAND_TIMEOUT_SECONDS = 30 * 60 LEGACY_MIGRATION_TAG = "v1.0.0" LEGACY_MIGRATION_TAG_REF = f"refs/tags/{LEGACY_MIGRATION_TAG}" EXPECTED_LEGACY_TAG_OBJECT = "2d7d306a37da89f1c860c7f0be161c45386acf61" EXPECTED_LEGACY_TAG_COMMIT = "593c173b453236a6872d0a4e88e7a51a67a21cde" class FreshCloneError(RuntimeError): """The anonymous exact-commit release rehearsal failed.""" def _run( arguments: list[str], *, cwd: Path, environment: dict[str, str] | None = None, ) -> subprocess.CompletedProcess[str]: completed = subprocess.run( arguments, cwd=cwd, env=environment, check=False, capture_output=True, text=True, timeout=COMMAND_TIMEOUT_SECONDS, ) if completed.returncode != 0: diagnostic = (completed.stdout + completed.stderr)[-4000:] raise FreshCloneError(f"Command failed ({' '.join(arguments)}):\n{diagnostic.rstrip()}") return completed def _head_commit() -> str: commit = _run(["git", "rev-parse", "HEAD"], cwd=ROOT).stdout.strip() if COMMIT_PATTERN.fullmatch(commit) is None: raise FreshCloneError("Current release commit is invalid") return commit def _verify_source(commit: str) -> None: if _head_commit() != commit: raise FreshCloneError("Requested fresh-clone commit is not the current candidate") if _run(["git", "status", "--porcelain"], cwd=ROOT).stdout: raise FreshCloneError("Fresh-clone proof requires a clean source candidate") origin = _run(["git", "remote", "get-url", "origin"], cwd=ROOT).stdout.strip() if origin not in EXPECTED_ORIGINS: raise FreshCloneError("Origin is not the public DocForge2 successor repository") remote = _run( ["git", "ls-remote", PUBLIC_REPOSITORY, "refs/heads/main", "refs/heads/dev"], cwd=ROOT, ).stdout remote_commits = {line.split()[0] for line in remote.splitlines() if line.split()} if commit not in remote_commits: raise FreshCloneError("Release candidate is not published on main or dev") def _digest(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def validate_legacy_tag(*, object_type: str, tag_object: str, tag_commit: str) -> None: if object_type != "tag": raise FreshCloneError(f"{LEGACY_MIGRATION_TAG} is not an annotated tag") if tag_object != EXPECTED_LEGACY_TAG_OBJECT: raise FreshCloneError( f"{LEGACY_MIGRATION_TAG} tag object does not match the frozen release" ) if tag_commit != EXPECTED_LEGACY_TAG_COMMIT: raise FreshCloneError(f"{LEGACY_MIGRATION_TAG} commit does not match the frozen release") def obtain_legacy_tag( clone: Path, ) -> tuple[subprocess.CompletedProcess[str], str, str]: fetch = _run( [ "git", "fetch", "--no-tags", "origin", f"{LEGACY_MIGRATION_TAG_REF}:{LEGACY_MIGRATION_TAG_REF}", ], cwd=clone, ) object_type = _run( ["git", "cat-file", "-t", LEGACY_MIGRATION_TAG_REF], cwd=clone, ).stdout.strip() tag_object = _run( ["git", "rev-parse", LEGACY_MIGRATION_TAG_REF], cwd=clone, ).stdout.strip() tag_commit = _run( ["git", "rev-parse", f"{LEGACY_MIGRATION_TAG_REF}^{{commit}}"], cwd=clone, ).stdout.strip() validate_legacy_tag( object_type=object_type, tag_object=tag_object, tag_commit=tag_commit, ) return fetch, tag_object, tag_commit def run_fresh_clone_gate(commit: str) -> dict[str, object]: """Clone the public successor anonymously and run its complete release gate.""" if COMMIT_PATTERN.fullmatch(commit) is None: raise FreshCloneError("Fresh-clone commit must be one full lowercase SHA-1") _verify_source(commit) started = time.perf_counter_ns() with tempfile.TemporaryDirectory(prefix="docforge-m5-fresh-clone-") as directory_name: parent = Path(directory_name) clone = parent / "DocForge2" clone_result = _run( ["git", "clone", "--no-tags", PUBLIC_REPOSITORY, str(clone)], cwd=parent, ) _run(["git", "checkout", "--detach", commit], cwd=clone) checked_out = _run(["git", "rev-parse", "HEAD"], cwd=clone).stdout.strip() if checked_out != commit: raise FreshCloneError("Anonymous clone did not check out the requested commit") tag_fetch, tag_object, tag_commit = obtain_legacy_tag(clone) fsck = _run(["git", "fsck", "--full"], cwd=clone) environment = dict(os.environ) environment["UV_LINK_MODE"] = "copy" sync = _run(["uv", "sync", "--frozen", "--offline"], cwd=clone, environment=environment) npm = _run(["npm", "ci", "--offline"], cwd=clone, environment=environment) gate = _run(["make", "release-gate"], cwd=clone, environment=environment) status = _run(["git", "status", "--porcelain"], cwd=clone).stdout if status: raise FreshCloneError("Release gate left the anonymous clone dirty") version = _run( [str(clone / ".venv/bin/python"), "-m", "docforge.cli", "--version"], cwd=clone, environment=environment, ).stdout.strip() elapsed_ms = round((time.perf_counter_ns() - started) / 1_000_000, 3) return { "schema_version": 2, "repository": PUBLIC_REPOSITORY, "authentication": "anonymous_https", "commit": commit, "legacy_migration_tag": { "name": LEGACY_MIGRATION_TAG, "annotated_tag_object": tag_object, "commit": tag_commit, }, "clean_after_gate": True, "version_surface": version, "elapsed_ms": elapsed_ms, "logs": { "clone_sha256": _digest(clone_result.stdout + clone_result.stderr), "legacy_tag_fetch_sha256": _digest(tag_fetch.stdout + tag_fetch.stderr), "fsck_sha256": _digest(fsck.stdout + fsck.stderr), "sync_sha256": _digest(sync.stdout + sync.stderr), "npm_sha256": _digest(npm.stdout + npm.stderr), "release_gate_sha256": _digest(gate.stdout + gate.stderr), "release_gate_bytes": len((gate.stdout + gate.stderr).encode("utf-8")), }, } def _write_output(path: Path, payload: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) descriptor, temporary_name = tempfile.mkstemp(prefix=".m5-fresh-clone-", dir=path.parent) temporary = Path(temporary_name) try: with os.fdopen(descriptor, "wb") as handle: handle.write(payload) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) directory_fd = os.open(path.parent, os.O_RDONLY) try: os.fsync(directory_fd) finally: os.close(directory_fd) except Exception: temporary.unlink(missing_ok=True) raise def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--commit") parser.add_argument("--output", type=Path) arguments = parser.parse_args() try: commit = arguments.commit or _head_commit() evidence = run_fresh_clone_gate(commit) except (FreshCloneError, OSError, subprocess.SubprocessError) as error: print(str(error), file=sys.stderr) return 2 payload = json.dumps(evidence, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n" if arguments.output is not None: _write_output(arguments.output, payload) print(payload.decode("utf-8"), end="") return 0 if __name__ == "__main__": raise SystemExit(main())