dongarandClaude Opus 5
2fb25b8195
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m14s
Nothing had ever assembled the agent. main.py was still the hello-world placeholder from the scaffold, and no code outside the test suite constructed an Orchestrator — which is how two fully built, fully tested components came to be reachable from nothing: the delegation harness, and the Gitea comment poller. Unit tests that construct a component directly cannot catch that; only assembly can. - config.py validates every variable before anything is built and names all of them at once, rather than failing at first use twenty minutes into a milestone on the call that mattered. Whitespace-only counts as missing, since a token of spaces passes a truthiness check and fails at the first request instead - agent.py builds the whole graph and owns startup and shutdown order: reconcile before accepting commands, and on stop, stop watching, then stop talking, then close the store last - The Gitea account is read via whoami rather than configured: it names the identity every permission is evaluated against, and proves the token works before anything depends on it - Poller lifecycle is now a state-machine concern: watch the plan issue while a gate stands, the PR while it is open, and stop on leaving. The orchestrator owns when to watch; an injected factory owns how to build one - The executor reports each turn, because it runs synchronously inside the transport callback: without it the agent is silent for a whole milestone and a working run looks exactly like a hung one - main.py loads .env, builds, and shuts down cleanly on SIGINT/SIGTERM The entry-point tests changed shape: nisse-dev no longer greets, it refuses to start unconfigured and says everything that is missing. Both tests now run with a deliberately empty environment and a .env path that does not exist, because the working directory's .env holds live credentials and a test must never be one cwd away from starting a real agent. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
121 lines
4.4 KiB
Python
121 lines
4.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 clone(self, remote_url: str, dest: Path, *, token: str | None = None) -> None:
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
self._run("clone", "-q", remote_url, str(dest), secret=token)
|
|
|
|
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 checkout_existing_branch(self, name: str) -> None:
|
|
self._run("checkout", "-q", name)
|
|
|
|
def branch_exists(self, name: str) -> bool:
|
|
try:
|
|
self._run("rev-parse", "--verify", "--quiet", f"refs/heads/{name}")
|
|
except GitError:
|
|
return False
|
|
return True
|
|
|
|
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 discard_branch(self, branch: str, *, back_to: str) -> None:
|
|
"""Throw away a branch and everything on it.
|
|
|
|
Used when a delegation aborts: A13 gives a delegation the same
|
|
atomicity that reconciliation gives an interrupted transition, and
|
|
half-finished work must not survive to be built upon.
|
|
"""
|
|
self._run("checkout", "-q", "--force", back_to)
|
|
self._run("branch", "-q", "-D", branch)
|
|
|
|
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
|