aios/active-runtime/app/aios_authority_git_executor.py

"""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 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 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)