gitea_admin/pragent
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: opencode review engine + .opencode factory

Browse Source 
Replace the single Python model-call reviewer with an opencode agent
factory. A primary 'pragent' agent reads a brief (title/body/diff/config/
prior reviews), inspects the checked-out repo, runs the repo's own linters
via bash, loads review-methodology + findings-schema skills, and emits a
{summary, findings} JSON with per-finding severity/path/line/problem/fix/
suggestion/reference. Dormant security/tests/perf subagent lenses fan out
only on large/risky diffs (lean by default).

pilot/opencode_review.py: fetches the repo archive at the head sha into a
temp workdir, writes .pragent/brief.md, drops the factory, runs
'opencode run --pure --agent pragent --dir <workdir>' headlessly. Isolates
HOME (shared, warmed), strips ANTHROPIC_* env (leaked host vars caused
ProviderModelNotFoundError), stdin=DEVNULL (opencode blocks on stdin),
maps the bare OLLAMA_MODEL to the provider-prefixed ref. No Gitea I/O —
ai_review.review_pr parses + anchors + posts (reuses all v2 logic/tests).

PRAGENT_ENGINE=opencode (default) selects it; =ollama keeps the legacy
direct-call path. Verified end-to-end: posts a real review with a summary
section, inline [CRITICAL]/[HIGH] comments + apply-able suggestions +
reference links, and the sha dedupe marker. 49 tests pass.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Marcos
2026-08-17 23:01:59 +00:00
parent 8758b22802
commit 6e3a9eb5b0
13 changed files with 1336 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pilot/ai_review.py
+149 -47
View File
@@ -130,19 +130,25 @@ def parse_text_blocks(content: list) -> str:
return "\n".join(out).strip()
def format_review_body(findings: str, model: str, sha: str) -> str:
def format_review_body(findings: str, model: str, sha: str, summary: str = "") -> str:
"""Format the posted review summary body.
`findings` is the bullet text for findings that could NOT be anchored inline
(or, on the legacy/no-inline path, the whole review). Empty -> "No issues
found.". The hidden sha marker is always appended for the dedupe pass.
found.". `summary` (optional, opencode engine) is rendered as a "Summary"
section right under the header. The hidden sha marker is always appended
for the dedupe pass.
"""
header = REVIEW_HEADER.format(model=model, sha=sha[:8] if sha else "unknown")
findings = (findings or "").strip()
if not findings:
findings = "No issues found."
marker = SHA_MARKER.format(sha=sha) if sha else ""
body = f"{header}\n\n{findings}"
parts = [header]
if summary:
parts.append(summary.strip())
parts.append(findings)
body = "\n\n".join(parts)
if marker:
body += f"\n{marker}"
return body
@@ -258,6 +264,43 @@ def _strip_path_prefix(p: str) -> str:
# ---------------------------------------------------------------------------
def _normalize_finding(f: dict) -> dict | None:
"""Validate + normalize one raw finding dict. Returns None if it's unusable
(missing path/line). Normalises severity, keeps `reference` (default "")."""
if not isinstance(f, dict):
return None
path = f.get("path")
line = f.get("line")
if not isinstance(path, str) or not path.strip():
return None
if not isinstance(line, int) or line < 1:
return None
sev = str(f.get("severity", "medium")).strip().lower()
if sev not in SEVERITIES:
sev = "medium"
reference = str(f.get("reference", "") or "").strip()
return {
"severity": sev,
"path": path.strip(),
"line": line,
"problem": str(f.get("problem", "")).strip(),
"fix": str(f.get("fix", "")).strip(),
"suggestion": str(f.get("suggestion", "") or "").strip(),
"reference": reference,
}
def _last_json_block(text: str) -> str | None:
"""Return the substring of the last fenced ```json block in text, or None.
Falls back to _extract_first_json_object when no fence is present."""
s = text or ""
# Find all ```json ... ``` fenced blocks; take the last.
blocks = list(re.finditer(r"```(?:json)?\s*(\{.*?\})\s*```", s, re.DOTALL))
if blocks:
return blocks[-1].group(1)
return _extract_first_json_object(s)
def parse_findings(text: str) -> list[dict]:
"""Parse the model's JSON response into a list of finding dicts.
@@ -266,23 +309,7 @@ def parse_findings(text: str) -> list[dict]:
Drops findings missing path/line or with an unknown severity (normalised).
Never raises — returns [] on any parse failure.
"""
if not text:
return []
s = text.strip()
# Strip a single wrapping code fence if present.
if s.startswith("```"):
s = re.sub(r"^```[a-zA-Z]*\n?", "", s)
s = re.sub(r"\n?```$", "", s).strip()
data = None
try:
data = json.loads(s)
except json.JSONDecodeError:
obj = _extract_first_json_object(s)
if obj is not None:
try:
data = json.loads(obj)
except json.JSONDecodeError:
data = None
data = _parse_json_tolerant(text)
if not isinstance(data, dict):
return []
findings = data.get("findings")
@@ -290,28 +317,74 @@ def parse_findings(text: str) -> list[dict]:
return []
out = []
for f in findings:
if not isinstance(f, dict):
continue
path = f.get("path")
line = f.get("line")
if not isinstance(path, str) or not path.strip():
continue
if not isinstance(line, int) or line < 1:
continue
sev = str(f.get("severity", "medium")).strip().lower()
if sev not in SEVERITIES:
sev = "medium"
out.append({
"severity": sev,
"path": path.strip(),
"line": line,
"problem": str(f.get("problem", "")).strip(),
"fix": str(f.get("fix", "")).strip(),
"suggestion": str(f.get("suggestion", "") or "").strip(),
})
n = _normalize_finding(f)
if n is not None:
out.append(n)
return out
def parse_review_output(text: str) -> tuple[str, list[dict]]:
"""Parse the opengine's stdout into (summary, findings).
Accepts `{"summary": "...", "findings": [...]}` (the opencode pragent agent)
or a bare `{"findings": [...]}`. `summary` defaults to "". Uses the LAST
```json fenced block (the pragent agent emits JSON as the final block), with
a tolerant fallback. Never raises.
"""
blob = _last_json_block(text)
if blob is None:
return "", []
try:
data = json.loads(blob)
except json.JSONDecodeError:
return "", []
if not isinstance(data, dict):
return "", []
summary = str(data.get("summary", "") or "").strip()
findings = data.get("findings")
out = []
if isinstance(findings, list):
for f in findings:
n = _normalize_finding(f)
if n is not None:
out.append(n)
return summary, out
def _parse_json_tolerant(text: str) -> dict | None:
"""Parse a JSON object from text: try the last fenced block, then a direct
parse, then the first balanced object. Returns None on any failure."""
if not text:
return None
blob = _last_json_block(text)
if blob is not None:
try:
d = json.loads(blob)
if isinstance(d, dict):
return d
except json.JSONDecodeError:
pass
s = text.strip()
if s.startswith("```"):
s = re.sub(r"^```[a-zA-Z]*\n?", "", s)
s = re.sub(r"\n?```$", "", s).strip()
try:
d = json.loads(s)
if isinstance(d, dict):
return d
except json.JSONDecodeError:
pass
obj = _extract_first_json_object(text)
if obj is not None:
try:
d = json.loads(obj)
if isinstance(d, dict):
return d
except json.JSONDecodeError:
pass
return None
def _extract_first_json_object(s: str) -> str | None:
"""Return the substring of the first balanced top-level `{ ... }` in s."""
start = s.find("{")
@@ -362,7 +435,8 @@ def inline_comment_body(f: dict) -> str:
"""Render one finding as a positional review-comment body.
Includes a ```suggestion fence only if the model produced non-empty
replacement code. Gitea renders that as an apply-able suggestion.
replacement code. Gitea renders that as an apply-able suggestion. Appends a
`📎 ref:` link when the finding carries a `reference` URL.
"""
sev = f["severity"].upper()
body = f"**[{sev}]** {f['problem']}"
@@ -370,6 +444,9 @@ def inline_comment_body(f: dict) -> str:
body += f"\n\nFix: {f['fix']}"
if f["suggestion"]:
body += f"\n\n```suggestion\n{f['suggestion']}\n```"
ref = f.get("reference", "")
if ref:
body += f"\n\n📎 ref: {ref}"
return body
@@ -379,7 +456,8 @@ def summary_bullets(findings: list[dict]) -> str:
for f in findings:
loc = f"{f['path']}:{f['line']}" if f["line"] else f["path"]
fix = f" — fix: {f['fix']}" if f["fix"] else ""
lines.append(f"- **[{f['severity'].upper()}]** `{loc}` — {f['problem']}{fix}")
ref = f" ({f.get('reference', '')})" if f.get("reference") else ""
lines.append(f"- **[{f['severity'].upper()}]** `{loc}` — {f['problem']}{fix}{ref}")
return "\n".join(lines)
@@ -621,16 +699,40 @@ def review_pr(
config = fetch_repo_config(api, repo, sha, token)
prior = prior_review_bodies(reviews, sha)
user_prompt = build_user_prompt(title, body, diff, config, prior)
raw_findings = call_model(ollama_url, model, SYSTEM_PROMPT, user_prompt, max_tokens)
findings = parse_findings(raw_findings)
engine = os.environ.get("PRAGENT_ENGINE", "opencode").strip().lower()
review_summary = ""
if engine == "opencode":
# The review "brain" runs on opencode: it gets the checked-out repo,
# the brief, and the pragent agent factory; returns stdout with a
# summary + findings JSON. We parse + anchor + post here.
import opencode_review # local import keeps the ollama path dep-free
# opencode wants a provider-prefixed model ref (headroom/glm-5.2:cloud);
# `model` here is the bare id (OLLAMA_MODEL). OPENCODE_MODEL overrides
# with the full ref; otherwise we prefix the configured provider.
oc_model = os.environ.get("OPENCODE_MODEL") or f"headroom/{model}"
stdout = opencode_review.run(
api=api, repo=repo, index=index, sha=sha, token=token,
title=title, body=body, diff=diff, config=config,
prior_reviews=prior, model=oc_model,
)
review_summary, findings = parse_review_output(stdout)
if not findings and not review_summary:
# opencode produced nothing parseable — fall back to a note.
post_review(api, repo, index, token, format_review_body(
"AI review produced no parseable output.", model, sha))
return True
else:
user_prompt = build_user_prompt(title, body, diff, config, prior)
raw_findings = call_model(ollama_url, model, SYSTEM_PROMPT, user_prompt, max_tokens)
findings = parse_findings(raw_findings)
anchors = parse_diff_anchors(diff)
anchored, unanchored = split_findings(findings, anchors)
# Summary body: the unanchored bullets (or "No issues found."), plus a
# one-line note when inline comments were posted so the summary isn't
# empty-looking.
# empty-looking. The opencode engine also carries a prose summary.
bullets = summary_bullets(unanchored)
summary_parts = []
if anchored:
@@ -639,12 +741,12 @@ def review_pr(
summary_parts.append(bullets)
if not summary_parts:
summary_parts.append("No issues found.")
summary_body = format_review_body("\n\n".join(summary_parts), model, sha)
summary_body = format_review_body("\n\n".join(summary_parts), model, sha, summary=review_summary)
post_inline_review(api, repo, index, token, summary_body, anchored)
print(
f"pragent: reviewed {repo}#{index} sha={sha[:8]} "
f"findings={len(findings)} inline={len(anchored)}",
f"engine={engine} findings={len(findings)} inline={len(anchored)}",
flush=True,
)
return True