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)