dongarandClaude Opus 5.5
ab469b60bb
A project's trust.merge is operator (the default) or approved (L3, D5, #65). Set to approved, the harness, never the model, merges a PR in the PR watch once an approving review from a gitea.operators account is on its current commit, its checks pass (none at all counts; pending waits) and it's mergeable; the item then goes on as after any merge. Each reason to hold off (failed checks, conflicts, a refused merge) is said once per commit. Gitea enforces it: protection requires approvals from an operators-only whitelist and dismisses stale ones. At start, for each approved project, the agent checks by behaviour that its own unapproved merge is refused, with a throwaway branch and PR made through the API (never the workspace); if it isn't refused, the operator merges until a restart. - gitea: approvals, combined check status, a PR's head commit and mergeability, API-made branches and their deletion. - Test and bring-up repositories get the protection. - Tests: the decision (unit), the hard wall on a real Gitea (test_trust_live), an e2e test where the operator only approves, and a walk property, merges_only_approved, checked against a mutant that ignores which commit was approved. Closes #65 Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
400 lines
14 KiB
Python
400 lines
14 KiB
Python
"""Gitea client (architecture D4, D5).
|
|
|
|
Outbound: open PRs and post comments through a dedicated service account with
|
|
a scoped token. Inbound is deliberately *polled* rather than webhook-driven,
|
|
so the agent VM stays outbound-only (D7), and intake is structured JSON —
|
|
never routed through a chat bridge.
|
|
|
|
``find_pull_request`` is the external-truth query that makes opening a PR
|
|
idempotent across a crash: boot reconciliation (L8) asks Gitea whether the
|
|
effect already happened rather than repeating it.
|
|
"""
|
|
|
|
import base64
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from nisse_dev.reconcile import Reconciler
|
|
from nisse_dev.state import Intent
|
|
|
|
PR_PAGE_SIZE = 50
|
|
MAX_PR_PAGES = 40 # 2000 PRs; beyond that, say so rather than guess
|
|
|
|
|
|
class GiteaError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PullRequest:
|
|
number: int
|
|
state: str
|
|
head: str
|
|
url: str
|
|
merged: bool = False
|
|
head_sha: str = "" # the commit the PR is at now
|
|
mergeable: bool | None = None # None: Gitea didn't say
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Approval:
|
|
"""An approving review, and the commit it approved (L3: approvals bind
|
|
to exactly what was approved)."""
|
|
|
|
author: str
|
|
commit: str
|
|
stale: bool = False # Gitea's mark: commits were pushed since
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReviewNote:
|
|
"""Something a reviewer said in a review: its body, or a comment on a
|
|
line (#69). ``key`` is unique across both kinds."""
|
|
|
|
key: str
|
|
author: str
|
|
body: str
|
|
where: str # "path:line" for a line comment, "" for a review body
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Comment:
|
|
id: int
|
|
author: str
|
|
body: str
|
|
created_at: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Issue:
|
|
number: int
|
|
title: str
|
|
body: str
|
|
author: str
|
|
labels: tuple[str, ...]
|
|
|
|
|
|
class GiteaClient:
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
token: str,
|
|
*,
|
|
timeout_s: float = 30.0,
|
|
transport: httpx.BaseTransport | None = None,
|
|
) -> None:
|
|
self._http = httpx.Client(
|
|
base_url=f"{base_url.rstrip('/')}/api/v1",
|
|
headers={"Authorization": f"token {token}"},
|
|
timeout=timeout_s,
|
|
transport=transport,
|
|
)
|
|
|
|
def _request(self, method: str, path: str, **kwargs: Any) -> Any:
|
|
try:
|
|
response = self._http.request(method, path, **kwargs)
|
|
except httpx.HTTPError as exc:
|
|
raise GiteaError(f"gitea request failed: {exc}") from exc
|
|
if response.status_code >= 400:
|
|
raise GiteaError(
|
|
f"gitea {method} {path} returned {response.status_code}: "
|
|
f"{response.text[:300]}"
|
|
)
|
|
return response.json() if response.content else None
|
|
|
|
def ssh_url(self, repo: str) -> str:
|
|
"""The repository's SSH clone URL, which carries no credential."""
|
|
return str(self._request("GET", f"/repos/{repo}")["ssh_url"])
|
|
|
|
def clone_url(self, repo: str) -> str:
|
|
"""The repository's HTTPS clone URL, which carries no credential."""
|
|
return str(self._request("GET", f"/repos/{repo}")["clone_url"])
|
|
|
|
def whoami(self) -> str:
|
|
return str(self._request("GET", "/user")["login"])
|
|
|
|
def open_pull_request(
|
|
self, repo: str, *, head: str, base: str, title: str, body: str
|
|
) -> PullRequest:
|
|
data = self._request(
|
|
"POST",
|
|
f"/repos/{repo}/pulls",
|
|
json={"head": head, "base": base, "title": title, "body": body},
|
|
)
|
|
return _pull_request(data)
|
|
|
|
def pull_request(self, repo: str, index: int) -> PullRequest:
|
|
return _pull_request(self._request("GET", f"/repos/{repo}/pulls/{index}"))
|
|
|
|
def review_notes(self, repo: str, index: int) -> list[ReviewNote]:
|
|
"""Review bodies and inline line comments on a PR (#69): the PR's
|
|
conversation comments come from list_comments."""
|
|
notes: list[ReviewNote] = []
|
|
for review in (
|
|
self._request("GET", f"/repos/{repo}/pulls/{index}/reviews") or []
|
|
):
|
|
author = (review.get("user") or {}).get("login", "")
|
|
body = str(review.get("body") or "").strip()
|
|
if body:
|
|
notes.append(ReviewNote(f"review-{review['id']}", author, body, ""))
|
|
if not review.get("comments_count"):
|
|
continue
|
|
for comment in (
|
|
self._request(
|
|
"GET",
|
|
f"/repos/{repo}/pulls/{index}/reviews/{review['id']}/comments",
|
|
)
|
|
or []
|
|
):
|
|
line = comment.get("line") or comment.get("position") or ""
|
|
notes.append(
|
|
ReviewNote(
|
|
key=f"line-{comment['id']}",
|
|
author=(comment.get("user") or {}).get("login", author),
|
|
body=str(comment.get("body") or ""),
|
|
where=f"{comment.get('path', '')}:{line}",
|
|
)
|
|
)
|
|
return notes
|
|
|
|
def find_pull_request(self, repo: str, head: str) -> PullRequest | None:
|
|
"""External truth for L8: did the PR for this branch already get
|
|
opened? Includes closed PRs — a crash after opening must not be
|
|
answered with "no PR is open, so open another".
|
|
|
|
Pages to the end rather than sampling the first page: a "not found"
|
|
that only means "not in the newest 50" would licence a duplicate PR.
|
|
If the history is longer than this walk, raise instead of guessing —
|
|
the reconciler turns that into "unverifiable" and gates to the human.
|
|
"""
|
|
for page in range(1, MAX_PR_PAGES + 1):
|
|
data = self._request(
|
|
"GET",
|
|
f"/repos/{repo}/pulls",
|
|
params={"state": "all", "limit": PR_PAGE_SIZE, "page": page},
|
|
)
|
|
items = data or []
|
|
for item in items:
|
|
if (item.get("head") or {}).get("ref") == head:
|
|
return _pull_request(item)
|
|
if len(items) < PR_PAGE_SIZE:
|
|
return None # reached the end: definitively absent
|
|
raise GiteaError(
|
|
f"searched {MAX_PR_PAGES * PR_PAGE_SIZE} pull requests in {repo} "
|
|
f"without reaching the end; cannot confirm whether {head!r} was "
|
|
"already opened"
|
|
)
|
|
|
|
def list_comments(
|
|
self, repo: str, index: int, since: str | None = None
|
|
) -> list[Comment]:
|
|
params: dict[str, str] = {}
|
|
if since is not None:
|
|
params["since"] = since
|
|
data = self._request(
|
|
"GET", f"/repos/{repo}/issues/{index}/comments", params=params
|
|
)
|
|
return [
|
|
Comment(
|
|
id=item["id"],
|
|
author=(item.get("user") or {}).get("login", ""),
|
|
body=item.get("body", ""),
|
|
created_at=item.get("created_at", ""),
|
|
)
|
|
for item in data or []
|
|
]
|
|
|
|
def post_comment(self, repo: str, index: int, body: str) -> Comment:
|
|
data = self._request(
|
|
"POST", f"/repos/{repo}/issues/{index}/comments", json={"body": body}
|
|
)
|
|
return Comment(
|
|
id=data["id"],
|
|
author=(data.get("user") or {}).get("login", ""),
|
|
body=data.get("body", ""),
|
|
created_at=data.get("created_at", ""),
|
|
)
|
|
|
|
def create_issue(self, repo: str, title: str, body: str) -> int:
|
|
data = self._request(
|
|
"POST", f"/repos/{repo}/issues", json={"title": title, "body": body}
|
|
)
|
|
return int(data["number"])
|
|
|
|
def close_pull_request(self, repo: str, index: int) -> None:
|
|
self._request("PATCH", f"/repos/{repo}/pulls/{index}", json={"state": "closed"})
|
|
|
|
def assigned_issues(self, repo: str, assignee: str) -> list[Issue]:
|
|
"""Open issues (not PRs) assigned to ``assignee``: intake (D1)."""
|
|
data = self._request(
|
|
"GET",
|
|
f"/repos/{repo}/issues",
|
|
params={
|
|
"type": "issues",
|
|
"state": "open",
|
|
"assigned_by": assignee,
|
|
"limit": 50,
|
|
},
|
|
)
|
|
return [
|
|
Issue(
|
|
number=int(item["number"]),
|
|
title=str(item.get("title") or ""),
|
|
body=str(item.get("body") or ""),
|
|
author=str(item["user"]["login"]),
|
|
labels=tuple(str(label["name"]) for label in item.get("labels") or []),
|
|
)
|
|
for item in data or []
|
|
if not item.get("pull_request")
|
|
]
|
|
|
|
def open_dependencies(self, repo: str, index: int) -> list[str]:
|
|
"""The issues this one depends on that are still open, as
|
|
"owner/repo#N": what blocks it (L1)."""
|
|
data = self._request("GET", f"/repos/{repo}/issues/{index}/dependencies")
|
|
return [
|
|
f"{(item.get('repository') or {}).get('full_name', repo)}#{item['number']}"
|
|
for item in data or []
|
|
if item.get("state") == "open"
|
|
]
|
|
|
|
def assigned_by(self, repo: str, index: int, assignee: str) -> str | None:
|
|
"""Who last assigned the issue to ``assignee``, from its timeline;
|
|
None if the timeline doesn't say."""
|
|
events = self._request("GET", f"/repos/{repo}/issues/{index}/timeline") or []
|
|
who = None
|
|
for event in events:
|
|
if (
|
|
event.get("type") == "assignees"
|
|
and not event.get("removed_assignee")
|
|
and (event.get("assignee") or {}).get("login") == assignee
|
|
):
|
|
who = (event.get("user") or {}).get("login")
|
|
return who
|
|
|
|
def assigned_at(self, repo: str, index: int, assignee: str) -> str | None:
|
|
"""When the issue was last assigned to ``assignee`` (ISO 8601), from
|
|
its timeline, oldest first and paged to the end; None if the
|
|
timeline doesn't say."""
|
|
when = None
|
|
for page in range(1, MAX_PR_PAGES + 1):
|
|
events = (
|
|
self._request(
|
|
"GET",
|
|
f"/repos/{repo}/issues/{index}/timeline",
|
|
params={"limit": PR_PAGE_SIZE, "page": page},
|
|
)
|
|
or []
|
|
)
|
|
for event in events:
|
|
if (
|
|
event.get("type") == "assignees"
|
|
and not event.get("removed_assignee")
|
|
and (event.get("assignee") or {}).get("login") == assignee
|
|
):
|
|
when = event.get("created_at")
|
|
if len(events) < PR_PAGE_SIZE:
|
|
break
|
|
return when
|
|
|
|
def approvals(self, repo: str, index: int) -> list[Approval]:
|
|
"""The PR's approving reviews that still stand: not dismissed."""
|
|
return [
|
|
Approval(
|
|
author=(review.get("user") or {}).get("login", ""),
|
|
commit=str(review.get("commit_id") or ""),
|
|
stale=bool(review.get("stale")),
|
|
)
|
|
for review in (
|
|
self._request("GET", f"/repos/{repo}/pulls/{index}/reviews") or []
|
|
)
|
|
if review.get("state") == "APPROVED" and not review.get("dismissed")
|
|
]
|
|
|
|
def checks(self, repo: str, sha: str) -> str:
|
|
"""A commit's combined check status: success, pending, failure or
|
|
error; "none" when no check reported on it at all."""
|
|
data = self._request("GET", f"/repos/{repo}/commits/{sha}/status") or {}
|
|
if not data.get("total_count") and not data.get("statuses"):
|
|
return "none"
|
|
return str(data.get("state") or "pending")
|
|
|
|
def branch_with_file(
|
|
self, repo: str, *, base: str, branch: str, path: str, text: str, message: str
|
|
) -> None:
|
|
"""A new branch off ``base`` with one file added, made through the
|
|
API: no clone, no workspace."""
|
|
self._request(
|
|
"POST",
|
|
f"/repos/{repo}/contents/{path}",
|
|
json={
|
|
"content": base64.b64encode(text.encode()).decode(),
|
|
"message": message,
|
|
"branch": base,
|
|
"new_branch": branch,
|
|
},
|
|
)
|
|
|
|
def delete_branch(self, repo: str, name: str) -> None:
|
|
self._request("DELETE", f"/repos/{repo}/branches/{name}")
|
|
|
|
def repository(self, repo: str) -> dict[str, Any]:
|
|
"""The repository as this account sees it, permissions included."""
|
|
return dict(self._request("GET", f"/repos/{repo}"))
|
|
|
|
def branch(self, repo: str, name: str) -> dict[str, Any]:
|
|
"""A branch as this account sees it: protected, user_can_push."""
|
|
return dict(self._request("GET", f"/repos/{repo}/branches/{name}"))
|
|
|
|
def merge_pull_request(self, repo: str, index: int) -> None:
|
|
self._request(
|
|
"POST", f"/repos/{repo}/pulls/{index}/merge", json={"Do": "merge"}
|
|
)
|
|
|
|
def assign_issue(self, repo: str, index: int, assignee: str) -> None:
|
|
self._request(
|
|
"PATCH", f"/repos/{repo}/issues/{index}", json={"assignees": [assignee]}
|
|
)
|
|
|
|
def close_issue(self, repo: str, index: int) -> None:
|
|
self._request(
|
|
"PATCH", f"/repos/{repo}/issues/{index}", json={"state": "closed"}
|
|
)
|
|
|
|
def close(self) -> None:
|
|
self._http.close()
|
|
|
|
|
|
def _pull_request(data: dict[str, Any]) -> PullRequest:
|
|
return PullRequest(
|
|
number=data["number"],
|
|
state=data.get("state", ""),
|
|
head=(data.get("head") or {}).get("ref", ""),
|
|
url=data.get("html_url", ""),
|
|
merged=bool(data.get("merged")),
|
|
head_sha=(data.get("head") or {}).get("sha", ""),
|
|
mergeable=data.get("mergeable"),
|
|
)
|
|
|
|
|
|
def pull_request_reconciler(client: GiteaClient, repo: str) -> Reconciler:
|
|
"""Reconciler for an ``open_pr`` intent (L8 boot step 2).
|
|
|
|
An unreachable Gitea yields None — unverifiable, so the agent gates to
|
|
the human rather than risking a duplicate PR. That is the conservative
|
|
direction: a delayed decision costs nothing, a repeated side effect does.
|
|
"""
|
|
|
|
def check(intent: Intent) -> bool | None:
|
|
branch = intent.payload.get("branch")
|
|
if not branch:
|
|
return None
|
|
try:
|
|
return client.find_pull_request(repo, branch) is not None
|
|
except GiteaError:
|
|
return None
|
|
|
|
return check
|