#!/usr/bin/env python3 """pragent pilot — opencode review engine (the "brain" host). When `PRAGENT_ENGINE=opencode` (the default), `ai_review.review_pr` delegates the analysis to this module instead of making one direct model call. It: 1. fetches the target repo's archive at the PR head sha into a temp workdir (so the reviewer has the real files, not just the diff text); 2. writes a `.pragent/brief.md` (title, description, diff, repo config, prior reviews, sha, anchor hint) for the `pragent` agent to read; 3. drops pragent's `opencode.json` + `.opencode/` factory into the workdir; 4. runs `opencode run --pure --agent pragent --dir --model ` headlessly and returns the agent's stdout (the summary + findings JSON). The caller (`ai_review.review_pr`) parses that stdout into `(summary, findings)`, validates the findings against diff anchors, and posts the review to Gitea — so this module does NO Gitea I/O and NO parsing. It is pure review-engine glue. Stdlib only. Fail-open: `run()` raises on failure; `review_pr` catches and posts a short failure note. Env: PRAGENT_FACTORY_DIR repo root holding opencode.json + .opencode/ (default: this file's parent's parent — the pragent repo root). PRAGENT_OPENCODE_BIN path to the opencode CLI (default: shutil.which / the known linuxbrew path). PRAGENT_RTK_DIR dir holding the `rtk` binary, prepended to PATH for the agent's bash tool (default: /home/marcos/.headroom/bin). PRAGENT_WORK_ROOT parent for temp workdirs (default: /tmp/pragent-work). PRAGENT_KEEP_WORK if set, leave the workdir on disk for debugging. PRAGENT_REVIEW_TIMEOUT seconds to allow opencode to run (default: 480). """ import io import json import os import shutil import subprocess import tarfile import tempfile import urllib.error import urllib.request # Where the factory lives (opencode.json + .opencode/). Default: the pragent # repo root (this file is at /pilot/opencode_review.py). _DEFAULT_FACTORY = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) RTK_DIR = os.environ.get("PRAGENT_RTK_DIR", "/home/marcos/.headroom/bin") WORK_ROOT = os.environ.get("PRAGENT_WORK_ROOT", "/tmp/pragent-work") TIMEOUT = int(os.environ.get("PRAGENT_REVIEW_TIMEOUT", "480")) def _factory_dir() -> str: return os.environ.get("PRAGENT_FACTORY_DIR", _DEFAULT_FACTORY) def _opencode_bin() -> str: b = os.environ.get("PRAGENT_OPENCODE_BIN") if b: return b found = shutil.which("opencode") if found: return found return "/home/linuxbrew/.linuxbrew/bin/opencode" # --------------------------------------------------------------------------- # Archive fetch + untar # --------------------------------------------------------------------------- def fetch_archive(api: str, repo: str, sha: str, token: str, dest: str) -> None: """Download `GET {api}/api/v1/repos/{repo}/archive/{sha}.tar.gz` and extract into `dest`, stripping the archive's single top-level directory so the repo files sit directly at `dest/` (matching the diff's `+++ b/foo` paths). """ url = f"{api.rstrip('/')}/api/v1/repos/{repo}/archive/{sha}.tar.gz" req = urllib.request.Request(url, headers={"Authorization": f"token {token}"}) with urllib.request.urlopen(req, timeout=120) as r: blob = r.read() _extract_tar_strip_one(blob, dest) def _extract_tar_strip_one(blob: bytes, dest: str) -> None: """Extract a tar.gz blob into dest, stripping one common top-level dir. If every member shares a single top-level prefix, that prefix is removed (so `repo-sha/foo` -> `dest/foo`). If members have no common prefix, extract as-is. Handles dirs, files, symlinks; ignores absolute paths / `..` for safety. """ os.makedirs(dest, exist_ok=True) with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as tar: members = tar.getmembers() # Find the common top-level prefix (the part before the first '/'). top_levels = set() for m in members: name = m.name.lstrip("/") if not name: continue top_levels.add(name.split("/", 1)[0]) prefix = "" if len(top_levels) == 1: (prefix,) = top_levels prefix += "/" # strip "topdir/" for m in members: name = m.name.lstrip("/") if not name: continue # Safety: no absolute, no parent traversal. if ".." in name.split("/"): continue rel = name[len(prefix):] if prefix else name if not rel or rel == "/": continue target = os.path.join(dest, rel) if m.isdir(): os.makedirs(target, exist_ok=True) continue if m.issym(): parent = os.path.dirname(target) os.makedirs(parent, exist_ok=True) try: if os.path.lexists(target): os.remove(target) os.symlink(m.linkname, target) except OSError: pass continue if m.isreg(): parent = os.path.dirname(target) os.makedirs(parent, exist_ok=True) f = tar.extractfile(m) if f is None: continue with open(target, "wb") as out: shutil.copyfileobj(f, out) # --------------------------------------------------------------------------- # Brief + factory drop # --------------------------------------------------------------------------- BRIEF_PATH = ".pragent/brief.md" _BRIEF_TEMPLATE = """\ # pragent review brief - **repo:** {repo} - **pr:** #{index} - **head_sha:** `{sha}` ## Title {title} ## Description {description} ## Repo review config (.pr-review.json) {config} ## Prior reviews (already posted — do NOT repeat these points) {prior} ## How to anchor inline comments Each finding `line` MUST be a line that exists in the POST-CHANGE version of `path` — a context line (leading space in the diff) or an added `+` line. Never a removed `-` line. Use the closest context line you can see if unsure. ## Diff ```diff {diff} ``` """ def write_brief( workdir: str, *, repo: str, index: str, sha: str, title: str, description: str, diff: str, config: dict | None, prior_reviews: list[str] | None, ) -> str: """Render `.pragent/brief.md` in the workdir. Returns the path written.""" path = os.path.join(workdir, ".pragent") os.makedirs(path, exist_ok=True) brief = os.path.join(path, "brief.md") cfg = "_(none)_" if config: cfg = json.dumps(config, indent=2, ensure_ascii=False) prior = "_(none)_" if prior_reviews: prior = "\n\n---\n\n".join(prior_reviews) if len(prior) > 8000: prior = prior[:8000] + "\n…[prior reviews truncated]" content = _BRIEF_TEMPLATE.format( repo=repo or "?", index=index or "?", sha=sha or "?", title=title or "(none)", description=description.strip() or "_(none)_", config=cfg, prior=prior, diff=diff or "_(empty)_", ) with open(brief, "w", encoding="utf-8") as f: f.write(content) return brief def drop_factory(workdir: str) -> None: """Copy the pragent `opencode.json` + `.opencode/` into the workdir so `opencode run --dir ` discovers them as project config. Overwrites any existing ones (the workdir is a throwaway archive checkout).""" src = _factory_dir() oc_json = os.path.join(src, "opencode.json") if os.path.isfile(oc_json): shutil.copy2(oc_json, os.path.join(workdir, "opencode.json")) src_oc = os.path.join(src, ".opencode") dst_oc = os.path.join(workdir, ".opencode") if os.path.isdir(dst_oc): shutil.rmtree(dst_oc) if os.path.isdir(src_oc): shutil.copytree(src_oc, dst_oc) # --------------------------------------------------------------------------- # opencode invocation # --------------------------------------------------------------------------- _PROMPT = ( "Read .pragent/brief.md and review this pull request as pragent. " "Load the review-methodology and findings-schema skills, inspect the " "changed files and surrounding code in this repo, run any available " "linters/typecheck on the changed files via bash, and delegate to the " "security/tests/perf subagents only if the diff is large or " "security-sensitive. End your message with a short prose summary followed " "by the findings JSON code block per the findings-schema skill." ) def _shared_home() -> str: """A persistent shared HOME for opencode across reviews. opencode bootstraps its runtime (bun-installs `@opencode-ai` into `$HOME/.config/opencode/node_modules` + fetches a models cache) on its FIRST run in a fresh HOME — and that first run exits WITHOUT producing the answer. A shared, warmed HOME makes every review a warm run (fast + reliable) and is safe for this single-reviewer bot (one review at a time). The provider/model/permission config (`opencode.json`) is installed here as the isolated home's GLOBAL config; the `.opencode/` agents/skills/commands are dropped per-workdir as PROJECT config. Clean split: infra shared, the review factory per-PR. """ home = os.path.join(WORK_ROOT, ".opencode-home") os.makedirs(home, exist_ok=True) return home def _ensure_global_config(home: str) -> None: """Install the pragent opencode.json as the isolated home's global config so the provider/model/permission are always present (warm-up + every review), regardless of --dir. Idempotent.""" dst_dir = os.path.join(home, ".config", "opencode") os.makedirs(dst_dir, exist_ok=True) dst = os.path.join(dst_dir, "opencode.json") src = os.path.join(_factory_dir(), "opencode.json") if not os.path.isfile(src): return # Copy if missing or changed (compare mtime/size to avoid pointless writes). if not os.path.isfile(dst) or os.path.getmtime(src) > os.path.getmtime(dst): shutil.copy2(src, dst) def _build_env(home: str) -> dict: """Build the subprocess env for an opencode run. - HOME -> the isolated shared home (so the host user's ~/.config/opencode is not merged; the pragent opencode.json is installed there as the global config by _ensure_global_config). - Drop XDG_*_HOME (force config resolution under the isolated HOME). - Drop ANTHROPIC_* (host vars like ANTHROPIC_BASE_URL / ANTHROPIC_AUTH_TOKEN / ANTHROPIC_DEFAULT_*_MODEL leak from the user's shell and confuse opencode's @ai-sdk/anthropic provider — ANTHROPIC_DEFAULT_SONNET_MODEL=glm-5.2:cloud makes opencode look for provider "glm-5.2:cloud" → ProviderModelNotFoundError. The headroom provider's config options.baseURL/apiKey are self-contained). - Drop stray OPENCODE_* except the LSP flag (set explicitly below). - Prepend the rtk dir to PATH so the agent's bash tool can call `rtk`. """ env = dict(os.environ) env["HOME"] = home for k in ("XDG_CONFIG_HOME", "XDG_DATA_HOME", "XDG_STATE_HOME"): env.pop(k, None) for k in list(env): if k.startswith("ANTHROPIC_") or ( k.startswith("OPENCODE_") and k != "OPENCODE_EXPERIMENTAL_LSP_TOOL" ): env.pop(k, None) path = env.get("PATH", "") env["PATH"] = (RTK_DIR + os.pathsep + path) if RTK_DIR else path env.setdefault("OPENCODE_EXPERIMENTAL_LSP_TOOL", "true") return env def _warm_opencode(home: str, model: str) -> None: """One-time warm-up: trigger opencode's runtime install so the real run is a warm run. Runs with the global config present (provider resolvable) so it doesn't poison the models cache with a negative entry. Idempotent via a marker file. stdin=DEVNULL + a trivial prompt make this a fast no-op once the runtime is installed.""" marker = os.path.join(home, ".pragent.warmed") if os.path.exists(marker): return _ensure_global_config(home) env = _build_env(home) try: subprocess.run( [_opencode_bin(), "run", "--pure", "--model", model, "ok"], cwd=home, env=env, capture_output=True, text=True, stdin=subprocess.DEVNULL, timeout=240, ) except (subprocess.TimeoutExpired, Exception): pass # warm-up output is discarded; the install is what matters try: open(marker, "w").close() except OSError: pass def run_opencode(workdir: str, model: str, timeout: int | None = None) -> str: """Run the pragent agent headlessly in `workdir`. Returns the agent's stdout. Isolates from the host user's global opencode config by pointing HOME at a shared temp dir (so ~/.config/opencode is not merged) and passing --pure (no external plugins). The workdir's opencode.json + .opencode/ (dropped by drop_factory) are the only project config discovered; the shared home's global opencode.json supplies the provider/model/permission. PATH prepends the rtk dir so the agent's bash tool can call `rtk`. Warms the HOME first (cold runs produce no output) and retries once on empty stdout. stdin=DEVNULL is critical: opencode blocks on stdin (permission prompt / interactive input) when run headlessly via subprocess, hanging until timeout. """ bin_ = _opencode_bin() home = _shared_home() _warm_opencode(home, model) env = _build_env(home) cmd = [ bin_, "run", "--pure", "--agent", "pragent", "--dir", workdir, "--model", model, _PROMPT, ] last_err = "" for attempt in range(2): try: proc = subprocess.run( cmd, cwd=workdir, env=env, capture_output=True, text=True, stdin=subprocess.DEVNULL, timeout=timeout or TIMEOUT, ) except subprocess.TimeoutExpired as e: last_err = f"opencode timed out after {e.timeout}s" continue out = (proc.stdout or "").strip() if out: return proc.stdout last_err = f"opencode empty stdout (rc={proc.returncode}); stderr: {(proc.stderr or '')[-1500:]}" raise RuntimeError(last_err or "opencode produced no output") # --------------------------------------------------------------------------- # Orchestrator entry point # --------------------------------------------------------------------------- def run( *, api: str, repo: str, index: str, sha: str, token: str, title: str, body: str, diff: str, config: dict | None, prior_reviews: list[str] | None, model: str, ) -> str: """End-to-end: checkout archive → brief → drop factory → opencode → stdout. Returns the raw opencode stdout (summary + findings JSON). Raises on any failure; the caller (review_pr) fails open. The workdir is removed unless PRAGENT_KEEP_WORK is set. """ os.makedirs(WORK_ROOT, exist_ok=True) workdir = tempfile.mkdtemp(prefix=f"{repo.replace('/', '_')}-{sha[:8]}-", dir=WORK_ROOT) keep = bool(os.environ.get("PRAGENT_KEEP_WORK")) try: fetch_archive(api, repo, sha, token, workdir) write_brief( workdir, repo=repo, index=index, sha=sha, title=title, description=body, diff=diff, config=config, prior_reviews=prior_reviews, ) drop_factory(workdir) stdout = run_opencode(workdir, model) if not stdout.strip(): raise RuntimeError("opencode produced no output") return stdout finally: if not keep: shutil.rmtree(workdir, ignore_errors=True)