"""Fail-closed Authority Git side-effect boundary. Only this module may issue the narrowly scoped ref CAS primitives used by the durable executor. It never accepts force updates, arbitrary refs, tags, or direct production mutations. """ from __future__ import annotations from dataclasses import dataclass import os from pathlib import Path import re import subprocess import tempfile from typing import Mapping, Sequence OID_RE = re.compile(r"^[0-9a-f]{40}$") TASK_BRANCH_RE = re.compile(r"^refs/heads/agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$") class AuthorityGitError(RuntimeError): pass def _oid(value: str, field: str) -> str: if not isinstance(value, str) or not OID_RE.fullmatch(value): raise AuthorityGitError(f"{field}_INVALID") return value @dataclass(frozen=True, slots=True) class AuthorityGitExecutor: git_dir: Path worktree_root: Path = Path("/var/lib/aios-gitops/worktrees") def __post_init__(self) -> None: if not self.git_dir.is_absolute() or self.git_dir.is_symlink(): raise AuthorityGitError("AUTHORITY_GIT_DIR_INVALID") if not self.worktree_root.is_absolute() or self.worktree_root.is_symlink(): raise AuthorityGitError("WORKTREE_ROOT_INVALID") def _git(self, args: Sequence[str], *, check: bool = True) -> str: result = subprocess.run( ["git", f"--git-dir={self.git_dir}", *args], text=True, capture_output=True, check=False, ) if check and result.returncode != 0: raise AuthorityGitError("AUTHORITY_GIT_COMMAND_FAILED") return result.stdout.strip() def read_ref(self, ref: str) -> str | None: if not ref.startswith("refs/") or ref.startswith(("refs/tags/", "refs/remotes/")): raise AuthorityGitError("REF_READ_SCOPE_DENIED") value = self._git(["rev-parse", "--verify", ref], check=False) return value if value and OID_RE.fullmatch(value) else None def commit_exists(self, oid: str) -> bool: _oid(oid, "commit_oid") result = subprocess.run(["git", f"--git-dir={self.git_dir}", "cat-file", "-e", oid + "^{commit}"], capture_output=True, check=False) return result.returncode == 0 def ancestry_check(self, older_oid: str, newer_oid: str) -> bool: _oid(older_oid, "older_oid") _oid(newer_oid, "newer_oid") result = subprocess.run( ["git", f"--git-dir={self.git_dir}", "merge-base", "--is-ancestor", older_oid, newer_oid], capture_output=True, check=False, ) if result.returncode not in (0, 1): raise AuthorityGitError("ANCESTRY_CHECK_FAILED") return result.returncode == 0 def create_detached_worktree( self, base_oid: str, worktree_name: str, root: Path | None = None, ) -> Path: _oid(base_oid, "base_oid") if not worktree_name or "/" in worktree_name or "\\" in worktree_name or worktree_name in {".", ".."}: raise AuthorityGitError("WORKTREE_NAME_INVALID") parent = root or self.worktree_root if not parent.is_absolute() or parent.is_symlink(): raise AuthorityGitError("WORKTREE_ROOT_INVALID") path = parent / worktree_name if path.exists() or path.is_symlink(): raise AuthorityGitError("WORKTREE_ALREADY_EXISTS") parent.mkdir(parents=True, exist_ok=True) self._git(["worktree", "add", "--detach", str(path), base_oid]) return path def create_task_branch(self, branch: str, base_oid: str) -> str: """Create an isolated agent branch once, using an empty-ref CAS.""" ref = branch if branch.startswith("refs/heads/") else "refs/heads/" + branch if not TASK_BRANCH_RE.fullmatch(ref): raise AuthorityGitError("TASK_BRANCH_SCOPE_DENIED") _oid(base_oid, "base_oid") if not self.commit_exists(base_oid): raise AuthorityGitError("BASE_COMMIT_NOT_FOUND") current = self.read_ref(ref) if current is not None: if current == base_oid: return current raise AuthorityGitError("AUTHORITY_BRANCH_EXPECTATION_MISMATCH") self._git(["update-ref", ref, base_oid, "0" * 40]) readback = self.read_ref(ref) if readback != base_oid: raise AuthorityGitError("AUTHORITY_CAS_READBACK_MISMATCH") return readback def advance_task_branch_CAS(self, branch: str, expected_oid: str, new_oid: str) -> str: if not TASK_BRANCH_RE.fullmatch(branch): raise AuthorityGitError("TASK_BRANCH_SCOPE_DENIED") _oid(expected_oid, "expected_oid") _oid(new_oid, "new_oid") current = self.read_ref(branch) if current != expected_oid: if current == new_oid: return new_oid raise AuthorityGitError("AUTHORITY_BRANCH_EXPECTATION_MISMATCH") if not self.ancestry_check(expected_oid, new_oid): raise AuthorityGitError("NEW_OID_NOT_DESCENDANT") self._git(["update-ref", branch, new_oid, expected_oid]) readback = self.read_ref(branch) if readback != new_oid: raise AuthorityGitError("AUTHORITY_CAS_READBACK_MISMATCH") return readback def ff_CAS_main_after_verified_protected_merge(self, expected_oid: str, new_oid: str) -> str: _oid(expected_oid, "expected_oid") _oid(new_oid, "new_oid") if not self.ancestry_check(expected_oid, new_oid): raise AuthorityGitError("MAIN_FF_REQUIRED") current = self.read_ref("refs/heads/main") if current != expected_oid: if current == new_oid: return new_oid raise AuthorityGitError("AUTHORITY_MAIN_EXPECTATION_MISMATCH") self._git(["update-ref", "refs/heads/main", new_oid, expected_oid]) readback = self.read_ref("refs/heads/main") if readback != new_oid: raise AuthorityGitError("AUTHORITY_MAIN_CAS_READBACK_MISMATCH") return readback def commit_file_in_isolated_worktree( self, *, base_oid: str, files: Mapping[str, bytes], message: str, worktree_name: str, ) -> str: _oid(base_oid, "base_oid") if not worktree_name or "/" in worktree_name or "\\" in worktree_name or worktree_name in {".", ".."}: raise AuthorityGitError("WORKTREE_NAME_INVALID") root = self.worktree_root / worktree_name if root.exists() or root.is_symlink(): raise AuthorityGitError("WORKTREE_ALREADY_EXISTS") self.worktree_root.mkdir(parents=True, exist_ok=True) self._git(["worktree", "add", "--detach", str(root), base_oid]) try: for relative, content in files.items(): path = Path(relative) if path.is_absolute() or ".." in path.parts or not path.parts or path.parts[0] == ".git": raise AuthorityGitError("FILE_PATH_INVALID") target = root / path target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(content) result = subprocess.run( ["git", "-C", str(root), "add", "--", *files.keys()], text=True, capture_output=True, check=False, ) if result.returncode != 0: raise AuthorityGitError("WORKTREE_ADD_FAILED") result = subprocess.run( ["git", "-C", str(root), "-c", "user.name=aios-durable-job", "-c", "user.email=aios-durable-job@localhost", "commit", "--no-verify", "-m", message], text=True, capture_output=True, check=False, ) if result.returncode != 0: raise AuthorityGitError("WORKTREE_COMMIT_FAILED") return subprocess.check_output(["git", "-C", str(root), "rev-parse", "HEAD"], text=True).strip() finally: subprocess.run(["git", f"--git-dir={self.git_dir}", "worktree", "remove", "--force", str(root)], capture_output=True, check=False)