ee902354c2
All checks were successful
Tests / Declarative: Post Actions passed: 1387
The pull request named the incident - sonar/ariadne/AZ2y0FYFKy9i4pkIpNlV - and left the reviewer to find what that key meant. Their first question is what the finding actually said: which rule, how severe, what the offending lines look like in context. An opaque key is not an answer to that. The link is derived from the incident id rather than threaded through the flow. The id already carries the project and the key, so a second path for them to travel would only be a second thing that can disagree with the first. A build-driven repair has no finding, so it gets no line. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
378 lines
14 KiB
Python
378 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import re
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from ..utils.logging import get_logger
|
|
from . import hermes_sonar_client
|
|
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
HTTP_OK = 200
|
|
HTTP_CREATED = 201
|
|
HTTP_NOT_FOUND = 404
|
|
HTTP_CONFLICT = 409
|
|
HTTP_UNPROCESSABLE = 422
|
|
|
|
_DEFAULT_TIMEOUT_SECONDS = 15.0
|
|
_PROTECTED_BRANCHES = {"master", "main"}
|
|
_REPAIR_BRANCH_PREFIX = "hermes-repair/"
|
|
_OPEN_PULLS_LIMIT = 50
|
|
_COMMIT_IDENTITY = {"name": "Hermes Agent", "email": "hermes@bstein.dev"}
|
|
_COMMIT_OK_STATUSES = {HTTP_OK, HTTP_CREATED}
|
|
# The Hermes console reopens a finished run from its id at this route.
|
|
RUN_PATH = "/chat?resume="
|
|
_BRANCH_RETRY_STATUSES = {HTTP_NOT_FOUND, HTTP_UNPROCESSABLE}
|
|
_BRANCH_UNSAFE = re.compile(r"[^A-Za-z0-9._-]+")
|
|
|
|
|
|
def fetch_file(cfg: dict, path: str) -> tuple[str | None, str | None]:
|
|
"""Fetch one file's raw contents from Gitea at the base branch.
|
|
|
|
Inputs: `cfg` with gitea_base_url, gitea_token, owner, repo, base_branch,
|
|
and timeout_seconds; the repository-relative file path.
|
|
Outputs: (contents, error) with exactly one side set. Never raises and
|
|
never logs the token.
|
|
"""
|
|
|
|
base_url = _base_url(cfg)
|
|
if not base_url:
|
|
return None, "gitea base url is empty"
|
|
url = f"{base_url}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/raw/{path}"
|
|
try:
|
|
with httpx.Client(timeout=_timeout(cfg)) as client:
|
|
response = client.get(url, headers=_headers(cfg), params={"ref": _base_branch(cfg)})
|
|
except Exception as exc:
|
|
return None, f"file fetch failed: {exc}"
|
|
if response.status_code != HTTP_OK:
|
|
return None, f"file fetch http {response.status_code}"
|
|
return response.text, None
|
|
|
|
|
|
def find_open_proposal(cfg: dict) -> dict[str, Any]:
|
|
"""Find an already-open Hermes repair pull request for this repository.
|
|
|
|
Inputs: `cfg` as for `fetch_file`. Outputs:
|
|
{"found", "pr_number", "url", "branch", "error"}. An open pull request
|
|
counts as an existing proposal when its head branch starts with
|
|
`hermes-repair/` and its base branch equals cfg["base_branch"]; the
|
|
lowest-numbered match (the original proposal) is returned so repeated
|
|
checks stay stable while the branch sits unmerged.
|
|
|
|
Fails open by design: every HTTP, parse, or transport failure returns
|
|
found=False with `error` set, because a false "duplicate" would silently
|
|
suppress legitimate repair work — a missed duplicate is only noise, a
|
|
wrong duplicate is lost work. Never raises and never logs the token.
|
|
"""
|
|
|
|
base_url = _base_url(cfg)
|
|
if not base_url:
|
|
return _no_proposal("gitea base url is empty")
|
|
try:
|
|
with httpx.Client(timeout=_timeout(cfg)) as client:
|
|
response = client.get(
|
|
f"{base_url}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/pulls",
|
|
headers=_headers(cfg),
|
|
params={"state": "open", "limit": _OPEN_PULLS_LIMIT},
|
|
)
|
|
except Exception as exc:
|
|
return _no_proposal(f"open proposal lookup failed: {exc}")
|
|
if response.status_code != HTTP_OK:
|
|
return _no_proposal(f"open proposal lookup http {response.status_code}")
|
|
return _oldest_repair_pull(response, _base_branch(cfg))
|
|
|
|
|
|
def push_branch(
|
|
cfg: dict, incident_id: str, ref: Any, patch: Any, patched_contents: str
|
|
) -> dict[str, Any]:
|
|
"""Commit the patched file to a fresh repair branch via the Gitea API.
|
|
|
|
Inputs: `cfg` as for `fetch_file`; the incident id; `ref`, the token that
|
|
names the branch - the failed build number for a build-driven proposal, or
|
|
a sweep token when the proposal did not come from a build; the validated
|
|
ProposedPatch; and the fully patched file contents. Outputs:
|
|
{"branch", "committed", "error"}. Never force-pushes, refuses master/main
|
|
or empty branch names, never raises, and never logs the token.
|
|
"""
|
|
|
|
branch = _branch_name(ref)
|
|
if not branch.strip() or branch in _PROTECTED_BRANCHES:
|
|
return {"branch": branch, "committed": False, "error": f"refusing branch {branch!r}"}
|
|
base_url = _base_url(cfg)
|
|
if not base_url:
|
|
return {"branch": branch, "committed": False, "error": "gitea base url is empty"}
|
|
try:
|
|
with httpx.Client(timeout=_timeout(cfg)) as client:
|
|
error = _commit_patch(client, cfg, branch, incident_id, patch, patched_contents)
|
|
except Exception as exc:
|
|
error = f"branch push failed: {exc}"
|
|
if error is not None:
|
|
logger.info(
|
|
"hermes code repair branch push failed",
|
|
extra={"event": "hermes_code_repair", "status": "error", "branch": branch, "detail": error},
|
|
)
|
|
return {"branch": branch, "committed": error is None, "error": error}
|
|
|
|
|
|
def open_pull_request( # noqa: PLR0913 - the body needs the full proposal provenance
|
|
cfg: dict, incident_id: str, run_id: str, branch: str, patch: Any, analysis: str
|
|
) -> dict[str, Any]:
|
|
"""Open the human-review pull request for a pushed repair branch.
|
|
|
|
Inputs: `cfg` as for `fetch_file` plus optional `hermes_ui_url`; the
|
|
incident id; the Hermes run that produced the patch; the pushed branch
|
|
name; the validated ProposedPatch; and the model analysis.
|
|
Outputs: {"pr_number", "url", "error"}; an existing PR reported by a 409
|
|
counts as success when the payload identifies it. Never raises and never
|
|
logs the token.
|
|
"""
|
|
|
|
base_url = _base_url(cfg)
|
|
if not base_url:
|
|
return {"pr_number": None, "url": None, "error": "gitea base url is empty"}
|
|
payload = {
|
|
"head": branch,
|
|
"base": _base_branch(cfg),
|
|
"title": f"fix(hermes): repair {incident_id}",
|
|
"body": _pr_body(incident_id, patch, analysis, run_id, cfg),
|
|
}
|
|
try:
|
|
with httpx.Client(timeout=_timeout(cfg)) as client:
|
|
response = client.post(
|
|
f"{base_url}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/pulls",
|
|
headers=_headers(cfg),
|
|
json=payload,
|
|
)
|
|
except Exception as exc:
|
|
return {"pr_number": None, "url": None, "error": f"pull request failed: {exc}"}
|
|
if response.status_code == HTTP_CREATED:
|
|
return _pr_result(response)
|
|
if response.status_code == HTTP_CONFLICT:
|
|
existing = _pr_result(response)
|
|
if existing["pr_number"] is not None:
|
|
return existing
|
|
return {"pr_number": None, "url": None, "error": "pull request already exists"}
|
|
return {"pr_number": None, "url": None, "error": f"pull request http {response.status_code}"}
|
|
|
|
|
|
def _commit_patch( # noqa: PLR0913 - commit needs the full branch/identity context
|
|
client: httpx.Client,
|
|
cfg: dict,
|
|
branch: str,
|
|
incident_id: str,
|
|
patch: Any,
|
|
patched_contents: str,
|
|
) -> str | None:
|
|
"""PUT the patched file onto the new branch, trying both branch shapes."""
|
|
|
|
url = f"{_base_url(cfg)}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/contents/{patch.path}"
|
|
sha, sha_error = _file_sha(client, cfg, patch.path)
|
|
if sha_error is not None:
|
|
return sha_error
|
|
body = {
|
|
"message": f"fix(hermes): {patch.rationale} (incident {incident_id})",
|
|
"content": base64.b64encode(patched_contents.encode("utf-8")).decode("ascii"),
|
|
"sha": sha,
|
|
"author": dict(_COMMIT_IDENTITY),
|
|
"committer": dict(_COMMIT_IDENTITY),
|
|
}
|
|
first = client.put(url, headers=_headers(cfg), json={**body, "branch": _base_branch(cfg), "new_branch": branch})
|
|
if first.status_code in _COMMIT_OK_STATUSES:
|
|
return None
|
|
if first.status_code not in _BRANCH_RETRY_STATUSES:
|
|
return f"commit http {first.status_code}"
|
|
fallback = client.put(url, headers=_headers(cfg), json={**body, "branch": branch})
|
|
if fallback.status_code in _COMMIT_OK_STATUSES:
|
|
return None
|
|
return f"commit http {first.status_code} then fallback http {fallback.status_code}"
|
|
|
|
|
|
def _file_sha(client: httpx.Client, cfg: dict, path: str) -> tuple[str | None, str | None]:
|
|
"""Read the file's current blob sha at the base branch."""
|
|
|
|
url = f"{_base_url(cfg)}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/contents/{path}"
|
|
response = client.get(url, headers=_headers(cfg), params={"ref": _base_branch(cfg)})
|
|
if response.status_code != HTTP_OK:
|
|
return None, f"file sha http {response.status_code}"
|
|
sha = _json_payload(response).get("sha")
|
|
if not sha:
|
|
return None, "file sha missing from contents response"
|
|
return str(sha), None
|
|
|
|
|
|
def _oldest_repair_pull(response: Any, base_branch: str) -> dict[str, Any]:
|
|
"""Pick the lowest-numbered repair proposal out of an open-pulls payload."""
|
|
|
|
try:
|
|
payload = response.json()
|
|
except Exception as exc:
|
|
return _no_proposal(f"open proposal parse failed: {exc}")
|
|
if not isinstance(payload, list):
|
|
return _no_proposal("open proposal payload is not a list")
|
|
matches = [pull for pull in payload if _is_repair_pull(pull, base_branch)]
|
|
if not matches:
|
|
return {**_no_proposal(None), "open_count": 0}
|
|
oldest = min(matches, key=lambda pull: int(pull["number"]))
|
|
return {
|
|
"found": True,
|
|
"open_count": len(matches),
|
|
"pr_number": int(oldest["number"]),
|
|
"url": str(oldest.get("html_url") or "") or None,
|
|
"branch": str(oldest["head"].get("ref") or "") or None,
|
|
"error": None,
|
|
}
|
|
|
|
|
|
def _is_repair_pull(pull: Any, base_branch: str) -> bool:
|
|
"""Report whether one open pull request is a Hermes repair proposal."""
|
|
|
|
if not isinstance(pull, dict):
|
|
return False
|
|
number = pull.get("number")
|
|
if isinstance(number, bool) or not isinstance(number, int):
|
|
return False
|
|
head, base = pull.get("head"), pull.get("base")
|
|
if not isinstance(head, dict) or not isinstance(base, dict):
|
|
return False
|
|
head_ref = str(head.get("ref") or "")
|
|
return head_ref.startswith(_REPAIR_BRANCH_PREFIX) and str(base.get("ref") or "") == base_branch
|
|
|
|
|
|
def _no_proposal(error: str | None) -> dict[str, Any]:
|
|
"""Build the fail-open result meaning "no existing proposal found"."""
|
|
|
|
return {"found": False, "open_count": 0, "pr_number": None, "url": None, "branch": None, "error": error}
|
|
|
|
|
|
def _pr_result(response: Any) -> dict[str, Any]:
|
|
"""Map a pull-request response payload to the result shape."""
|
|
|
|
payload = _json_payload(response)
|
|
number = payload.get("number")
|
|
url = str(payload.get("html_url") or "") or None
|
|
if isinstance(number, bool) or not isinstance(number, int):
|
|
return {"pr_number": None, "url": url, "error": None}
|
|
return {"pr_number": number, "url": url, "error": None}
|
|
|
|
|
|
def run_url(ui_url: str, run_id: str) -> str:
|
|
"""Build the deep link that reopens one Hermes run in its console.
|
|
|
|
Inputs: the Hermes UI base url and a run id. Outputs: the resume link, or
|
|
"" when either is missing. Kept here so the pull request and the demo
|
|
monitor cannot drift into pointing at different pages.
|
|
"""
|
|
|
|
base = str(ui_url or "").rstrip("/")
|
|
run = str(run_id or "").strip()
|
|
return f"{base}{RUN_PATH}{run}" if base and run else ""
|
|
|
|
|
|
def _pr_body(incident_id: str, patch: Any, analysis: str, run_id: str, cfg: dict) -> str:
|
|
"""Render the markdown pull-request body for human review.
|
|
|
|
Names the Hermes run that produced the patch. "Proposed by Hermes" is a
|
|
claim; the run id is the receipt, and without it a reviewer has no way to
|
|
read the prompt the model was given or the tool calls it made. The
|
|
diagnosis issues have carried this from the start - the pull requests,
|
|
which are the more consequential artifact, did not.
|
|
"""
|
|
|
|
lines = [
|
|
f"## Hermes repair proposal for incident {incident_id}",
|
|
"",
|
|
f"**Incident:** {incident_id}",
|
|
f"**File:** `{patch.path}`",
|
|
]
|
|
finding = hermes_sonar_client.issue_url(incident_id, cfg.get("sonar_ui_url"))
|
|
if finding:
|
|
# A sweep proposal exists because of one finding; the reviewer's first
|
|
# question is what it said, and an issue key is not an answer.
|
|
lines.append(f"**SonarQube finding:** {finding}")
|
|
lines += [
|
|
f"**Analysis:** {analysis}",
|
|
f"**Rationale:** {patch.rationale}",
|
|
]
|
|
if run_id:
|
|
ui_url = str(cfg.get("hermes_ui_url") or "").rstrip("/")
|
|
if ui_url:
|
|
lines.append(f"**Hermes run:** [{run_id}]({run_url(ui_url, run_id)})")
|
|
lines.append(
|
|
"That link opens the run itself: the prompt Hermes was given, the evidence "
|
|
"bundle it read, the tools it called, and the JSON it returned."
|
|
)
|
|
else:
|
|
lines.append(f"**Hermes run:** `{run_id}`")
|
|
lines += [
|
|
"",
|
|
"Proposed by Hermes; validated and pushed by Ariadne; "
|
|
"requires human review — no automatic merge.",
|
|
]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _branch_name(ref: Any) -> str:
|
|
"""Derive the repair branch name from a build number or sweep token.
|
|
|
|
Sanitized rather than interpolated: the token now reaches here from a
|
|
SonarQube finding key as well as from a build number, and a ref is one of
|
|
the few places where an unexpected character stops being cosmetic.
|
|
"""
|
|
|
|
token = _BRANCH_UNSAFE.sub("-", str(ref)).strip("-")
|
|
return f"{_REPAIR_BRANCH_PREFIX}{token or 'unknown'}"
|
|
|
|
|
|
def _json_payload(response: Any) -> dict[str, Any]:
|
|
"""Return a response's JSON body as a dict, tolerating garbage."""
|
|
|
|
try:
|
|
payload = response.json()
|
|
except Exception:
|
|
return {}
|
|
return payload if isinstance(payload, dict) else {}
|
|
|
|
|
|
def _headers(cfg: dict) -> dict[str, str]:
|
|
"""Build the Gitea token authorization header."""
|
|
|
|
return {"Authorization": f"token {cfg.get('gitea_token') or ''}"}
|
|
|
|
|
|
def _base_url(cfg: dict) -> str:
|
|
"""Return the configured Gitea base URL without a trailing slash."""
|
|
|
|
return str(cfg.get("gitea_base_url") or "").strip().rstrip("/")
|
|
|
|
|
|
def _base_branch(cfg: dict) -> str:
|
|
"""Return the configured base branch name."""
|
|
|
|
return str(cfg.get("base_branch") or "")
|
|
|
|
|
|
def _owner(cfg: dict) -> str:
|
|
"""Return the configured repository owner."""
|
|
|
|
return str(cfg.get("owner") or "")
|
|
|
|
|
|
def _repo(cfg: dict) -> str:
|
|
"""Return the configured repository name."""
|
|
|
|
return str(cfg.get("repo") or "")
|
|
|
|
|
|
def _timeout(cfg: dict) -> float:
|
|
"""Return the bounded request timeout in seconds."""
|
|
|
|
try:
|
|
value = float(cfg.get("timeout_seconds"))
|
|
except (TypeError, ValueError):
|
|
return _DEFAULT_TIMEOUT_SECONDS
|
|
return value if value > 0 else _DEFAULT_TIMEOUT_SECONDS
|