dongarandClaude Opus 5
2071e6ab6d
The standalone half of the Gitea layer; orchestrator wiring follows once the executor loop lands, to avoid colliding with it in control.py. - gitea.py: GiteaClient over httpx — open PRs, list and post issue comments, close PRs. find_pull_request is the A3 external-truth query that makes opening a PR idempotent across a crash, and it checks closed PRs too, since a crash after opening must not be answered with "nothing is open, so open another" - pull_request_reconciler: an unreachable Gitea returns None rather than a guess, so the agent gates to the human instead of risking a duplicate side effect - git.py: branch, commit, push for the workspace. The token is passed per-invocation rather than stored as a remote, so it never lands in .git/config, and anything raised is scrubbed of it - commit_all returns None when there is nothing to commit: an empty milestone is a fact to report, not an error - Live test does the real round trip against the sandbox repo and cleans up after itself Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
97 lines
3.4 KiB
Python
97 lines
3.4 KiB
Python
"""Git operations for the agent's workspace (spec B2 outbound).
|
|
|
|
Push authentication uses the Gitea service-account token embedded in the
|
|
remote URL. The token is never written to disk in the repo config and is
|
|
scrubbed from anything this module raises or returns, so a failed push cannot
|
|
leak the credential into a transcript, a log line, or a PR comment.
|
|
"""
|
|
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
TIMEOUT_S = 120
|
|
|
|
|
|
class GitError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GitRepo:
|
|
root: Path
|
|
|
|
def _run(self, *args: str, secret: str | None = None) -> str:
|
|
try:
|
|
completed = subprocess.run(
|
|
["git", *args],
|
|
cwd=self.root,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=TIMEOUT_S,
|
|
check=False,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
raise GitError(f"git {args[0]} failed: {_scrub(str(exc), secret)}") from exc
|
|
if completed.returncode != 0:
|
|
raise GitError(
|
|
f"git {args[0]} failed: "
|
|
f"{_scrub(completed.stderr.strip() or completed.stdout.strip(), secret)}"
|
|
)
|
|
return completed.stdout.strip()
|
|
|
|
# -- inspection -------------------------------------------------------
|
|
|
|
def current_branch(self) -> str:
|
|
# --show-current rather than rev-parse: it answers correctly on a
|
|
# freshly initialised repo, where HEAD points at an unborn branch.
|
|
return self._run("branch", "--show-current")
|
|
|
|
def head_sha(self) -> str:
|
|
return self._run("rev-parse", "HEAD")
|
|
|
|
def has_changes(self) -> bool:
|
|
return bool(self._run("status", "--porcelain"))
|
|
|
|
# -- mutation ---------------------------------------------------------
|
|
|
|
def init(self, *, default_branch: str = "main") -> None:
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
self._run("init", "-q", "-b", default_branch)
|
|
|
|
def configure_identity(self, name: str, email: str) -> None:
|
|
self._run("config", "user.name", name)
|
|
self._run("config", "user.email", email)
|
|
|
|
def checkout_new_branch(self, name: str) -> None:
|
|
self._run("checkout", "-q", "-b", name)
|
|
|
|
def commit_all(self, message: str) -> str | None:
|
|
"""Returns the new commit sha, or None when there was nothing to
|
|
commit — an empty milestone is a fact to report, not an error."""
|
|
if not self.has_changes():
|
|
return None
|
|
self._run("add", "-A")
|
|
self._run("commit", "-q", "-m", message)
|
|
return self.head_sha()
|
|
|
|
def push(self, remote_url: str, branch: str, *, token: str | None = None) -> None:
|
|
# The URL is passed per-invocation rather than stored as a remote, so
|
|
# the token never lands in .git/config.
|
|
self._run("push", "-q", remote_url, f"HEAD:refs/heads/{branch}", secret=token)
|
|
|
|
def delete_remote_branch(
|
|
self, remote_url: str, branch: str, *, token: str | None = None
|
|
) -> None:
|
|
self._run("push", "-q", remote_url, f":refs/heads/{branch}", secret=token)
|
|
|
|
|
|
def authenticated_url(base_url: str, repo: str, user: str, token: str) -> str:
|
|
host = base_url.split("://", 1)[-1].rstrip("/")
|
|
scheme = base_url.split("://", 1)[0] if "://" in base_url else "https"
|
|
return f"{scheme}://{user}:{token}@{host}/{repo}.git"
|
|
|
|
|
|
def _scrub(text: str, secret: str | None) -> str:
|
|
return text.replace(secret, "***") if secret else text
|