#!/usr/bin/env python3 from __future__ import annotations import argparse import json import re import sys import tempfile from pathlib import Path from typing import Any SCHEMA_VERSION = 1 KINDS = {"comment", "autofix", "report"} SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,79}$") SHA_RE = re.compile(r"^[0-9a-f]{40}$") def fail(message: str) -> None: raise SystemExit(message) def require_slug(value: str, label: str) -> str: if not SLUG_RE.fullmatch(value): fail(f"Invalid {label}: {value!r}") return value def require_kind(kind: str) -> str: if kind not in KINDS: fail(f"Invalid handoff kind: {kind!r}") return kind def artifact_name(kind: str, handoff_id: str) -> str: return f"handoff-{require_kind(kind)}-{require_slug(handoff_id, 'handoff id')}" def artifact_pattern(kind: str) -> str: return f"handoff-{require_kind(kind)}-*" def handoff_dir(root: Path, kind: str, handoff_id: str) -> Path: return root / "handoff" / require_kind(kind) / require_slug(handoff_id, "handoff id") def metadata_path(root: Path, kind: str, handoff_id: str) -> Path: return handoff_dir(root, kind, handoff_id) / "metadata.json" def payload_path(root: Path, kind: str, handoff_id: str) -> Path: kind = require_kind(kind) filename = "body.md" if kind == "comment" else "patch.diff" if kind == "autofix" else "request.json" return handoff_dir(root, kind, handoff_id) / filename def load_metadata(path: Path) -> dict[str, Any]: try: data = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as error: fail(f"Invalid JSON in {path}: {error}") if not isinstance(data, dict): fail(f"Metadata must be an object: {path}") return data def require_int(value: Any, label: str) -> int: if isinstance(value, bool): fail(f"{label} must be an integer") if isinstance(value, int): number = value elif isinstance(value, str) and value.isdigit(): number = int(value) else: fail(f"{label} must be an integer") if number <= 0: fail(f"{label} must be positive") return number def require_sha(value: Any, label: str) -> str: if not isinstance(value, str) or not SHA_RE.fullmatch(value): fail(f"{label} must be a 40-character lowercase hex SHA") return value def require_text(value: Any, label: str) -> str: if not isinstance(value, str) or not value.strip(): fail(f"{label} must be a non-empty string") return value def require_relative_path(value: Any, label: str) -> str: path = require_text(value, label) candidate = Path(path) if candidate.is_absolute() or ".." in candidate.parts: fail(f"{label} must be a repository-relative path without '..': {path!r}") return path def validate_common(entry_dir: Path, expected_kind: str) -> dict[str, Any]: data = load_metadata(entry_dir / "metadata.json") if data.get("schema_version") != SCHEMA_VERSION: fail(f"Unsupported schema_version in {entry_dir}") kind = data.get("kind") if kind != expected_kind: fail(f"Expected kind {expected_kind!r}, got {kind!r} in {entry_dir}") handoff_id = require_slug(require_text(data.get("id"), "id"), "id") allowed_directory_names = {handoff_id, artifact_name(expected_kind, handoff_id)} if entry_dir.name not in allowed_directory_names: fail(f"Metadata id {handoff_id!r} does not match directory {entry_dir.name!r}") normalized: dict[str, Any] = { "schema_version": SCHEMA_VERSION, "kind": kind, "id": handoff_id, "pr_number": require_int(data.get("pr_number"), "pr_number"), "head_sha": require_sha(data.get("head_sha"), "head_sha"), "base_sha": require_sha(data.get("base_sha"), "base_sha"), "run_id": require_int(data.get("run_id"), "run_id"), "path": str(entry_dir), } return normalized def validate_comment(entry_dir: Path) -> dict[str, Any]: normalized = validate_common(entry_dir, "comment") body_path = entry_dir / "body.md" if not body_path.is_file(): fail(f"Missing comment body: {body_path}") marker = require_text(load_metadata(entry_dir / "metadata.json").get("marker"), "marker") body = body_path.read_text(encoding="utf-8") if marker not in body: fail(f"Comment marker is not present in body: {entry_dir}") normalized.update({"marker": marker, "body_path": str(body_path)}) return normalized def validate_autofix(entry_dir: Path) -> dict[str, Any]: metadata = load_metadata(entry_dir / "metadata.json") normalized = validate_common(entry_dir, "autofix") patch_path = entry_dir / "patch.diff" if not patch_path.is_file(): fail(f"Missing autofix patch: {patch_path}") allowed_paths = metadata.get("allowed_paths") if not isinstance(allowed_paths, list) or not allowed_paths: fail("allowed_paths must be a non-empty list") normalized.update( { "allowed_paths": [require_relative_path(item, "allowed_paths item") for item in allowed_paths], "commit_message": require_text(metadata.get("commit_message"), "commit_message"), "patch_path": str(patch_path), } ) return normalized def require_artifact_pattern(value: Any, label: str) -> str: pattern = require_text(value, label) if "/" in pattern or ".." in pattern: fail(f"{label} must be an artifact name pattern, not a path: {pattern!r}") return pattern def validate_report(entry_dir: Path) -> dict[str, Any]: metadata = load_metadata(entry_dir / "metadata.json") normalized = validate_common(entry_dir, "report") normalized.update( { "report_type": require_slug(require_text(metadata.get("report_type"), "report_type"), "report_type"), "artifact_pattern": require_artifact_pattern(metadata.get("artifact_pattern"), "artifact_pattern"), "output_comment_id": require_slug(require_text(metadata.get("output_comment_id"), "output_comment_id"), "output_comment_id"), "marker": require_text(metadata.get("marker"), "marker"), } ) return normalized def validate_entry(kind: str, entry_dir: Path) -> dict[str, Any]: kind = require_kind(kind) if kind == "comment": return validate_comment(entry_dir) if kind == "autofix": return validate_autofix(entry_dir) return validate_report(entry_dir) def candidate_entry_dirs(root: Path, kind: str) -> list[Path]: kind = require_kind(kind) seen: set[Path] = set() entries: list[Path] = [] for metadata in root.rglob("metadata.json"): entry = metadata.parent data = load_metadata(metadata) if data.get("kind") != kind: continue parts = entry.parts accepted = ( len(parts) >= 3 and parts[-3] == "handoff" and parts[-2] == kind and SLUG_RE.fullmatch(parts[-1]) is not None ) or ( len(parts) >= 2 and parts[-2] == kind and SLUG_RE.fullmatch(parts[-1]) is not None ) or SLUG_RE.fullmatch(parts[-1]) is not None resolved = entry.resolve() if accepted and resolved not in seen: seen.add(resolved) entries.append(entry) return sorted(entries, key=lambda path: str(path)) def emit_json(data: Any) -> None: print(json.dumps(data, ensure_ascii=False, sort_keys=True)) def self_check() -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) comment = handoff_dir(root, "comment", "visual-pr-app") comment.mkdir(parents=True) marker = "" (comment / "body.md").write_text(f"{marker}\nVisual report\n", encoding="utf-8") (comment / "metadata.json").write_text( json.dumps( { "schema_version": SCHEMA_VERSION, "kind": "comment", "id": "visual-pr-app", "pr_number": 12, "head_sha": "a" * 40, "base_sha": "b" * 40, "run_id": 34, "marker": marker, } ), encoding="utf-8", ) autofix = handoff_dir(root, "autofix", "nix-pnpm-deps") autofix.mkdir(parents=True) (autofix / "patch.diff").write_text("diff --git a/nix/pnpm-deps.nix b/nix/pnpm-deps.nix\n", encoding="utf-8") (autofix / "metadata.json").write_text( json.dumps( { "schema_version": SCHEMA_VERSION, "kind": "autofix", "id": "nix-pnpm-deps", "pr_number": 12, "head_sha": "a" * 40, "base_sha": "b" * 40, "run_id": 34, "allowed_paths": ["nix/pnpm-deps.nix"], "commit_message": "chore(nix): refresh pnpm deps hash", } ), encoding="utf-8", ) report = handoff_dir(root, "report", "visual-pr") report.mkdir(parents=True) (report / "metadata.json").write_text( json.dumps( { "schema_version": SCHEMA_VERSION, "kind": "report", "id": "visual-pr", "pr_number": 12, "head_sha": "a" * 40, "base_sha": "b" * 40, "run_id": 34, "report_type": "visual-pr", "artifact_pattern": "visual-pr-capture-12-34-*", "output_comment_id": "visual-pr-report", "marker": "", } ), encoding="utf-8", ) assert artifact_name("comment", "visual-pr-app") == "handoff-comment-visual-pr-app" assert artifact_pattern("autofix") == "handoff-autofix-*" assert artifact_name("report", "visual-pr") == "handoff-report-visual-pr" assert validate_entry("comment", comment)["marker"] == marker assert validate_entry("autofix", autofix)["allowed_paths"] == ["nix/pnpm-deps.nix"] assert validate_entry("report", report)["artifact_pattern"] == "visual-pr-capture-12-34-*" assert len(candidate_entry_dirs(root, "comment")) == 1 assert len(candidate_entry_dirs(root, "autofix")) == 1 assert len(candidate_entry_dirs(root, "report")) == 1 print("handoff self-check passed") def main() -> None: parser = argparse.ArgumentParser(description="Manage GitHub Actions handoff artifact names, paths, and contracts.") subparsers = parser.add_subparsers(dest="command", required=True) for command in ["artifact-name", "dir", "metadata-path", "payload-path"]: item = subparsers.add_parser(command) item.add_argument("kind", choices=sorted(KINDS)) item.add_argument("id") item.add_argument("--root", default=".") pattern = subparsers.add_parser("artifact-pattern") pattern.add_argument("kind", choices=sorted(KINDS)) validate = subparsers.add_parser("validate") validate.add_argument("kind", choices=sorted(KINDS)) validate.add_argument("entry_dir") list_parser = subparsers.add_parser("list") list_parser.add_argument("kind", choices=sorted(KINDS)) list_parser.add_argument("root") subparsers.add_parser("self-check") args = parser.parse_args() if args.command == "artifact-name": print(artifact_name(args.kind, args.id)) elif args.command == "artifact-pattern": print(artifact_pattern(args.kind)) elif args.command == "dir": print(handoff_dir(Path(args.root), args.kind, args.id)) elif args.command == "metadata-path": print(metadata_path(Path(args.root), args.kind, args.id)) elif args.command == "payload-path": print(payload_path(Path(args.root), args.kind, args.id)) elif args.command == "validate": emit_json(validate_entry(args.kind, Path(args.entry_dir))) elif args.command == "list": for entry in candidate_entry_dirs(Path(args.root), args.kind): emit_json(validate_entry(args.kind, entry)) elif args.command == "self-check": self_check() if __name__ == "__main__": try: main() except BrokenPipeError: sys.exit(1)