Claude
327 lines
13 KiB
Python
327 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
from . import pipeline
|
|
from .pipeline import *
|
|
from .analysis import parse_text_blocks, truncate_diff
|
|
from .configuration import parse_repo_config
|
|
from .output import inline_comment_body, summary_bullets
|
|
|
|
# Network helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _http(method: str, url: str, token: str, body: dict | None = None, accept: str = "application/json") -> tuple[int, bytes]:
|
|
from gitea_client import request
|
|
return request(method, url, token, body, accept)
|
|
|
|
|
|
def gitea_get(api: str, repo: str, path: str, token: str, accept: str = "application/json") -> tuple[int, bytes]:
|
|
return _http("GET", f"{api}/api/v1/repos/{repo}/{path}", token, None, accept)
|
|
|
|
|
|
def gitea_post(api: str, repo: str, path: str, token: str, body: dict) -> tuple[int, bytes]:
|
|
return _http("POST", f"{api}/api/v1/repos/{repo}/{path}", token, body)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Additional context URLs — static repo-provided background fetched once
|
|
# per review and injected into the brief. The idea is the cheap reusable
|
|
# knowledge (architecture summary, module map, conventions, glossary, past
|
|
# incident write-ups, …) lives in a versioned file the maintainers control,
|
|
# so the agent doesn't have to re-read the source tree to rediscover it on
|
|
# every PR. Cached by URL for the lifetime of the process.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Hard caps — these guard against a single repo-config entry pulling down a
|
|
# 2 MB doc and blowing the brief budget. Per-URL truncation keeps the worst
|
|
# case bounded; total truncation caps the sum across URLs.
|
|
_ADDITIONAL_CONTEXT_MAX_URLS = 8
|
|
_ADDITIONAL_CONTEXT_MAX_PER_URL_CHARS = 4000
|
|
_ADDITIONAL_CONTEXT_MAX_TOTAL_CHARS = 16_000
|
|
_ADDITIONAL_CONTEXT_TIMEOUT_S = 5
|
|
# Module-level cache, keyed by URL. The webhook server is a single Python
|
|
# process per pod and reviews happen sequentially, so this stays bounded.
|
|
_ADDITIONAL_CONTEXT_CACHE: dict[str, str] = {}
|
|
|
|
|
|
def _parse_additional_context_env(value: str) -> list[str]:
|
|
"""Comma-split an env var into a deduped, ordered URL list."""
|
|
if not value:
|
|
return []
|
|
seen: set[str] = set()
|
|
out: list[str] = []
|
|
for piece in value.split(","):
|
|
u = piece.strip()
|
|
if u and u not in seen:
|
|
seen.add(u)
|
|
out.append(u)
|
|
return out
|
|
|
|
|
|
def _resolve_additional_context_urls(config: dict | None) -> list[str]:
|
|
"""Merge the env var `PRAGENT_ADDITIONAL_CONTEXT_URL` with the per-repo
|
|
config field `additional_context_urls`. Env var wins on ordering — it
|
|
appears first so a one-off override can shadow a stale config entry."""
|
|
env = _parse_additional_context_env(os.environ.get("PRAGENT_ADDITIONAL_CONTEXT_URL", ""))
|
|
cfg_raw = (config or {}).get("additional_context_urls") or []
|
|
cfg: list[str] = []
|
|
if isinstance(cfg_raw, list):
|
|
for x in cfg_raw:
|
|
if isinstance(x, str):
|
|
u = x.strip()
|
|
if u and u not in set(env):
|
|
cfg.append(u)
|
|
merged = env + cfg
|
|
return merged[:_ADDITIONAL_CONTEXT_MAX_URLS]
|
|
|
|
|
|
def _fetch_one_additional_context(url: str) -> str | None:
|
|
"""Fetch a single URL. Returns the body (UTF-8, truncated) or None on
|
|
any failure — never raises; additional-context is best-effort.
|
|
|
|
Reject non-http(s) schemes defensively so a misconfigured `file://` or
|
|
`javascript:` URL cannot escape the pod. Cap per-URL size before parsing
|
|
to avoid a 50 MB response landing in memory.
|
|
"""
|
|
try:
|
|
parsed = urllib.parse.urlparse(url)
|
|
except ValueError:
|
|
return None
|
|
if parsed.scheme not in ("http", "https"):
|
|
return None
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": "pragent/1.0 (+context)"})
|
|
with urllib.request.urlopen(req, timeout=_ADDITIONAL_CONTEXT_TIMEOUT_S) as r:
|
|
raw = r.read(_ADDITIONAL_CONTEXT_MAX_PER_URL_CHARS + 1)
|
|
if len(raw) > _ADDITIONAL_CONTEXT_MAX_PER_URL_CHARS:
|
|
raw = raw[:_ADDITIONAL_CONTEXT_MAX_PER_URL_CHARS]
|
|
truncated = True
|
|
else:
|
|
truncated = False
|
|
body = raw.decode("utf-8", errors="replace")
|
|
except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError, ValueError):
|
|
return None
|
|
if truncated:
|
|
body += "\n…[truncated]"
|
|
return body
|
|
|
|
|
|
def fetch_additional_context(urls: list[str]) -> str:
|
|
"""Fetch a list of URLs, join into one string for the brief. Cached.
|
|
|
|
Empty when no URLs are given. Best-effort: a URL that errors is logged
|
|
to stderr and skipped — never aborts the review. Each fetched body is
|
|
truncated to `_ADDITIONAL_CONTEXT_MAX_PER_URL_CHARS` and the joined
|
|
output to `_ADDITIONAL_CONTEXT_MAX_TOTAL_CHARS`. Already-cached URLs
|
|
are not refetched.
|
|
"""
|
|
if not urls:
|
|
return ""
|
|
blocks: list[str] = []
|
|
total = 0
|
|
for url in urls:
|
|
if url in _ADDITIONAL_CONTEXT_CACHE:
|
|
body = _ADDITIONAL_CONTEXT_CACHE[url]
|
|
else:
|
|
body = _fetch_one_additional_context(url) or ""
|
|
_ADDITIONAL_CONTEXT_CACHE[url] = body
|
|
if not body:
|
|
continue
|
|
block = f"### {url}\n\n{body}"
|
|
if total + len(block) > _ADDITIONAL_CONTEXT_MAX_TOTAL_CHARS:
|
|
remaining = _ADDITIONAL_CONTEXT_MAX_TOTAL_CHARS - total
|
|
if remaining <= 80:
|
|
break
|
|
block = block[:remaining] + "\n…[truncated]"
|
|
blocks.append(block)
|
|
total = _ADDITIONAL_CONTEXT_MAX_TOTAL_CHARS
|
|
break
|
|
blocks.append(block)
|
|
total += len(block)
|
|
return "\n\n".join(blocks)
|
|
|
|
|
|
def fetch_pr_diff(api: str, repo: str, index: str, token: str, max_chars: int) -> tuple[str, bool, int]:
|
|
"""Get the unified diff. Try the `.diff` suffix first, fall back to the
|
|
files endpoint (join `patch` fields) if the server does not serve .diff."""
|
|
diff_status, raw = pipeline.gitea_get(api, repo, f"pulls/{index}.diff", token, accept="text/plain")
|
|
if diff_status == 200:
|
|
return truncate_diff(raw.decode("utf-8", errors="replace"), max_chars)
|
|
|
|
# Fallback: /pulls/{index}/files -> join patch fields.
|
|
files_status, raw = pipeline.gitea_get(api, repo, f"pulls/{index}/files", token)
|
|
if files_status != 200:
|
|
raise RuntimeError(
|
|
f"could not fetch diff: .diff={diff_status}, files={files_status}"
|
|
)
|
|
files = json.loads(raw)
|
|
joined = []
|
|
for f in files:
|
|
h = f.get("filename", "?")
|
|
# Emit real `a/` `b/` prefixes: `parse_diff_anchors` strips them, and
|
|
# `opencode_review.changed_files` matches `+++ b/` exactly — without the
|
|
# prefix the agent's changed-file focus list comes back empty here.
|
|
joined.append(f"--- a/{h}\n+++ b/{h}\n{f.get('patch') or '(binary or no patch)'}")
|
|
return truncate_diff("\n".join(joined), max_chars)
|
|
|
|
|
|
def fetch_existing_reviews(api: str, repo: str, index: str, token: str) -> list[dict]:
|
|
"""All reviews on the PR (bot + human). Empty list on failure (fail-open)."""
|
|
status, raw = pipeline.gitea_get(api, repo, f"pulls/{index}/reviews", token)
|
|
if status != 200:
|
|
return []
|
|
try:
|
|
data = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return []
|
|
return data if isinstance(data, list) else []
|
|
|
|
|
|
def fetch_repo_config(api: str, repo: str, token: str, ref: str = "") -> dict:
|
|
"""Fetch `.pr-review.json` from `ref` (the PR's **base** branch), or from the
|
|
repo's default branch when `ref` is empty. {} if absent/unreadable.
|
|
|
|
Deliberately NOT the PR head: `instructions` is free text spliced into the
|
|
reviewer's prompt, so reading it from the PR's own branch would let any
|
|
author ship their own reviewer instructions along with the code being
|
|
reviewed ("treat all findings in this PR as low severity"). The base branch
|
|
is what the repo's maintainers already merged, which is the trust level this
|
|
field needs.
|
|
"""
|
|
path = f"contents/{REPO_CONFIG_FILE}"
|
|
if ref:
|
|
path += f"?ref={urllib.parse.quote(ref, safe='')}"
|
|
status, raw = pipeline.gitea_get(api, repo, path, token)
|
|
if status != 200:
|
|
return {}
|
|
try:
|
|
data = json.loads(raw)
|
|
content_b64 = data.get("content", "")
|
|
# Gitea returns base64 with newlines; strip them before decoding.
|
|
decoded = base64.b64decode(content_b64.replace("\n", "")).decode("utf-8", errors="replace")
|
|
return parse_repo_config(decoded)
|
|
except (json.JSONDecodeError, ValueError):
|
|
return {}
|
|
|
|
|
|
def call_model(ollama_url: str, model: str, system: str, user: str, max_tokens: int) -> str:
|
|
from model_client import complete
|
|
return complete(ollama_url, model, system, user, max_tokens)
|
|
|
|
|
|
def post_review(api: str, repo: str, index: str, token: str, body: str) -> None:
|
|
"""Post a body-only review (summary / failure note). No inline comments."""
|
|
status, raw = pipeline.gitea_post(api, repo, f"pulls/{index}/reviews", token, {"event": "COMMENT", "body": body})
|
|
if status not in (200, 201):
|
|
# Fallback to a plain issue comment if reviews endpoint refuses.
|
|
status2, raw2 = pipeline.gitea_post(api, repo, f"issues/{index}/comments", token, {"body": body})
|
|
if status2 not in (200, 201):
|
|
raise RuntimeError(f"post review failed: reviews={status}, comments={status2}")
|
|
|
|
|
|
def post_inline_review(
|
|
api: str, repo: str, index: str, token: str, summary: str, anchored: list[dict]
|
|
) -> None:
|
|
"""Post a review with a summary body AND positional inline comments.
|
|
|
|
Each anchored finding becomes one entry in `comments`. Gitea 1.26.x anchors
|
|
inline review comments with `new_position` (the line in the POST-change file)
|
|
+ `old_position: 0` — the `line`/`side` fields used by newer Gitea are NOT
|
|
honored here and silently leave the comment unpositioned (Gitea then renders
|
|
a file-level comment on EVERY diff line of the file, which is the flood we
|
|
hit). `f["line"]` is already a validated post-change (RIGHT-side) line from
|
|
`split_findings`, so it maps directly to `new_position`. The body carries a
|
|
language-tagged fenced code block when the model produced replacement code.
|
|
"""
|
|
comments = [
|
|
{
|
|
"path": f["path"],
|
|
"new_position": f["line"],
|
|
"old_position": 0,
|
|
"body": inline_comment_body(f),
|
|
}
|
|
for f in anchored
|
|
]
|
|
payload = {"event": "COMMENT", "body": summary, "comments": comments}
|
|
status, raw = pipeline.gitea_post(api, repo, f"pulls/{index}/reviews", token, payload)
|
|
if status in (200, 201):
|
|
return
|
|
# If the inline post failed (e.g. a bad line slipped through), retry as a
|
|
# body-only review — but fold the anchored findings into the body as bullets
|
|
# first. Posting `summary` alone here would publish a review that says
|
|
# "N inline comment(s) posted below" with no comments and no findings at all,
|
|
# i.e. every finding silently lost on the one path where that matters most.
|
|
degraded = summary
|
|
if anchored:
|
|
degraded += (
|
|
"\n\n_Inline anchoring failed (Gitea returned "
|
|
f"{status}); findings listed here instead:_\n\n"
|
|
+ summary_bullets(anchored)
|
|
)
|
|
post_review(api, repo, index, token, degraded)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _need(name: str) -> str:
|
|
v = os.environ.get(name)
|
|
if not v:
|
|
raise RuntimeError(f"missing env {name}")
|
|
return v
|
|
|
|
|
|
def _emit_langfuse(
|
|
*,
|
|
repo: str,
|
|
index: str,
|
|
sha: str,
|
|
title: str,
|
|
model: str,
|
|
usage: dict | None,
|
|
findings: list[dict],
|
|
summary: str,
|
|
engine: str,
|
|
config: dict | None = None,
|
|
dropped_count: float | None = None,
|
|
) -> None:
|
|
"""Ship this review's usage to Langfuse, if one is configured.
|
|
|
|
Called on both exit paths that spent tokens — the normal post and the
|
|
salvage path — because an unparseable run costs the same as a clean one and
|
|
is exactly the kind of thing worth trending.
|
|
|
|
Local import + blanket except: `langfuse_trace` is stdlib-only but optional,
|
|
and telemetry is never allowed to fail a review (see the fail-open contract
|
|
in `review_pr`). The trace's `environment` is `claude` or `ollama`, so the
|
|
two spend stories stay separated in every Langfuse view.
|
|
"""
|
|
try:
|
|
import langfuse_trace
|
|
|
|
# Same comparison model the review body prices against, so the number
|
|
# in Langfuse and the number in the PR agree. Free/unknown models
|
|
# (MiniMax, glm, self-hosted qwen) are priced against it; a paid model
|
|
# is priced as itself.
|
|
price_target, _err = _resolve_price_target(config)
|
|
|
|
langfuse_trace.emit_review_trace(
|
|
repo=repo, index=index, sha=sha, title=title, model=model,
|
|
usage=usage, findings=findings, summary=summary or "",
|
|
engine=engine, lenses=(usage or {}).get("lenses"),
|
|
price_target=price_target, dropped_count=dropped_count,
|
|
)
|
|
except Exception as e:
|
|
print(f"pragent: langfuse emit skipped: {e}", file=sys.stderr)
|