Add controlled coding agent automation

src/lai_automation/git.py

@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+import os
+import pwd
+import shlex
+import shutil
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from urllib.parse import urlparse
+
+from .config import Settings
+
+
+@dataclass(frozen=True)
+class ChangeSummary:
+    files: tuple[str, ...]
+    added_lines: int
+    deleted_lines: int
+
+    @property
+    def changed_lines(self) -> int:
+        return self.added_lines + self.deleted_lines
+
+
+class GitWorkspace:
+    def __init__(self, settings: Settings, report_id: int) -> None:
+        self.settings = settings
+        self.path = settings.workspace_root / f"failure-{report_id}"
+        self.branch = f"ai/failure-{report_id}"
+
+    def prepare(self) -> None:
+        repository = urlparse(self.settings.source_repository)
+        gitea = urlparse(self.settings.gitea_url)
+        if repository.scheme != "https" or repository.hostname != gitea.hostname:
+            raise ValueError("SOURCE_REPOSITORY must use HTTPS on the configured Gitea host")
+        shutil.rmtree(self.path, ignore_errors=True)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._git("clone", "--branch", self.settings.source_branch, "--single-branch", self.settings.source_repository, str(self.path), cwd=self.path.parent)
+        self._git("checkout", "-b", self.branch)
+
+    def run(
+        self,
+        command: str,
+        check: bool = True,
+        username: str | None = None,
+    ) -> subprocess.CompletedProcess[str]:
+        environment = None
+        drop_privileges = None
+        if username:
+            account = pwd.getpwnam(username)
+            environment = {
+                "HOME": account.pw_dir,
+                "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"),
+            }
+
+            def drop_privileges() -> None:
+                os.setgroups([])
+                os.setgid(account.pw_gid)
+                os.setuid(account.pw_uid)
+
+        return subprocess.run(
+            shlex.split(command),
+            cwd=self.path,
+            env=environment,
+            text=True,
+            capture_output=True,
+            check=check,
+            timeout=1800,
+            preexec_fn=drop_privileges,
+        )
+
+    def summary(self) -> ChangeSummary:
+        self._git("add", "-N", "--all")
+        output = self._git("diff", "--numstat").stdout
+        files: list[str] = []
+        added = deleted = 0
+        for line in output.splitlines():
+            add, remove, filename = line.split("\t", 2)
+            files.append(filename)
+            added += int(add) if add.isdigit() else 0
+            deleted += int(remove) if remove.isdigit() else 0
+        return ChangeSummary(tuple(files), added, deleted)
+
+    def commit(self, report_id: int) -> str:
+        self._git("add", "--all")
+        self._git("commit", "-m", f"Fix failure report #{report_id}")
+        return self._git("rev-parse", "HEAD").stdout.strip()
+
+    def push_branch(self) -> None:
+        self._git_authenticated("push", "origin", f"HEAD:{self.branch}")
+
+    def push_main(self) -> None:
+        self._git_authenticated("push", "origin", f"HEAD:{self.settings.source_branch}")
+
+    def cleanup(self) -> None:
+        if self.path.exists():
+            shutil.rmtree(self.path)
+
+    def _git(self, *args: str, cwd: Path | None = None) -> subprocess.CompletedProcess[str]:
+        return subprocess.run(
+            ["git", *args],
+            cwd=cwd or self.path,
+            text=True,
+            capture_output=True,
+            check=True,
+            timeout=300,
+        )
+
+    def _git_authenticated(self, *args: str) -> subprocess.CompletedProcess[str]:
+        askpass = self.path / ".git" / "lai-askpass.sh"
+        username_file = self.path / ".git" / "lai-bot-user"
+        token_file = self.path / ".git" / "lai-bot-token"
+        try:
+            username_file.write_text(self.settings.gitea_username)
+            token_file.write_text(self.settings.gitea_token)
+            username_file.chmod(0o600)
+            token_file.chmod(0o600)
+            askpass.write_text(
+                "#!/bin/sh\ncase \"$1\" in *Username*) cat \"$GIT_BOT_USER_FILE\";; *) cat \"$GIT_BOT_TOKEN_FILE\";; esac\n"
+            )
+            askpass.chmod(0o700)
+            env = {
+                "HOME": os.environ.get("HOME", "/root"),
+                "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
+                "GIT_ASKPASS": str(askpass),
+                "GIT_TERMINAL_PROMPT": "0",
+                "GIT_BOT_USER_FILE": str(username_file),
+                "GIT_BOT_TOKEN_FILE": str(token_file),
+            }
+            return subprocess.run(
+                ["git", *args],
+                cwd=self.path,
+                env=env,
+                text=True,
+                capture_output=True,
+                check=True,
+                timeout=300,
+            )
+        finally:
+            askpass.unlink(missing_ok=True)
+            username_file.unlink(missing_ok=True)
+            token_file.unlink(missing_ok=True)