Add Gitea client and git operations (B2) — step 6a, part 1
CI / checks (push) Successful in 16s
CI / checks (pull_request) Successful in 14s

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>
This commit is contained in:
dongarandClaude Opus 5 committed 2026-09-07 20:27:34 +00:00
1 parent 47b2bdfdbd
commit 2071e6ab6d
7 files changed
+614

No files matched your search

+96
View File
@@ -0,0 +1,96 @@
"""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